Validation - Package Integrity

from __future__ import annotations

import json
from collections import Counter
from pathlib import Path
from typing import Any, Iterable

from hash_utils import load_json, sha256_json, sha256_text
from validate_examples import validate_examples

ROOT = Path(__file__).resolve().parents[1]
EXAMPLES = ROOT / "examples"


def duplicates(values: Iterable[str]) -> list[str]:
 counts = Counter(values)
 return sorted(value for value, count in counts.items() if count > 1)


def check_hash(errors: list[str], label: str, actual: str, expected: str) -> None:
 if actual != expected:
 errors.append(f"{label}: expected {expected}, found {actual}")


def build_artifact_text(case: dict[str, Any]) -> dict[tuple[str, str], str]:
 artifacts: dict[tuple[str, str], str] = {
 ("task", case["task"]["taskId"]): case["task"]["prompt"],
 ("attempt", case["attempt"]["attemptId"]): case["attempt"]["content"],
 }
 for evidence in case["evidenceBundle"]:
 artifacts[("evidence", evidence["evidenceId"])] = evidence["content"]
 for reference in case.get("referenceAnswers", []):
 artifacts[("reference", reference["referenceId"])] = reference["content"]
 for trace in case["attempt"].get("toolTraces", []):
 artifacts[("tool_trace", trace["traceId"])] = json.dumps(
 {"input": trace["input"], "output": trace["output"]},
 ensure_ascii=False,
 sort_keys=True,
 )
 review = case.get("humanReview")
 if review:
 review_text = "\n".join(
 [
 review.get("label", ""),
 " ".join(review.get("reasonCodes", [])),
 review.get("rationale", ""),
 *[
 annotation.get("rationale", "")
 for annotation in review.get("criterionAnnotations", [])
 ],
 ]
 )
 artifacts[("human_review", review["annotationId"])] = review_text
 return artifacts


def check_span(errors: list[str], prefix: str, span: dict[str, Any], artifacts: dict[tuple[str, str], str]) -> None:
 key = (span["artifactType"], span["artifactId"])
 if key not in artifacts:
 errors.append(f"{prefix}: unknown artifact {key}")
 return
 text = artifacts[key]
 quote = span["quotedText"]
 start = span.get("startChar")
 end = span.get("endChar")
 if start is not None and end is not None:
 if start > end or end > len(text):
 errors.append(f"{prefix}: invalid character range {start}:{end} for length {len(text)}")
 elif text[start:end] != quote:
 errors.append(f"{prefix}: quotedText does not match declared character range")
 elif quote not in text:
 errors.append(f"{prefix}: quotedText not found in artifact {key}")


