Skills › Scientific Agent Skills › scholar-evaluation › _common.py
_common.py
1002 lines · skills/scholar-evaluation/scripts/_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Bounded, dependency-free helpers for local scholar-evaluation records."""
4from __future__ import annotations
6import json
7import math
8import re
9from dataclasses import dataclass
10from datetime import date
11from pathlib import Path
12from typing import Any, Iterable
14SCHEMA_VERSION = "2.0"
15NOTICE = (
16 "DESCRIPTIVE DEVELOPMENTAL ASSESSMENT ONLY - NOT A DECISION "
17 "RECOMMENDATION - QUALIFIED HUMAN REVIEW REQUIRED"
18)
19PROHIBITED_USES = {
20 "admissions",
21 "awards",
22 "discipline",
23 "funding",
24 "hiring",
25 "promotion",
26 "tenure",
27 "other_high_impact_personnel_decision",
28}
29ALLOWED_PURPOSE = "developmental_review_of_scholarly_work"
30ALLOWED_UNIT = "scholarly_work"
31ALLOWED_CLASSIFICATIONS = (
32 "synthetic",
33 "public_scholarly_work",
34 "deidentified_low_stakes",
35)
36RATING_STATUSES = ("rated", "missing", "not_applicable")
38MAX_INPUT_BYTES = 2 * 1024 * 1024
39MAX_OUTPUT_BYTES = 2 * 1024 * 1024
40MAX_TEXT_CHARS = 4_000
41MAX_LIST_ITEMS = 1_000
42MAX_OBJECT_FIELDS = 100
43MAX_DEPTH = 20
44MAX_TOTAL_NODES = 25_000
45MAX_CRITERIA = 50
46MAX_EVALUATIONS = 50
47MAX_CSV_ROWS = 20_000
49IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9._:-]{1,95}$")
50PRIVATE_FIELD_KEYS = {
51 "applicant_name",
52 "application_text",
53 "candidate_name",
54 "cv_text",
55 "date_of_birth",
56 "dob",
57 "document_text",
58 "email",
59 "full_name",
60 "home_address",
61 "person_name",
62 "phone",
63 "private_application",
64 "raw_application",
65 "resume_text",
66 "social_security_number",
67 "ssn",
68}
69PROXY_TERMS = (
70 "journal impact factor",
71 "impact factor",
72 "h-index",
73 "h index",
74 "citation count",
75 "altmetric",
76 "conference ranking",
77 "journal ranking",
78 "institution prestige",
79 "journal prestige",
80 "venue prestige",
81 "university ranking",
82)
85class ValidationError(ValueError):
86 """A deterministic input failure that never includes a supplied value."""
88 def __init__(self, code: str, path: str = "$") -> None:
89 super().__init__(code)
90 self.code = code
91 self.path = path
94@dataclass(frozen=True)
95class Issue:
96 """A minimized validation issue."""
98 code: str
99 path: str
100 level: str = "error"
102 def as_dict(self) -> dict[str, str]:
103 return {"code": self.code, "path": self.path, "level": self.level}
106def _pairs_no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
107 result: dict[str, Any] = {}
108 for key, value in pairs:
109 if key in result:
110 raise ValidationError("JSON_DUPLICATE_KEY")
111 result[key] = value
112 return result
115def _check_local_file(path: Path, suffix: str) -> None:
116 if path.suffix.lower() != suffix:
117 raise ValidationError("INPUT_SUFFIX_NOT_ALLOWED")
118 if not path.exists():
119 raise ValidationError("INPUT_NOT_FOUND")
120 if path.is_symlink():
121 raise ValidationError("INPUT_SYMLINK_NOT_ALLOWED")
122 if not path.is_file():
123 raise ValidationError("INPUT_NOT_REGULAR_FILE")
124 if path.stat().st_size > MAX_INPUT_BYTES:
125 raise ValidationError("INPUT_TOO_LARGE")
128def _scan_structure(value: Any, path: str = "$") -> None:
129 nodes = 0
130 stack: list[tuple[Any, str, int]] = [(value, path, 0)]
131 while stack:
132 current, current_path, depth = stack.pop()
133 nodes += 1
134 if nodes > MAX_TOTAL_NODES:
135 raise ValidationError("STRUCTURE_TOO_LARGE", current_path)
136 if depth > MAX_DEPTH:
137 raise ValidationError("STRUCTURE_TOO_DEEP", current_path)
138 if isinstance(current, dict):
139 if len(current) > MAX_OBJECT_FIELDS:
140 raise ValidationError("OBJECT_TOO_LARGE", current_path)
141 for key, child in current.items():
142 if not isinstance(key, str):
143 raise ValidationError("OBJECT_KEY_NOT_TEXT", current_path)
144 normalized = key.strip().lower()
145 if normalized in PRIVATE_FIELD_KEYS:
146 raise ValidationError("PRIVATE_FIELD_NOT_ALLOWED", f"{current_path}.{key}")
147 stack.append((child, f"{current_path}.{key}", depth + 1))
148 elif isinstance(current, list):
149 if len(current) > MAX_LIST_ITEMS:
150 raise ValidationError("LIST_TOO_LARGE", current_path)
151 for index, child in enumerate(current):
152 stack.append((child, f"{current_path}[{index}]", depth + 1))
153 elif isinstance(current, str) and len(current) > MAX_TEXT_CHARS:
154 raise ValidationError("TEXT_TOO_LONG", current_path)
155 elif current is not None and not isinstance(
156 current, (str, int, float, bool)
157 ):
158 raise ValidationError("VALUE_TYPE_NOT_ALLOWED", current_path)
159 if isinstance(current, float) and not math.isfinite(current):
160 raise ValidationError("NUMBER_NOT_FINITE", current_path)
163def read_json(path: Path | str) -> Any:
164 """Read bounded JSON with duplicate-key and private-field rejection."""
166 local_path = Path(path)
167 _check_local_file(local_path, ".json")
168 try:
169 data = json.loads(
170 local_path.read_text(encoding="utf-8"),
171 object_pairs_hook=_pairs_no_duplicates,
172 )
173 except UnicodeDecodeError as error:
174 raise ValidationError("INPUT_NOT_UTF8") from error
175 except (json.JSONDecodeError, RecursionError) as error:
176 raise ValidationError("JSON_INVALID") from error
177 _scan_structure(data)
178 return data
181def read_csv_text(path: Path | str) -> str:
182 """Read a bounded UTF-8 CSV file after local-path checks."""
184 local_path = Path(path)
185 _check_local_file(local_path, ".csv")
186 try:
187 return local_path.read_text(encoding="utf-8-sig")
188 except UnicodeDecodeError as error:
189 raise ValidationError("INPUT_NOT_UTF8") from error
192def write_json(
193 value: dict[str, Any], output: Path | None, *, force: bool = False
194) -> None:
195 """Write deterministic JSON locally or print it to standard output."""
197 rendered = json.dumps(value, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
198 if len(rendered.encode("utf-8")) > MAX_OUTPUT_BYTES:
199 raise ValidationError("OUTPUT_TOO_LARGE")
200 if output is None:
201 print(rendered, end="")
202 return
203 if output.suffix.lower() != ".json":
204 raise ValidationError("OUTPUT_SUFFIX_NOT_ALLOWED")
205 if output.is_symlink():
206 raise ValidationError("OUTPUT_SYMLINK_NOT_ALLOWED")
207 if output.exists() and not force:
208 raise ValidationError("OUTPUT_EXISTS")
209 if not output.parent.exists() or not output.parent.is_dir():
210 raise ValidationError("OUTPUT_PARENT_NOT_FOUND")
211 output.write_text(rendered, encoding="utf-8")
214def failure_report(error: ValidationError) -> dict[str, Any]:
215 return {
216 "schema_version": SCHEMA_VERSION,
217 "status": "error",
218 "issues": [Issue(error.code, error.path).as_dict()],
219 }
222def is_number(value: Any) -> bool:
223 if not isinstance(value, (int, float)) or isinstance(value, bool):
224 return False
225 try:
226 return math.isfinite(value)
227 except OverflowError:
228 return False
231def is_identifier(value: Any) -> bool:
232 return isinstance(value, str) and bool(IDENTIFIER_RE.fullmatch(value))
235def is_nonempty_text(value: Any, maximum: int = MAX_TEXT_CHARS) -> bool:
236 return isinstance(value, str) and bool(value.strip()) and len(value) <= maximum
239def is_reference(value: Any) -> bool:
240 if not is_nonempty_text(value, 500):
241 return False
242 lowered = value.strip().lower()
243 return not (
244 lowered.startswith(("http://", "https://", "file://"))
245 or "\n" in value
246 or "\r" in value
247 )
250def is_date(value: Any) -> bool:
251 if not isinstance(value, str):
252 return False
253 try:
254 date.fromisoformat(value)
255 except ValueError:
256 return False
257 return len(value) == 10
260def exact_keys(
261 value: Any,
262 expected: Iterable[str],
263 path: str,
264 issues: list[Issue],
265) -> bool:
266 if not isinstance(value, dict):
267 issues.append(Issue("SCHEMA_OBJECT_REQUIRED", path))
268 return False
269 expected_set = set(expected)
270 actual_set = set(value)
271 for missing in sorted(expected_set - actual_set):
272 issues.append(Issue("SCHEMA_REQUIRED_FIELD", f"{path}.{missing}"))
273 for unknown in sorted(actual_set - expected_set):
274 issues.append(Issue("SCHEMA_UNKNOWN_FIELD", f"{path}.{unknown}"))
275 return not (expected_set - actual_set)
278def validate_identifier(
279 value: Any, path: str, issues: list[Issue], code: str = "IDENTIFIER_INVALID"
280) -> None:
281 if not is_identifier(value):
282 issues.append(Issue(code, path))
285def validate_reference(
286 value: Any, path: str, issues: list[Issue], *, allow_empty: bool = False
287) -> None:
288 if allow_empty and value == "":
289 return
290 if not is_reference(value):
291 issues.append(Issue("LOCAL_REFERENCE_INVALID", path))
294def validate_text(
295 value: Any, path: str, issues: list[Issue], *, allow_empty: bool = False
296) -> None:
297 if allow_empty and value == "":
298 return
299 if not is_nonempty_text(value):
300 issues.append(Issue("TEXT_INVALID", path))
303def validate_bool(value: Any, path: str, issues: list[Issue]) -> None:
304 if not isinstance(value, bool):
305 issues.append(Issue("BOOLEAN_REQUIRED", path))
308def validate_text_list(
309 value: Any,
310 path: str,
311 issues: list[Issue],
312 *,
313 minimum: int = 0,
314 maximum: int = 100,
315 references: bool = False,
316) -> None:
317 if not isinstance(value, list):
318 issues.append(Issue("LIST_REQUIRED", path))
319 return
320 if not minimum <= len(value) <= maximum:
321 issues.append(Issue("LIST_LENGTH_INVALID", path))
322 for index, item in enumerate(value):
323 if references:
324 validate_reference(item, f"{path}[{index}]", issues)
325 else:
326 validate_text(item, f"{path}[{index}]", issues)
329def _validate_scale(scale: Any, issues: list[Issue]) -> tuple[float, float, float]:
330 path = "$.scale"
331 expected = {"minimum", "maximum", "step", "anchors"}
332 if not exact_keys(scale, expected, path, issues):
333 return 0.0, 0.0, 1.0
334 minimum = scale.get("minimum")
335 maximum = scale.get("maximum")
336 step = scale.get("step")
337 if not is_number(minimum) or not is_number(maximum) or not is_number(step):
338 issues.append(Issue("SCALE_NUMBER_REQUIRED", path))
339 return 0.0, 0.0, 1.0
340 minimum_f = float(minimum)
341 maximum_f = float(maximum)
342 step_f = float(step)
343 if minimum_f < 0 or maximum_f <= minimum_f or maximum_f - minimum_f > 10:
344 issues.append(Issue("SCALE_BOUNDS_INVALID", path))
345 if step_f <= 0 or step_f > maximum_f - minimum_f:
346 issues.append(Issue("SCALE_STEP_INVALID", f"{path}.step"))
347 anchors = scale.get("anchors")
348 if not isinstance(anchors, list) or not 2 <= len(anchors) <= 21:
349 issues.append(Issue("SCALE_ANCHORS_INVALID", f"{path}.anchors"))
350 return minimum_f, maximum_f, step_f
351 anchor_scores: list[float] = []
352 for index, anchor in enumerate(anchors):
353 anchor_path = f"{path}.anchors[{index}]"
354 if not exact_keys(anchor, {"score", "label", "description"}, anchor_path, issues):
355 continue
356 score = anchor.get("score")
357 if not is_number(score):
358 issues.append(Issue("ANCHOR_SCORE_INVALID", f"{anchor_path}.score"))
359 else:
360 anchor_scores.append(float(score))
361 validate_text(anchor.get("label"), f"{anchor_path}.label", issues)
362 validate_text(anchor.get("description"), f"{anchor_path}.description", issues)
363 if anchor_scores:
364 if len(anchor_scores) != len(set(anchor_scores)):
365 issues.append(Issue("ANCHOR_SCORE_DUPLICATE", f"{path}.anchors"))
366 if min(anchor_scores) != minimum_f or max(anchor_scores) != maximum_f:
367 issues.append(Issue("ANCHOR_BOUNDS_MISSING", f"{path}.anchors"))
368 ordered = sorted(anchor_scores)
369 if any(
370 not math.isclose(right - left, step_f, rel_tol=1e-9, abs_tol=0.0)
371 for left, right in zip(ordered, ordered[1:])
372 ):
373 issues.append(Issue("ANCHOR_STEP_MISMATCH", f"{path}.anchors"))
374 return minimum_f, maximum_f, step_f
377def _validate_proxy_absence(value: Any, path: str, issues: list[Issue]) -> None:
378 strings: list[str] = []
379 stack = [value]
380 while stack:
381 current = stack.pop()
382 if isinstance(current, str):
383 strings.append(current.lower())
384 elif isinstance(current, list):
385 stack.extend(current)
386 elif isinstance(current, dict):
387 stack.extend(current.values())
388 combined = "\n".join(strings)
389 if any(term in combined for term in PROXY_TERMS):
390 issues.append(Issue("PROXY_METRIC_CRITERION_PROHIBITED", path))
393def validate_rubric(rubric: Any) -> list[Issue]:
394 """Validate the strict local rubric schema."""
396 issues: list[Issue] = []
397 top_keys = {
398 "schema_version",
399 "rubric_id",
400 "title",
401 "intended_use",
402 "construct",
403 "provenance",
404 "scale",
405 "criteria",
406 "rater_protocol",
407 "governance",
408 }
409 if not exact_keys(rubric, top_keys, "$", issues):
410 return issues
411 if rubric.get("schema_version") != SCHEMA_VERSION:
412 issues.append(Issue("SCHEMA_VERSION_UNSUPPORTED", "$.schema_version"))
413 validate_identifier(rubric.get("rubric_id"), "$.rubric_id", issues)
414 validate_text(rubric.get("title"), "$.title", issues)
416 intended = rubric.get("intended_use")
417 if exact_keys(
418 intended,
419 {"purpose", "unit_of_assessment", "allowed_contexts", "prohibited_uses"},
420 "$.intended_use",
421 issues,
422 ):
423 if intended.get("purpose") != ALLOWED_PURPOSE:
424 issues.append(Issue("PURPOSE_NOT_ALLOWED", "$.intended_use.purpose"))
425 if intended.get("unit_of_assessment") != ALLOWED_UNIT:
426 issues.append(
427 Issue("UNIT_OF_ASSESSMENT_NOT_ALLOWED", "$.intended_use.unit_of_assessment")
428 )
429 validate_text_list(
430 intended.get("allowed_contexts"),
431 "$.intended_use.allowed_contexts",
432 issues,
433 minimum=1,
434 maximum=20,
435 )
436 prohibited = intended.get("prohibited_uses")
437 validate_text_list(
438 prohibited,
439 "$.intended_use.prohibited_uses",
440 issues,
441 minimum=len(PROHIBITED_USES),
442 maximum=30,
443 )
444 prohibited_set = (
445 set(prohibited)
446 if isinstance(prohibited, list)
447 and all(isinstance(item, str) for item in prohibited)
448 else set()
449 )
450 if isinstance(prohibited, list) and not PROHIBITED_USES.issubset(
451 prohibited_set
452 ):
453 issues.append(
454 Issue("PROHIBITED_USES_INCOMPLETE", "$.intended_use.prohibited_uses")
455 )
457 construct = rubric.get("construct")
458 if exact_keys(
459 construct,
460 {"label", "definition", "boundaries", "limitations"},
461 "$.construct",
462 issues,
463 ):
464 validate_text(construct.get("label"), "$.construct.label", issues)
465 validate_text(construct.get("definition"), "$.construct.definition", issues)
466 validate_text_list(
467 construct.get("boundaries"),
468 "$.construct.boundaries",
469 issues,
470 minimum=1,
471 maximum=20,
472 )
473 validate_text_list(
474 construct.get("limitations"),
475 "$.construct.limitations",
476 issues,
477 minimum=1,
478 maximum=20,
479 )
481 provenance = rubric.get("provenance")
482 if exact_keys(
483 provenance,
484 {
485 "rubric_version",
486 "owner_role",
487 "source_ids",
488 "content_validity_status",
489 "content_validity_evidence_ref",
490 "last_reviewed_date",
491 },
492 "$.provenance",
493 issues,
494 ):
495 validate_text(provenance.get("rubric_version"), "$.provenance.rubric_version", issues)
496 validate_text(provenance.get("owner_role"), "$.provenance.owner_role", issues)
497 validate_text_list(
498 provenance.get("source_ids"),
499 "$.provenance.source_ids",
500 issues,
501 minimum=1,
502 maximum=50,
503 references=True,
504 )
505 validity_status = provenance.get("content_validity_status")
506 if validity_status not in ("not_established", "pilot_evidence", "documented"):
507 issues.append(
508 Issue(
509 "CONTENT_VALIDITY_STATUS_INVALID",
510 "$.provenance.content_validity_status",
511 )
512 )
513 validate_reference(
514 provenance.get("content_validity_evidence_ref"),
515 "$.provenance.content_validity_evidence_ref",
516 issues,
517 allow_empty=validity_status == "not_established",
518 )
519 if validity_status != "documented":
520 issues.append(
521 Issue(
522 "CONTENT_VALIDITY_NOT_DOCUMENTED",
523 "$.provenance.content_validity_status",
524 "warning",
525 )
526 )
527 if not is_date(provenance.get("last_reviewed_date")):
528 issues.append(Issue("DATE_INVALID", "$.provenance.last_reviewed_date"))
530 scale_object = rubric.get("scale")
531 minimum, maximum, step = _validate_scale(scale_object, issues)
532 raw_scale_anchors = (
533 scale_object.get("anchors") if isinstance(scale_object, dict) else []
534 )
535 if not isinstance(raw_scale_anchors, list):
536 raw_scale_anchors = []
537 scale_anchor_scores = {
538 float(anchor["score"])
539 for anchor in raw_scale_anchors
540 if isinstance(anchor, dict) and is_number(anchor.get("score"))
541 }
543 criteria = rubric.get("criteria")
544 criterion_ids: set[str] = set()
545 total_weight = 0.0
546 if not isinstance(criteria, list) or not 1 <= len(criteria) <= MAX_CRITERIA:
547 issues.append(Issue("CRITERIA_INVALID", "$.criteria"))
548 else:
549 for index, criterion in enumerate(criteria):
550 path = f"$.criteria[{index}]"
551 if not exact_keys(
552 criterion,
553 {
554 "criterion_id",
555 "label",
556 "construct_component",
557 "weight",
558 "required",
559 "anchors",
560 "evidence_requirements",
561 "limitations",
562 },
563 path,
564 issues,
565 ):
566 continue
567 criterion_id = criterion.get("criterion_id")
568 validate_identifier(criterion_id, f"{path}.criterion_id", issues)
569 if isinstance(criterion_id, str):
570 if criterion_id in criterion_ids:
571 issues.append(Issue("CRITERION_ID_DUPLICATE", f"{path}.criterion_id"))
572 criterion_ids.add(criterion_id)
573 validate_text(criterion.get("label"), f"{path}.label", issues)
574 validate_text(
575 criterion.get("construct_component"),
576 f"{path}.construct_component",
577 issues,
578 )
579 validate_bool(criterion.get("required"), f"{path}.required", issues)
580 weight = criterion.get("weight")
581 if not is_number(weight) or not 0 < float(weight) <= 1:
582 issues.append(Issue("CRITERION_WEIGHT_INVALID", f"{path}.weight"))
583 else:
584 total_weight += float(weight)
585 validate_text_list(
586 criterion.get("evidence_requirements"),
587 f"{path}.evidence_requirements",
588 issues,
589 minimum=1,
590 maximum=20,
591 )
592 validate_text_list(
593 criterion.get("limitations"),
594 f"{path}.limitations",
595 issues,
596 minimum=1,
597 maximum=20,
598 )
599 anchors = criterion.get("anchors")
600 criterion_anchor_scores: set[float] = set()
601 if not isinstance(anchors, list) or not 2 <= len(anchors) <= 21:
602 issues.append(Issue("CRITERION_ANCHORS_INVALID", f"{path}.anchors"))
603 else:
604 for anchor_index, anchor in enumerate(anchors):
605 anchor_path = f"{path}.anchors[{anchor_index}]"
606 if not exact_keys(
607 anchor, {"score", "description"}, anchor_path, issues
608 ):
609 continue
610 if not is_number(anchor.get("score")):
611 issues.append(
612 Issue("ANCHOR_SCORE_INVALID", f"{anchor_path}.score")
613 )
614 else:
615 anchor_score = float(anchor["score"])
616 if anchor_score in criterion_anchor_scores:
617 issues.append(Issue("ANCHOR_SCORE_DUPLICATE", f"{anchor_path}.score"))
618 criterion_anchor_scores.add(anchor_score)
619 validate_text(
620 anchor.get("description"),
621 f"{anchor_path}.description",
622 issues,
623 )
624 if criterion_anchor_scores != scale_anchor_scores:
625 issues.append(Issue("CRITERION_ANCHORS_INCOMPLETE", f"{path}.anchors"))
626 _validate_proxy_absence(criterion, path, issues)
627 if criteria and not math.isclose(total_weight, 1.0, abs_tol=1e-9):
628 issues.append(Issue("CRITERION_WEIGHTS_MUST_SUM_TO_ONE", "$.criteria"))
630 protocol = rubric.get("rater_protocol")
631 if exact_keys(
632 protocol,
633 {
634 "minimum_raters",
635 "training_required",
636 "training_ref",
637 "calibration_required",
638 "calibration_ref",
639 "agreement_method",
640 "inter_rater_reliability_status",
641 "inter_rater_reliability_ref",
642 "drift_monitoring_required",
643 "drift_review_ref",
644 },
645 "$.rater_protocol",
646 issues,
647 ):
648 minimum_raters = protocol.get("minimum_raters")
649 if (
650 not isinstance(minimum_raters, int)
651 or isinstance(minimum_raters, bool)
652 or not 2 <= minimum_raters <= 50
653 ):
654 issues.append(
655 Issue("MINIMUM_RATERS_INVALID", "$.rater_protocol.minimum_raters")
656 )
657 for field in (
658 "training_required",
659 "calibration_required",
660 "drift_monitoring_required",
661 ):
662 validate_bool(protocol.get(field), f"$.rater_protocol.{field}", issues)
663 if protocol.get(field) is not True:
664 issues.append(
665 Issue("RATER_CONTROL_MUST_BE_REQUIRED", f"$.rater_protocol.{field}")
666 )
667 for field in ("training_ref", "calibration_ref", "drift_review_ref"):
668 validate_reference(protocol.get(field), f"$.rater_protocol.{field}", issues)
669 if protocol.get("agreement_method") != (
670 "exact_within_step_and_mean_absolute_difference"
671 ):
672 issues.append(
673 Issue(
674 "AGREEMENT_METHOD_INVALID", "$.rater_protocol.agreement_method"
675 )
676 )
677 reliability_status = protocol.get("inter_rater_reliability_status")
678 if reliability_status not in (
679 "not_established",
680 "pilot_evidence",
681 "documented",
682 ):
683 issues.append(
684 Issue(
685 "INTER_RATER_RELIABILITY_STATUS_INVALID",
686 "$.rater_protocol.inter_rater_reliability_status",
687 )
688 )
689 validate_reference(
690 protocol.get("inter_rater_reliability_ref"),
691 "$.rater_protocol.inter_rater_reliability_ref",
692 issues,
693 allow_empty=reliability_status == "not_established",
694 )
695 if reliability_status != "documented":
696 issues.append(
697 Issue(
698 "INTER_RATER_RELIABILITY_NOT_DOCUMENTED",
699 "$.rater_protocol.inter_rater_reliability_status",
700 "warning",
701 )
702 )
704 governance = rubric.get("governance")
705 governance_bool_fields = (
706 "accountable_committee_required",
707 "conflict_disclosure_required",
708 "recusal_required",
709 "appeal_process_required",
710 "accessibility_accommodations_required",
711 "data_protection_review_required",
712 "subgroup_bias_review_required",
713 )
714 governance_ref_fields = (
715 "committee_owner_role",
716 "appeal_process_ref",
717 "accessibility_process_ref",
718 "data_protection_process_ref",
719 "subgroup_review_ref",
720 "review_cycle_ref",
721 )
722 if exact_keys(
723 governance,
724 set(governance_bool_fields) | set(governance_ref_fields),
725 "$.governance",
726 issues,
727 ):
728 for field in governance_bool_fields:
729 validate_bool(governance.get(field), f"$.governance.{field}", issues)
730 if governance.get(field) is not True:
731 issues.append(
732 Issue("GOVERNANCE_CONTROL_MUST_BE_REQUIRED", f"$.governance.{field}")
733 )
734 for field in governance_ref_fields:
735 validate_reference(governance.get(field), f"$.governance.{field}", issues)
737 if maximum <= minimum or step <= 0:
738 issues.append(Issue("SCALE_UNUSABLE", "$.scale"))
739 return issues
742def scale_values(rubric: dict[str, Any]) -> set[float]:
743 return {
744 float(anchor["score"])
745 for anchor in rubric["scale"]["anchors"]
746 if is_number(anchor["score"])
747 }
750def validate_evaluation(
751 evaluation: Any, rubric: dict[str, Any]
752) -> list[Issue]:
753 """Validate one strict scholarly-work evaluation record."""
755 issues: list[Issue] = []
756 if not exact_keys(
757 evaluation,
758 {
759 "schema_version",
760 "evaluation_id",
761 "rubric_id",
762 "work_id",
763 "data_classification",
764 "purpose",
765 "ratings",
766 },
767 "$",
768 issues,
769 ):
770 return issues
771 if evaluation.get("schema_version") != SCHEMA_VERSION:
772 issues.append(Issue("SCHEMA_VERSION_UNSUPPORTED", "$.schema_version"))
773 validate_identifier(evaluation.get("evaluation_id"), "$.evaluation_id", issues)
774 validate_identifier(evaluation.get("rubric_id"), "$.rubric_id", issues)
775 validate_identifier(evaluation.get("work_id"), "$.work_id", issues)
776 if evaluation.get("rubric_id") != rubric.get("rubric_id"):
777 issues.append(Issue("RUBRIC_ID_MISMATCH", "$.rubric_id"))
778 if evaluation.get("data_classification") not in ALLOWED_CLASSIFICATIONS:
779 issues.append(Issue("DATA_CLASSIFICATION_NOT_ALLOWED", "$.data_classification"))
780 if evaluation.get("purpose") != ALLOWED_PURPOSE:
781 issues.append(Issue("PURPOSE_NOT_ALLOWED", "$.purpose"))
783 criteria = {criterion["criterion_id"]: criterion for criterion in rubric["criteria"]}
784 accepted_scores = scale_values(rubric)
785 maximum_uncertainty = float(rubric["scale"]["maximum"]) - float(
786 rubric["scale"]["minimum"]
787 )
788 ratings = evaluation.get("ratings")
789 seen: set[str] = set()
790 if not isinstance(ratings, list) or not 1 <= len(ratings) <= MAX_CRITERIA:
791 issues.append(Issue("RATINGS_INVALID", "$.ratings"))
792 return issues
793 for index, rating in enumerate(ratings):
794 path = f"$.ratings[{index}]"
795 if not exact_keys(
796 rating,
797 {
798 "criterion_id",
799 "status",
800 "score",
801 "uncertainty",
802 "evidence_ids",
803 "rationale_ref",
804 },
805 path,
806 issues,
807 ):
808 continue
809 criterion_id = rating.get("criterion_id")
810 validate_identifier(criterion_id, f"{path}.criterion_id", issues)
811 if isinstance(criterion_id, str):
812 if criterion_id in seen:
813 issues.append(Issue("RATING_DUPLICATE", f"{path}.criterion_id"))
814 seen.add(criterion_id)
815 if criterion_id not in criteria:
816 issues.append(Issue("CRITERION_UNKNOWN", f"{path}.criterion_id"))
817 status = rating.get("status")
818 if status not in RATING_STATUSES:
819 issues.append(Issue("RATING_STATUS_INVALID", f"{path}.status"))
820 evidence_ids = rating.get("evidence_ids")
821 if not isinstance(evidence_ids, list) or len(evidence_ids) > 100:
822 issues.append(Issue("EVIDENCE_IDS_INVALID", f"{path}.evidence_ids"))
823 evidence_ids = []
824 else:
825 hashable_evidence_ids = [
826 evidence_id
827 for evidence_id in evidence_ids
828 if isinstance(evidence_id, str)
829 ]
830 if len(hashable_evidence_ids) != len(set(hashable_evidence_ids)):
831 issues.append(Issue("EVIDENCE_ID_DUPLICATE", f"{path}.evidence_ids"))
832 for evidence_index, evidence_id in enumerate(evidence_ids):
833 validate_identifier(
834 evidence_id,
835 f"{path}.evidence_ids[{evidence_index}]",
836 issues,
837 "EVIDENCE_ID_INVALID",
838 )
839 validate_reference(rating.get("rationale_ref"), f"{path}.rationale_ref", issues)
840 if status == "rated":
841 score = rating.get("score")
842 uncertainty = rating.get("uncertainty")
843 if not is_number(score) or float(score) not in accepted_scores:
844 issues.append(Issue("SCORE_NOT_ON_SCALE", f"{path}.score"))
845 if (
846 not is_number(uncertainty)
847 or not 0 <= float(uncertainty) <= maximum_uncertainty
848 ):
849 issues.append(Issue("UNCERTAINTY_INVALID", f"{path}.uncertainty"))
850 if not evidence_ids:
851 issues.append(Issue("RATED_EVIDENCE_REQUIRED", f"{path}.evidence_ids"))
852 elif status in ("missing", "not_applicable"):
853 if rating.get("score") is not None:
854 issues.append(Issue("UNRATED_SCORE_MUST_BE_NULL", f"{path}.score"))
855 if rating.get("uncertainty") is not None:
856 issues.append(
857 Issue("UNRATED_UNCERTAINTY_MUST_BE_NULL", f"{path}.uncertainty")
858 )
859 if evidence_ids:
860 issues.append(
861 Issue("UNRATED_EVIDENCE_MUST_BE_EMPTY", f"{path}.evidence_ids")
862 )
863 missing_criteria = set(criteria) - seen
864 for criterion_id in sorted(missing_criteria):
865 issues.append(Issue("CRITERION_RATING_REQUIRED", f"$.ratings.{criterion_id}"))
866 return issues
869def error_issues(issues: Iterable[Issue]) -> list[Issue]:
870 return [issue for issue in issues if issue.level == "error"]
873def require_valid(issues: Iterable[Issue]) -> None:
874 errors = error_issues(issues)
875 if errors:
876 raise ValidationError(errors[0].code, errors[0].path)
879def rounded(value: float | None) -> float | None:
880 if value is None:
881 return None
882 return round(value, 6)
885def weights_by_criterion(rubric: dict[str, Any]) -> dict[str, float]:
886 return {
887 criterion["criterion_id"]: float(criterion["weight"])
888 for criterion in rubric["criteria"]
889 }
892def score_evaluation(
893 rubric: dict[str, Any],
894 evaluation: dict[str, Any],
895 *,
896 weights: dict[str, float] | None = None,
897 round_output: bool = True,
898) -> dict[str, Any]:
899 """Calculate bounded descriptive rubric math without a recommendation."""
901 selected_weights = weights or weights_by_criterion(rubric)
902 output_number = rounded if round_output else lambda value: value
903 scale_minimum = float(rubric["scale"]["minimum"])
904 scale_maximum = float(rubric["scale"]["maximum"])
905 rating_by_id = {
906 rating["criterion_id"]: rating for rating in evaluation["ratings"]
907 }
908 criteria_output: list[dict[str, Any]] = []
909 weighted_sum = 0.0
910 lower_sum = 0.0
911 upper_sum = 0.0
912 rated_weight = 0.0
913 missing_weight = 0.0
914 not_applicable_weight = 0.0
915 for criterion in rubric["criteria"]:
916 criterion_id = criterion["criterion_id"]
917 rating = rating_by_id[criterion_id]
918 weight = float(selected_weights[criterion_id])
919 status = rating["status"]
920 item: dict[str, Any] = {
921 "criterion_id": criterion_id,
922 "status": status,
923 "weight": output_number(weight),
924 "score": None,
925 "uncertainty": None,
926 "weighted_contribution": None,
927 "lower_contribution": None,
928 "upper_contribution": None,
929 }
930 if status == "rated":
931 score = float(rating["score"])
932 uncertainty = float(rating["uncertainty"])
933 lower = max(scale_minimum, score - uncertainty)
934 upper = min(scale_maximum, score + uncertainty)
935 contribution = weight * score
936 weighted_sum += contribution
937 lower_sum += weight * lower
938 upper_sum += weight * upper
939 rated_weight += weight
940 item.update(
941 {
942 "score": output_number(score),
943 "uncertainty": output_number(uncertainty),
944 "weighted_contribution": output_number(contribution),
945 "lower_contribution": output_number(weight * lower),
946 "upper_contribution": output_number(weight * upper),
947 }
948 )
949 elif status == "missing":
950 missing_weight += weight
951 else:
952 not_applicable_weight += weight
953 criteria_output.append(item)
954 applicable_weight = rated_weight + missing_weight
955 normalized = weighted_sum / rated_weight if rated_weight else None
956 lower_score = lower_sum / rated_weight if rated_weight else None
957 upper_score = upper_sum / rated_weight if rated_weight else None
958 coverage = rated_weight / applicable_weight if applicable_weight else None
959 warnings: list[str] = []
960 if missing_weight > 0:
961 warnings.append("MISSING_RATINGS_EXCLUDED_FROM_NORMALIZED_SCORE")
962 if not_applicable_weight > 0:
963 warnings.append("NOT_APPLICABLE_RATINGS_EXCLUDED")
964 if coverage is None or coverage < 1.0:
965 warnings.append("INCOMPLETE_APPLICABLE_COVERAGE")
966 if normalized is None:
967 warnings.append("NO_NORMALIZED_SCORE_AVAILABLE")
968 return {
969 "schema_version": SCHEMA_VERSION,
970 "report_type": "bounded_descriptive_rubric_score",
971 "notice": NOTICE,
972 "evaluation_id": evaluation["evaluation_id"],
973 "work_id": evaluation["work_id"],
974 "rubric_id": rubric["rubric_id"],
975 "scale": {"minimum": scale_minimum, "maximum": scale_maximum},
976 "formula": (
977 "normalized_score = sum(score_i * weight_i for rated criteria) "
978 "/ sum(weight_i for rated criteria); missing and not_applicable "
979 "criteria are reported and excluded"
980 ),
981 "criteria": criteria_output,
982 "aggregates": {
983 "total_weight": output_number(sum(selected_weights.values())),
984 "applicable_weight": output_number(applicable_weight),
985 "rated_weight": output_number(rated_weight),
986 "missing_weight": output_number(missing_weight),
987 "not_applicable_weight": output_number(not_applicable_weight),
988 "coverage_of_applicable_weight": output_number(coverage),
989 "weighted_sum": output_number(weighted_sum),
990 "normalized_score": output_number(normalized),
991 "uncertainty_interval": {
992 "lower": output_number(lower_score),
993 "upper": output_number(upper_score),
994 "method": (
995 "weighted aggregation of criterion-level bounded uncertainty "
996 "intervals; not a confidence interval"
997 ),
998 },
999 },
1000 "warnings": warnings,
1001 "decision_recommendation_provided": False,
1002 }