103 lines
4.8 KiB
Python
103 lines
4.8 KiB
Python
"""Schemas and independent ownership/evidence checks for the local pilot."""
|
|
|
|
from collections import Counter
|
|
|
|
|
|
def obj(properties):
|
|
"""Require all documented properties and reject unspecified output fields."""
|
|
return {"type": "object", "properties": properties, "required": list(properties),
|
|
"additionalProperties": False}
|
|
|
|
|
|
def array(items):
|
|
"""Describe an array without forcing invented entries into unknown fields."""
|
|
return {"type": "array", "items": items}
|
|
|
|
|
|
TEXT = {"type": "string"}
|
|
EVIDENCE = obj({"field": TEXT, "quote": TEXT})
|
|
FACT = obj({"value": TEXT, "evidence": array(EVIDENCE)})
|
|
DIMENSIONS = ("target_interface", "setup_fixtures", "action_sequence",
|
|
"observations_assertions", "special_mechanisms", "parameters")
|
|
PROFILE = obj({
|
|
"campaign_id": TEXT, "suite_id": TEXT, "case_id": TEXT,
|
|
"facts": obj({key: {"anyOf": [array(FACT), {"type": "null"}]} for key in DIMENSIONS}),
|
|
"inferred_suggestions": array(obj({"suggestion": TEXT, "supporting_fields": array(TEXT),
|
|
"uncertainty": TEXT})),
|
|
"missing_information": array(TEXT),
|
|
})
|
|
FAMILY = obj({
|
|
"campaign_id": TEXT, "suite_id": TEXT,
|
|
"families": array(obj({
|
|
"family_name": TEXT, "implementation_task_title": TEXT,
|
|
"member_case_ids": array(TEXT), "common_implementation_approach": TEXT,
|
|
"supporting_code": array(TEXT), "why_together": TEXT,
|
|
"objectives": array(obj({"case_id": TEXT, "objective": TEXT, "variations": array(TEXT)})),
|
|
"source_evidence": array(obj({"case_id": TEXT, "field": TEXT, "quote": TEXT})),
|
|
"inferred_implementation_suggestions": array(TEXT),
|
|
"uncertainties": array(TEXT), "singleton_reason": {"type": ["string", "null"]},
|
|
})),
|
|
})
|
|
|
|
|
|
def _evidence(item, fields):
|
|
"""Check exact evidence provenance, without claiming to judge its semantics."""
|
|
field, quote = item.get("field"), item.get("quote")
|
|
if field not in fields or not isinstance(quote, str) or not quote or quote not in str(fields[field]):
|
|
raise ValueError("unsupported source evidence")
|
|
|
|
|
|
def validate_profile(profile, source):
|
|
"""Reject changed identifiers and invented field references/quotations."""
|
|
for key in ("campaign_id", "suite_id", "case_id"):
|
|
if profile.get(key) != source[key]:
|
|
raise ValueError("profile ownership/ID mismatch")
|
|
if set(profile["facts"]) != set(DIMENSIONS):
|
|
raise ValueError("profile dimensions missing or invented")
|
|
for facts in profile["facts"].values():
|
|
if facts is None:
|
|
continue
|
|
if not isinstance(facts, list):
|
|
raise ValueError("facts must be an array or null")
|
|
for fact in facts:
|
|
if not isinstance(fact.get("value"), str) or not fact.get("evidence"):
|
|
raise ValueError("facts require source evidence")
|
|
for evidence in fact["evidence"]:
|
|
_evidence(evidence, source["fields"])
|
|
for suggestion in profile["inferred_suggestions"]:
|
|
if set(suggestion["supporting_fields"]) - set(source["fields"]):
|
|
raise ValueError("suggestion references invented fields")
|
|
|
|
|
|
def validate_families(result, sources):
|
|
"""Require an exact suite partition and independently traceable objectives."""
|
|
owners = {(case["campaign_id"], case["suite_id"]) for case in sources}
|
|
if len(owners) != 1 or (result.get("campaign_id"), result.get("suite_id")) not in owners:
|
|
raise ValueError("family campaign/suite mismatch")
|
|
cases = {case["case_id"]: case for case in sources}
|
|
if len(cases) != len(sources):
|
|
raise ValueError("duplicate source case IDs")
|
|
assigned = []
|
|
singletons = 0
|
|
for family in result["families"]:
|
|
members = family["member_case_ids"]
|
|
if not members or any(member not in cases for member in members):
|
|
raise ValueError("empty family or invented case IDs")
|
|
assigned.extend(members)
|
|
objectives = [item["case_id"] for item in family["objectives"]]
|
|
if Counter(objectives) != Counter(members):
|
|
raise ValueError("family objectives do not match members exactly")
|
|
if len(members) == 1:
|
|
singletons += 1
|
|
if not family.get("singleton_reason"):
|
|
raise ValueError("singleton requires an engineering explanation")
|
|
for evidence in family["source_evidence"]:
|
|
if evidence["case_id"] not in members:
|
|
raise ValueError("evidence refers to a case outside the family")
|
|
_evidence(evidence, cases[evidence["case_id"]]["fields"])
|
|
if Counter(assigned) != Counter(cases.keys()):
|
|
raise ValueError("suite contains omitted or multiply assigned case IDs")
|
|
return {"case_count": len(cases), "family_count": len(result["families"]),
|
|
"singleton_families": singletons,
|
|
"singleton_family_fraction": singletons / len(result["families"])}
|