def validate_referential_integrity() -> list[str]:
 errors: list[str] = []
 contract = load_json(EXAMPLES / "project-contract.example.json")
 case = load_json(EXAMPLES / "case-input.example.json")
 evaluation = load_json(EXAMPLES / "evaluation-output.example.json")
 benchmark = load_json(EXAMPLES / "benchmark-item.example.json")
 resolution_packet = load_json(EXAMPLES / "human-resolution-packet.example.json")

 contract_hash = sha256_json(contract)
 case_hash = sha256_json(case)

 # Contract content and identity.
 instruction_ids = [doc["instructionId"] for doc in contract["instructionDocuments"]]
 criterion_ids = [criterion["criterionId"] for criterion in contract["criteria"]]
 evidence_class_ids = [item["classId"] for item in contract["sourcePolicy"]["evidenceClasses"]]
 rule_ids = [rule["ruleId"] for rule in contract["decisionPolicy"]["rules"]]

 for label, values in [
 ("instruction IDs", instruction_ids),
 ("criterion IDs", criterion_ids),
 ("evidence class IDs", evidence_class_ids),
 ("decision rule IDs", rule_ids),
 ]:
 dupes = duplicates(values)
 if dupes:
 errors.append(f"Duplicate {label}: {dupes}")

 instruction_by_id = {doc["instructionId"]: doc for doc in contract["instructionDocuments"]}
 criterion_by_id = {criterion["criterionId"]: criterion for criterion in contract["criteria"]}
 rule_by_id = {rule["ruleId"]: rule for rule in contract["decisionPolicy"]["rules"]}
 declared_targets = set(contract["decisionTargets"])

 for doc in contract["instructionDocuments"]:
 check_hash(errors, f"instruction {doc['instructionId']} hash", doc["contentHash"], sha256_text(doc["content"]))

 for criterion in contract["criteria"]:
 cid = criterion["criterionId"]
 for ref in criterion["instructionRefs"]:
 if ref["instructionId"] not in instruction_by_id:
 errors.append(f"{cid}: unknown instruction ref {ref['instructionId']}")
 elif ref.get("quotedText") and ref["quotedText"] not in instruction_by_id[ref["instructionId"]]["content"]:
 errors.append(f"{cid}: quoted instruction text not found in {ref['instructionId']}")
 for eid in criterion["evidencePolicy"]["allowedEvidenceClassIds"]:
 if eid not in evidence_class_ids:
 errors.append(f"{cid}: unknown evidence class {eid}")
 for dep in criterion["dependencies"]["requiredBeforeCriterionIds"] + criterion["dependencies"]["impliesCriterionIds"]:
 if dep not in criterion_by_id:
 errors.append(f"{cid}: unknown dependency {dep}")
 for other in criterion["precedence"]["overridesCriterionIds"] + criterion["precedence"]["overriddenByCriterionIds"]:
 if other not in criterion_by_id:
 errors.append(f"{cid}: unknown precedence criterion {other}")

 for rule in contract["decisionPolicy"]["rules"]:
 if rule["target"] not in declared_targets:
 errors.append(f"Rule {rule['ruleId']} targets undeclared target {rule['target']}")
 for cid in rule["match"].get("criterionIds", []):
 if cid != "*" and cid not in criterion_by_id:
 errors.append(f"Rule {rule['ruleId']} references unknown criterion {cid}")

 # Case hashes and references.
 if case["contractRef"]["contractId"] != contract["contractId"]:
 errors.append("Case contractId does not match contract")
 if case["contractRef"]["contractVersion"] != contract["contractVersion"]:
 errors.append("Case contractVersion does not match contract")
 check_hash(errors, "case contract hash", case["contractRef"]["contentHash"], contract_hash)
 check_hash(errors, "task content hash", case["task"]["contentHash"], sha256_text(case["task"]["prompt"]))
 check_hash(errors, "attempt content hash", case["attempt"]["contentHash"], sha256_text(case["attempt"]["content"]))

 evidence_ids = [evidence["evidenceId"] for evidence in case["evidenceBundle"]]
 if duplicates(evidence_ids):
 errors.append(f"Duplicate case evidence IDs: {duplicates(evidence_ids)}")
 for evidence in case["evidenceBundle"]:
 if evidence["evidenceClassId"] not in evidence_class_ids:
 errors.append(f"Evidence {evidence['evidenceId']} uses unknown class {evidence['evidenceClassId']}")
 check_hash(errors, f"evidence {evidence['evidenceId']} hash", evidence["contentHash"], sha256_text(evidence["content"]))
 for reference in case.get("referenceAnswers", []):
 check_hash(errors, f"reference {reference['referenceId']} hash", reference["contentHash"], sha256_text(reference["content"]))
 for annotation in case.get("humanReview", {}).get("criterionAnnotations", []):
 if annotation["criterionId"] not in criterion_by_id:
 errors.append(f"Human review references unknown criterion {annotation['criterionId']}")

 artifacts = build_artifact_text(case)
 for annotation_index, annotation in enumerate(case.get("humanReview", {}).get("criterionAnnotations", [])):
 for span_index, span in enumerate(annotation.get("citedSpans", [])):
 check_span(errors, f"humanReview criterionAnnotations[{annotation_index}].citedSpans[{span_index}]", span, artifacts)

 # Evaluation references and internal links.
 if evaluation["caseRef"]["caseId"] != case["caseId"]:
 errors.append("Evaluation caseId does not match case")
 check_hash(errors, "evaluation case hash", evaluation["caseRef"]["contentHash"], case_hash)
 check_hash(errors, "evaluation contract hash", evaluation["contractRef"]["contentHash"], contract_hash)
 check_hash(errors, "audit case hash", evaluation["audit"]["caseInputHash"], case_hash)
 check_hash(errors, "audit contract hash", evaluation["audit"]["contractHash"], contract_hash)

 finding_ids = [finding["findingId"] for finding in evaluation["findings"]]
 claim_ids = [claim["claimId"] for claim in evaluation["claims"]]
 expected_ids = [element["elementId"] for element in evaluation["expectedElements"]]
 relationship_ids = [rel["relationshipId"] for rel in evaluation["evidenceRelationships"]]
 run_ids = [run["modelRunId"] for run in evaluation["runManifest"]]
 validator_run_ids = [run["validatorRunId"] for run in evaluation["validatorManifest"]]
 validator_ids = [run["validatorId"] for run in evaluation["validatorManifest"]]
 for label, values in [
 ("finding IDs", finding_ids),
 ("claim IDs", claim_ids),
 ("expected-element IDs", expected_ids),
 ("evidence-relationship IDs", relationship_ids),
 ("model-run IDs", run_ids),
 ("validator-run IDs", validator_run_ids),
 ]:
 dupes = duplicates(values)
 if dupes:
 errors.append(f"Duplicate evaluation {label}: {dupes}")

 run_id_set = set(run_ids)
 validator_id_set = set(validator_ids)
 for run in evaluation["runManifest"]:
 for parent_id in run.get("parentRunIds", []):
 if parent_id not in run_id_set:
 errors.append(f"Model run {run['modelRunId']} references unknown parent run {parent_id}")

 applicable_assessment_ids = {
 assessment["criterionId"]
 for assessment in evaluation["criterionAssessments"]
 if assessment["applicability"] == "applies"
 }
 for cid in applicable_assessment_ids:
 for validator_id in criterion_by_id[cid]["evidencePolicy"]["deterministicValidatorIds"]:
 if validator_id not in validator_id_set:
 errors.append(f"Applicable criterion {cid} requires missing validator run {validator_id}")

 finding_by_id = {finding["findingId"]: finding for finding in evaluation["findings"]}
 claim_set = set(claim_ids)
 expected_set = set(expected_ids)
 evidence_set = set(evidence_ids)

 for index, finding in enumerate(evaluation["findings"]):
 fid = finding["findingId"]
 if finding.get("criterionId") and finding["criterionId"] not in criterion_by_id:
 errors.append(f"Finding {fid} references unknown criterion {finding['criterionId']}")
 for claim_id in finding.get("claimIds", []):
 if claim_id not in claim_set:
 errors.append(f"Finding {fid} references unknown claim {claim_id}")
 for element_id in finding.get("expectedElementIds", []):
 if element_id not in expected_set:
 errors.append(f"Finding {fid} references unknown expected element {element_id}")
 for evidence_id in finding.get("evidenceIds", []):
 if evidence_id not in evidence_set:
 errors.append(f"Finding {fid} references unknown evidence {evidence_id}")
 source = finding["source"]
 if source.get("modelRunId") and source["modelRunId"] not in run_id_set:
 errors.append(f"Finding {fid} references unknown model run {source['modelRunId']}")
 if source["kind"] == "deterministic_validator" and source["componentId"] not in validator_id_set:
 errors.append(f"Finding {fid} references unknown deterministic validator {source['componentId']}")
 for span_index, span in enumerate(finding.get("targetSpans", [])):
 check_span(errors, f"findings[{index}].targetSpans[{span_index}]", span, artifacts)

 for index, claim in enumerate(evaluation["claims"]):
 for evidence_id in claim.get("citedEvidenceIds", []):
 if evidence_id not in evidence_set:
 errors.append(f"Claim {claim['claimId']} cites unknown evidence {evidence_id}")
 for span_index, span in enumerate(claim["attemptSpans"]):
 check_span(errors, f"claims[{index}].attemptSpans[{span_index}]", span, artifacts)

 for index, element in enumerate(evaluation["expectedElements"]):
 if element["criterionId"] not in criterion_by_id:
 errors.append(f"Expected element {element['elementId']} references unknown criterion")
 for span_index, span in enumerate(element.get("observedSpans", [])):
 check_span(errors, f"expectedElements[{index}].observedSpans[{span_index}]", span, artifacts)

 for rel in evaluation["evidenceRelationships"]:
 if rel["evidenceId"] not in evidence_set:
 errors.append(f"Relationship {rel['relationshipId']} references unknown evidence")
 subject_id = rel["subjectId"]
 if rel["subjectType"] == "claim" and subject_id not in claim_set:
 errors.append(f"Relationship {rel['relationshipId']} references unknown claim {subject_id}")
 if rel["subjectType"] == "expected_element" and subject_id not in expected_set:
 errors.append(f"Relationship {rel['relationshipId']} references unknown expected element {subject_id}")
 if rel["subjectType"] == "finding" and subject_id not in finding_by_id:
 errors.append(f"Relationship {rel['relationshipId']} references unknown finding {subject_id}")
 for span_index, span in enumerate(rel.get("evidenceSpans", [])):
 check_span(errors, f"relationship {rel['relationshipId']} evidenceSpans[{span_index}]", span, artifacts)

 assessment_ids = [assessment["criterionId"] for assessment in evaluation["criterionAssessments"]]
 if duplicates(assessment_ids):
 errors.append(f"Duplicate criterion assessments: {duplicates(assessment_ids)}")
 for assessment in evaluation["criterionAssessments"]:
 cid = assessment["criterionId"]
 if cid not in criterion_by_id:
 errors.append(f"Assessment references unknown criterion {cid}")
 continue
 all_finding_refs = (
 assessment["positiveFindingIds"]
 + assessment["negativeFindingIds"]
 + assessment["omissionFindingIds"]
 + assessment["counterEvidenceFindingIds"]
 + assessment["unresolvedFindingIds"]
 )
 for fid in all_finding_refs:
 if fid not in finding_by_id:
 errors.append(f"Assessment {cid} references unknown finding {fid}")
 if assessment["verdict"] == "met" and criterion_by_id[cid]["proofPolicy"]["positiveProofRequired"]:
 confirmed_positive = [
 fid for fid in assessment["positiveFindingIds"]
 if fid in finding_by_id
 and finding_by_id[fid]["findingType"] == "positive"
 and finding_by_id[fid]["status"] == "confirmed"
 ]
 if not confirmed_positive:
 errors.append(f"Assessment {cid} is met without confirmed positive proof")
 if assessment["verdict"] == "not_met":
 confirmed_failure = [
 fid for fid in assessment["negativeFindingIds"] + assessment["omissionFindingIds"]
 if fid in finding_by_id and finding_by_id[fid]["status"] == "confirmed"
 ]
 if not confirmed_failure:
 errors.append(f"Assessment {cid} is not_met without confirmed negative or omission evidence")
 if assessment["applicability"] == "does_not_apply" and assessment["verdict"] != "not_applicable":
 errors.append(f"Assessment {cid} does not apply but verdict is {assessment['verdict']}")

 for decision_list_name in ["provisionalDecisions", "finalDecisions"]:
 decisions = evaluation[decision_list_name]
 targets = [decision["target"] for decision in decisions]
 if duplicates(targets):
 errors.append(f"Duplicate {decision_list_name} targets: {duplicates(targets)}")
 missing = declared_targets - set(targets)
 if missing:
 errors.append(f"{decision_list_name} missing declared targets: {sorted(missing)}")
 for decision in decisions:
 if decision["target"] not in declared_targets:
 errors.append(f"Decision targets undeclared target {decision['target']}")
 for rid in decision["triggeredRuleIds"]:
 if rid not in rule_by_id:
 errors.append(f"Decision references unknown rule {rid}")
 elif rule_by_id[rid]["target"] != decision["target"]:
 errors.append(f"Decision target {decision['target']} uses rule {rid} for {rule_by_id[rid]['target']}")
 for cid in decision["decisiveCriterionIds"]:
 if cid not in criterion_by_id:
 errors.append(f"Decision references unknown decisive criterion {cid}")
 for fid in decision["unresolvedFindingIds"]:
 if fid not in finding_by_id:
 errors.append(f"Decision references unknown unresolved finding {fid}")

 for feedback in evaluation["feedback"]:
 if feedback.get("criterionId") and feedback["criterionId"] not in criterion_by_id:
 errors.append(f"Feedback {feedback['feedbackId']} references unknown criterion")
 for fid in feedback["findingIds"]:
 if fid not in finding_by_id:
 errors.append(f"Feedback {feedback['feedbackId']} references unknown finding {fid}")

 # Benchmark links.
 check_hash(errors, "benchmark contract hash", benchmark["contractRef"]["contentHash"], contract_hash)
 check_hash(errors, "benchmark case hash", benchmark["caseRef"]["contentHash"], case_hash)
 for criterion_gold in benchmark["goldStandard"]["criterionGold"]:
 if criterion_gold["criterionId"] not in criterion_by_id:
 errors.append(f"Benchmark references unknown criterion {criterion_gold['criterionId']}")
 for field in ["positiveEvidenceSpans", "negativeEvidenceSpans"]:
 for index, span in enumerate(criterion_gold[field]):
 check_span(errors, f"benchmark {criterion_gold['criterionId']} {field}[{index}]", span, artifacts)
 for target_decision in benchmark["goldStandard"]["acceptableDecisions"]:
 if target_decision["target"] not in declared_targets:
 errors.append(f"Benchmark acceptable decision targets undeclared target {target_decision['target']}")


 # Standalone human-resolution packet integrity.
 packet_finding_ids = {finding["findingId"] for finding in resolution_packet["contextFindings"]}
 request_ids = set(resolution_packet["interaction"]["request"]["contextFindingIds"])
 if packet_finding_ids != request_ids:
 errors.append(
 "Human-resolution packet context findings do not exactly match request contextFindingIds"
 )
 if resolution_packet["interaction"]["decisionTarget"] not in declared_targets:
 errors.append("Human-resolution packet targets an undeclared decision target")
 for cid in resolution_packet["interaction"]["request"]["triggerCriterionIds"]:
 if cid not in criterion_by_id:
 errors.append(f"Human-resolution packet references unknown criterion {cid}")

 return errors


def main() -> None:
 schema_failures = validate_examples()
 if schema_failures:
 raise SystemExit(1)

 errors = validate_referential_integrity()
 if errors:
 print("\nREFERENTIAL INTEGRITY FAILURES")
 for error in errors:
 print(f" - {error}")
 raise SystemExit(1)

 print("PASS cross-document referential integrity")
 print("PASS canonical example hashes")
 print("PASS positive-proof and failure-proof invariants")


if __name__ == "__main__":
 main()