856 lines
40 KiB
Python
856 lines
40 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
REPO_ROOT = SCRIPT_DIR.parents[1]
|
|
MATRIX_POLICY_PATH = SCRIPT_DIR / "student_loop_matrix_policy.json"
|
|
DEFAULT_POLICY_PATH = SCRIPT_DIR / "student_loop_policy.json"
|
|
RUNS_ROOT = REPO_ROOT / ".app" / "student-loop" / "runs"
|
|
MATRIX_RUNS_ROOT = REPO_ROOT / ".app" / "student-loop" / "matrix-runs"
|
|
STUDENT_LOOP_COMMAND = SCRIPT_DIR / "run_student_loop.py"
|
|
COVERAGE_UNIVERSE_VERSION = "student-loop-strict.v2"
|
|
|
|
|
|
def load_json(path: Path) -> dict[str, Any]:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict):
|
|
raise AssertionError(f"{path.name} must contain a JSON object")
|
|
return value
|
|
|
|
|
|
def validate_matrix_policy(policy: dict[str, Any]) -> None:
|
|
required_keys = {
|
|
"schemaVersion",
|
|
"coverageUniverseVersion",
|
|
"requiredCleanCycles",
|
|
"maxMatrixCycles",
|
|
"maxRepairIterations",
|
|
"maxRuntimeMinutes",
|
|
"stopOnHighRisk",
|
|
"allowedArtifactRoots",
|
|
"coverageAdequacy",
|
|
"scenarios",
|
|
}
|
|
missing = sorted(required_keys - set(policy))
|
|
if missing:
|
|
raise AssertionError(f"strict matrix policy missing keys: {missing}")
|
|
if policy["schemaVersion"] != 1:
|
|
raise AssertionError("strict matrix policy schemaVersion must be 1")
|
|
if policy["coverageUniverseVersion"] != COVERAGE_UNIVERSE_VERSION:
|
|
raise AssertionError("strict matrix coverageUniverseVersion is not recognized")
|
|
for integer_key in ("requiredCleanCycles", "maxMatrixCycles", "maxRepairIterations", "maxRuntimeMinutes"):
|
|
if not isinstance(policy.get(integer_key), int) or int(policy[integer_key]) < 1:
|
|
raise AssertionError(f"strict matrix policy {integer_key} must be a positive integer")
|
|
if int(policy["maxMatrixCycles"]) < int(policy["requiredCleanCycles"]):
|
|
raise AssertionError("maxMatrixCycles must be greater than or equal to requiredCleanCycles")
|
|
if int(policy["requiredCleanCycles"]) < 2:
|
|
raise AssertionError("requiredCleanCycles must be at least 2 for best-boundary closure")
|
|
if not isinstance(policy["scenarios"], list) or not policy["scenarios"]:
|
|
raise AssertionError("strict matrix policy must define at least one scenario")
|
|
allowed_artifact_roots = policy.get("allowedArtifactRoots")
|
|
if allowed_artifact_roots != [".app/student-loop/runs", ".app/student-loop/matrix-runs"]:
|
|
raise AssertionError("strict matrix policy allowedArtifactRoots must be limited to student-loop artifact roots")
|
|
|
|
seen: set[str] = set()
|
|
required_scenario_fields = {
|
|
"id",
|
|
"required",
|
|
"status",
|
|
"learnerState",
|
|
"viewport",
|
|
"networkCondition",
|
|
"dataCondition",
|
|
"faultMode",
|
|
"expectedOutcome",
|
|
"requiredEvidence",
|
|
"adequacyAxisCoverage",
|
|
"env",
|
|
}
|
|
for index, scenario in enumerate(policy["scenarios"]):
|
|
if not isinstance(scenario, dict):
|
|
raise AssertionError(f"strict matrix scenario {index} must be an object")
|
|
missing_fields = sorted(required_scenario_fields - set(scenario))
|
|
if missing_fields:
|
|
raise AssertionError(f"strict matrix scenario {index} missing fields: {missing_fields}")
|
|
scenario_id = str(scenario["id"]).strip()
|
|
if not scenario_id:
|
|
raise AssertionError(f"strict matrix scenario {index} has an empty id")
|
|
if scenario_id in seen:
|
|
raise AssertionError(f"strict matrix policy has duplicate scenario id: {scenario_id}")
|
|
seen.add(scenario_id)
|
|
if scenario.get("required") is not True:
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} must be required")
|
|
if scenario.get("status") != "required":
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} must use status=required, not {scenario.get('status')}")
|
|
if str(scenario.get("status")) == "planned":
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} cannot be planned")
|
|
for text_field in ("learnerState", "viewport", "networkCondition", "dataCondition", "faultMode", "expectedOutcome"):
|
|
if not str(scenario.get(text_field, "")).strip():
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} must define {text_field}")
|
|
required_evidence = scenario["requiredEvidence"]
|
|
if not isinstance(required_evidence, list) or not required_evidence or any(not str(item).strip() for item in required_evidence):
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} must include nonempty requiredEvidence")
|
|
axis_coverage = scenario["adequacyAxisCoverage"]
|
|
if not isinstance(axis_coverage, dict) or not axis_coverage:
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} must include adequacyAxisCoverage")
|
|
for axis_id, values in axis_coverage.items():
|
|
if not isinstance(axis_id, str) or not axis_id.strip():
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} has an invalid adequacy axis id")
|
|
if not isinstance(values, list) or not values or any(not isinstance(value, str) or not value.strip() for value in values):
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} adequacy axis {axis_id} must list nonempty string values")
|
|
env = scenario["env"]
|
|
if not isinstance(env, dict):
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} env must be an object")
|
|
for key, value in env.items():
|
|
if not isinstance(key, str) or not isinstance(value, str):
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} env must map strings to strings")
|
|
scenario_mode = env.get("STUDENT_LOOP_SCENARIO_MODE", "full")
|
|
if scenario_mode not in {"full", "probe_only"}:
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} has invalid STUDENT_LOOP_SCENARIO_MODE: {scenario_mode}")
|
|
if scenario_id == "default_full_journey" and scenario_mode != "full":
|
|
raise AssertionError("default_full_journey must use full scenario mode")
|
|
if scenario_id != "default_full_journey" and scenario_mode != "probe_only":
|
|
raise AssertionError(f"strict matrix synthetic scenario {scenario_id} must use probe_only mode")
|
|
validate_coverage_adequacy(policy)
|
|
|
|
|
|
def validate_coverage_adequacy(policy: dict[str, Any]) -> None:
|
|
adequacy = policy.get("coverageAdequacy")
|
|
if not isinstance(adequacy, dict):
|
|
raise AssertionError("strict matrix policy coverageAdequacy must be an object")
|
|
required_keys = {
|
|
"schemaVersion",
|
|
"minimumCleanCycles",
|
|
"closureClaimTemplate",
|
|
"residualRiskStatement",
|
|
"scenarioAxes",
|
|
"highRiskPairings",
|
|
"exclusions",
|
|
}
|
|
missing = sorted(required_keys - set(adequacy))
|
|
if missing:
|
|
raise AssertionError(f"coverage adequacy missing keys: {missing}")
|
|
if adequacy["schemaVersion"] != 1:
|
|
raise AssertionError("coverage adequacy schemaVersion must be 1")
|
|
minimum_clean_cycles = adequacy["minimumCleanCycles"]
|
|
if not isinstance(minimum_clean_cycles, int) or minimum_clean_cycles < 2:
|
|
raise AssertionError("coverage adequacy minimumCleanCycles must be at least 2")
|
|
if int(policy["requiredCleanCycles"]) < minimum_clean_cycles:
|
|
raise AssertionError("requiredCleanCycles must satisfy coverage adequacy minimumCleanCycles")
|
|
claim_template = str(adequacy.get("closureClaimTemplate", ""))
|
|
if "inside the declared coverage universe" not in claim_template:
|
|
raise AssertionError("coverage adequacy closureClaimTemplate must use bounded coverage-universe wording")
|
|
if any(term in claim_template.lower() for term in ("no bugs exist", "absolute absence", "outside that universe is clean")):
|
|
raise AssertionError("coverage adequacy closureClaimTemplate must not claim absolute bug absence")
|
|
if not str(adequacy.get("residualRiskStatement", "")).strip():
|
|
raise AssertionError("coverage adequacy residualRiskStatement must be nonempty")
|
|
|
|
axes = adequacy["scenarioAxes"]
|
|
if not isinstance(axes, list) or not axes:
|
|
raise AssertionError("coverage adequacy scenarioAxes must be a nonempty list")
|
|
axis_values: dict[str, set[str]] = {}
|
|
for axis in axes:
|
|
if not isinstance(axis, dict):
|
|
raise AssertionError("coverage adequacy scenarioAxes entries must be objects")
|
|
axis_id = str(axis.get("id", "")).strip()
|
|
values = axis.get("requiredValues")
|
|
if not axis_id:
|
|
raise AssertionError("coverage adequacy axis id must be nonempty")
|
|
if axis_id in axis_values:
|
|
raise AssertionError(f"coverage adequacy has duplicate axis id: {axis_id}")
|
|
if not isinstance(values, list) or not values:
|
|
raise AssertionError(f"coverage adequacy axis {axis_id} must list requiredValues")
|
|
normalized_values: set[str] = set()
|
|
for value in values:
|
|
value_id = str(value).strip()
|
|
if not value_id:
|
|
raise AssertionError(f"coverage adequacy axis {axis_id} has an empty value")
|
|
if value_id in normalized_values:
|
|
raise AssertionError(f"coverage adequacy axis {axis_id} has duplicate value: {value_id}")
|
|
normalized_values.add(value_id)
|
|
axis_values[axis_id] = normalized_values
|
|
|
|
scenarios = policy["scenarios"]
|
|
scenario_ids = {str(scenario["id"]) for scenario in scenarios}
|
|
covered: dict[str, set[str]] = {axis: set() for axis in axis_values}
|
|
for scenario in scenarios:
|
|
scenario_id = str(scenario["id"])
|
|
for axis_id, values in scenario["adequacyAxisCoverage"].items():
|
|
if axis_id not in axis_values:
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} references unknown adequacy axis: {axis_id}")
|
|
for value in values:
|
|
value_id = str(value)
|
|
if value_id not in axis_values[axis_id]:
|
|
raise AssertionError(f"strict matrix scenario {scenario_id} references unknown adequacy value: {axis_id}={value_id}")
|
|
covered[axis_id].add(value_id)
|
|
|
|
exclusions = adequacy["exclusions"]
|
|
if not isinstance(exclusions, list):
|
|
raise AssertionError("coverage adequacy exclusions must be a list")
|
|
excluded: dict[str, set[str]] = {axis: set() for axis in axis_values}
|
|
exclusion_ids: set[str] = set()
|
|
for exclusion in exclusions:
|
|
if not isinstance(exclusion, dict):
|
|
raise AssertionError("coverage adequacy exclusions must contain objects")
|
|
required = {"id", "axis", "value", "rationale", "proof", "owner", "residualRisk"}
|
|
missing_exclusion = sorted(required - set(exclusion))
|
|
if missing_exclusion:
|
|
raise AssertionError(f"coverage adequacy exclusion missing keys: {missing_exclusion}")
|
|
exclusion_id = str(exclusion["id"]).strip()
|
|
axis_id = str(exclusion["axis"]).strip()
|
|
value_id = str(exclusion["value"]).strip()
|
|
if not exclusion_id or exclusion_id in exclusion_ids:
|
|
raise AssertionError(f"coverage adequacy exclusion has invalid or duplicate id: {exclusion_id}")
|
|
exclusion_ids.add(exclusion_id)
|
|
if axis_id not in axis_values or value_id not in axis_values[axis_id]:
|
|
raise AssertionError(f"coverage adequacy exclusion references unknown axis value: {axis_id}={value_id}")
|
|
for text_key in ("rationale", "proof", "owner", "residualRisk"):
|
|
if not str(exclusion.get(text_key, "")).strip():
|
|
raise AssertionError(f"coverage adequacy exclusion {exclusion_id} must include {text_key}")
|
|
excluded[axis_id].add(value_id)
|
|
|
|
missing_axis_values = {
|
|
axis_id: sorted(values - covered[axis_id] - excluded[axis_id])
|
|
for axis_id, values in axis_values.items()
|
|
if values - covered[axis_id] - excluded[axis_id]
|
|
}
|
|
if missing_axis_values:
|
|
raise AssertionError(f"coverage adequacy missing axis coverage: {missing_axis_values}")
|
|
|
|
pairings = adequacy["highRiskPairings"]
|
|
if not isinstance(pairings, list) or not pairings:
|
|
raise AssertionError("coverage adequacy highRiskPairings must be a nonempty list")
|
|
pairing_ids: set[str] = set()
|
|
for pairing in pairings:
|
|
if not isinstance(pairing, dict):
|
|
raise AssertionError("coverage adequacy highRiskPairings must contain objects")
|
|
required = {"id", "required", "axisValues", "scenarioIds"}
|
|
missing_pairing = sorted(required - set(pairing))
|
|
if missing_pairing:
|
|
raise AssertionError(f"coverage adequacy high-risk pairing missing keys: {missing_pairing}")
|
|
pairing_id = str(pairing["id"]).strip()
|
|
if not pairing_id or pairing_id in pairing_ids:
|
|
raise AssertionError(f"coverage adequacy high-risk pairing has invalid or duplicate id: {pairing_id}")
|
|
pairing_ids.add(pairing_id)
|
|
if pairing.get("required") is not True:
|
|
raise AssertionError(f"coverage adequacy high-risk pairing {pairing_id} must be required")
|
|
axis_value_map = pairing["axisValues"]
|
|
if not isinstance(axis_value_map, dict) or not axis_value_map:
|
|
raise AssertionError(f"coverage adequacy high-risk pairing {pairing_id} must define axisValues")
|
|
for axis_id, value in axis_value_map.items():
|
|
value_id = str(value)
|
|
if axis_id not in axis_values or value_id not in axis_values[axis_id]:
|
|
raise AssertionError(f"coverage adequacy high-risk pairing {pairing_id} references unknown axis value: {axis_id}={value_id}")
|
|
pairing_scenarios = pairing["scenarioIds"]
|
|
if not isinstance(pairing_scenarios, list) or not pairing_scenarios:
|
|
raise AssertionError(f"coverage adequacy high-risk pairing {pairing_id} must list scenarioIds")
|
|
for scenario_id in pairing_scenarios:
|
|
if str(scenario_id) not in scenario_ids:
|
|
raise AssertionError(f"coverage adequacy high-risk pairing {pairing_id} references unknown scenario: {scenario_id}")
|
|
|
|
|
|
def validate_default_policy_for_strict(default_policy: dict[str, Any], matrix_policy: dict[str, Any]) -> None:
|
|
coverage = default_policy.get("scenarioCoverage")
|
|
if not isinstance(coverage, dict):
|
|
raise AssertionError("default student loop policy must define scenarioCoverage")
|
|
required = {
|
|
str(evidence)
|
|
for scenario in matrix_policy["scenarios"]
|
|
for evidence in scenario.get("requiredEvidence", [])
|
|
}
|
|
missing = sorted(required - set(coverage))
|
|
if missing:
|
|
raise AssertionError(f"default student loop policy coverage missing strict evidence keys: {missing}")
|
|
|
|
|
|
def utc_timestamp_id() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
|
|
|
|
def allocate_timestamp_id(root: Path) -> str:
|
|
while True:
|
|
candidate = utc_timestamp_id()
|
|
if not (root / candidate).exists():
|
|
return candidate
|
|
time.sleep(1.05)
|
|
|
|
|
|
def run_matrix(*, selected_scenario_id: str | None = None) -> dict[str, Any]:
|
|
matrix_policy = load_json(MATRIX_POLICY_PATH)
|
|
default_policy = load_json(DEFAULT_POLICY_PATH)
|
|
validate_matrix_policy(matrix_policy)
|
|
validate_default_policy_for_strict(default_policy, matrix_policy)
|
|
|
|
matrix_run_id = allocate_timestamp_id(MATRIX_RUNS_ROOT)
|
|
matrix_run_root = MATRIX_RUNS_ROOT / matrix_run_id
|
|
matrix_run_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
scenarios = matrix_policy["scenarios"]
|
|
if selected_scenario_id:
|
|
scenarios = [scenario for scenario in scenarios if scenario["id"] == selected_scenario_id]
|
|
if not scenarios:
|
|
raise AssertionError(f"unknown strict matrix scenario: {selected_scenario_id}")
|
|
|
|
scenario_results = []
|
|
for scenario in scenarios:
|
|
scenario_results.append(run_scenario(matrix_policy, matrix_run_root, scenario))
|
|
|
|
summary = build_matrix_summary(matrix_policy, matrix_run_id, scenario_results, selected_scenario_id)
|
|
summary_path = matrix_run_root / "summary.json"
|
|
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(f"strict matrix summary: {summary_path}")
|
|
return summary
|
|
|
|
|
|
def run_scenario(matrix_policy: dict[str, Any], matrix_run_root: Path, scenario: dict[str, Any]) -> dict[str, Any]:
|
|
scenario_id = str(scenario["id"])
|
|
run_id = allocate_timestamp_id(RUNS_ROOT)
|
|
run_root = RUNS_ROOT / run_id
|
|
scenario_log = matrix_run_root / f"{scenario_id}.log"
|
|
|
|
env = {
|
|
**os.environ,
|
|
**{str(key): str(value) for key, value in scenario.get("env", {}).items()},
|
|
"STUDENT_LOOP_RUN_ID": run_id,
|
|
"STUDENT_LOOP_STRICT_SCENARIO_ID": scenario_id,
|
|
"STUDENT_LOOP_COVERAGE_UNIVERSE_VERSION": str(matrix_policy["coverageUniverseVersion"]),
|
|
}
|
|
completed = subprocess.run(
|
|
[sys.executable, str(STUDENT_LOOP_COMMAND)],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
)
|
|
scenario_log.write_text(completed.stdout[-20000:], encoding="utf-8")
|
|
|
|
report_path = run_root / "artifacts" / "report.json"
|
|
report = load_report_if_present(report_path)
|
|
result = evaluate_scenario_report(
|
|
scenario=scenario,
|
|
run_id=run_id,
|
|
run_root=run_root,
|
|
report_path=report_path,
|
|
report=report,
|
|
return_code=completed.returncode,
|
|
)
|
|
result["log_path"] = repo_relative_identity(scenario_log)
|
|
return result
|
|
|
|
|
|
def load_report_if_present(report_path: Path) -> dict[str, Any] | None:
|
|
if not report_path.exists():
|
|
return None
|
|
value = json.loads(report_path.read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict):
|
|
raise AssertionError(f"student loop report must be an object: {report_path}")
|
|
return value
|
|
|
|
|
|
def evaluate_scenario_report(
|
|
*,
|
|
scenario: dict[str, Any],
|
|
run_id: str,
|
|
run_root: Path,
|
|
report_path: Path,
|
|
report: dict[str, Any] | None,
|
|
return_code: int,
|
|
) -> dict[str, Any]:
|
|
report_path_valid = report_path.resolve() == (run_root / "artifacts" / "report.json").resolve()
|
|
report_path_under_runs = is_subpath(report_path, RUNS_ROOT)
|
|
run_id_matches = bool(report and report.get("run", {}).get("run_id") == run_id)
|
|
required_evidence = {}
|
|
missing_required_evidence = []
|
|
failed_findings = []
|
|
failed_steps = []
|
|
forbidden_terms_found: list[str] = []
|
|
|
|
if report:
|
|
coverage = report.get("scenario_coverage", {})
|
|
for evidence_id in scenario["requiredEvidence"]:
|
|
evidence_status = coverage.get(evidence_id, {}).get("status") if isinstance(coverage, dict) else None
|
|
required_evidence[evidence_id] = evidence_status
|
|
if evidence_status != "executed":
|
|
missing_required_evidence.append(evidence_id)
|
|
failed_findings = [
|
|
finding
|
|
for finding in report.get("findings", [])
|
|
if isinstance(finding, dict) and finding.get("result") == "fail"
|
|
]
|
|
failed_steps = [
|
|
step
|
|
for step in report.get("steps", [])
|
|
if isinstance(step, dict) and step.get("status") == "failed"
|
|
]
|
|
report_text = json.dumps(report, ensure_ascii=False)
|
|
for term in load_json(DEFAULT_POLICY_PATH).get("forbiddenReportTerms", []):
|
|
if str(term).lower() in report_text.lower():
|
|
forbidden_terms_found.append(str(term))
|
|
else:
|
|
missing_required_evidence = list(scenario["requiredEvidence"])
|
|
|
|
failure_reasons = []
|
|
if return_code != 0:
|
|
failure_reasons.append(f"student loop exited {return_code}")
|
|
if not report:
|
|
failure_reasons.append("missing student loop report")
|
|
if not report_path_valid or not report_path_under_runs:
|
|
failure_reasons.append("invalid report path")
|
|
if not run_id_matches:
|
|
failure_reasons.append("report run id does not match scenario run directory")
|
|
if missing_required_evidence:
|
|
failure_reasons.append("missing required evidence")
|
|
if failed_findings:
|
|
failure_reasons.append("failing oracle findings")
|
|
if failed_steps:
|
|
failure_reasons.append("failed student loop steps")
|
|
if forbidden_terms_found:
|
|
failure_reasons.append("forbidden terms found")
|
|
|
|
return {
|
|
"id": scenario["id"],
|
|
"status": "passed" if not failure_reasons else "failed",
|
|
"return_code": return_code,
|
|
"run_id": run_id,
|
|
"report_path": repo_relative_identity(report_path),
|
|
"report_path_valid": report_path_valid,
|
|
"report_path_under_runs": report_path_under_runs,
|
|
"run_id_matches": run_id_matches,
|
|
"required_evidence": required_evidence,
|
|
"adequacy_axis_coverage": scenario.get("adequacyAxisCoverage", {}),
|
|
"missing_required_evidence": missing_required_evidence,
|
|
"failed_findings_count": len(failed_findings),
|
|
"failure_categories": sorted({str(finding.get("category")) for finding in failed_findings}),
|
|
"failed_steps_count": len(failed_steps),
|
|
"forbidden_terms_found": sorted(set(forbidden_terms_found)),
|
|
"failure_reasons": failure_reasons,
|
|
}
|
|
|
|
|
|
def build_coverage_adequacy_summary(
|
|
matrix_policy: dict[str, Any],
|
|
scenario_results: list[dict[str, Any]],
|
|
*,
|
|
selected_scenario_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
adequacy = matrix_policy["coverageAdequacy"]
|
|
axes = adequacy["scenarioAxes"]
|
|
axis_required_values = {
|
|
str(axis["id"]): {str(value) for value in axis["requiredValues"]}
|
|
for axis in axes
|
|
}
|
|
passed_results = {
|
|
str(result["id"]): result
|
|
for result in scenario_results
|
|
if result.get("status") == "passed"
|
|
}
|
|
policy_scenarios = {str(scenario["id"]): scenario for scenario in matrix_policy["scenarios"]}
|
|
covered: dict[str, dict[str, list[str]]] = {
|
|
axis_id: {value: [] for value in sorted(values)}
|
|
for axis_id, values in axis_required_values.items()
|
|
}
|
|
|
|
for scenario_id in passed_results:
|
|
scenario = policy_scenarios.get(scenario_id)
|
|
if not scenario:
|
|
continue
|
|
for axis_id, values in scenario.get("adequacyAxisCoverage", {}).items():
|
|
for value in values:
|
|
if axis_id in covered and value in covered[axis_id]:
|
|
covered[axis_id][value].append(scenario_id)
|
|
|
|
exclusions = adequacy.get("exclusions", [])
|
|
excluded_values: dict[str, dict[str, list[dict[str, Any]]]] = {
|
|
axis_id: {}
|
|
for axis_id in axis_required_values
|
|
}
|
|
unproved_exclusions = []
|
|
for exclusion in exclusions:
|
|
axis_id = str(exclusion.get("axis", ""))
|
|
value = str(exclusion.get("value", ""))
|
|
proof = str(exclusion.get("proof", "")).strip()
|
|
if axis_id in excluded_values:
|
|
excluded_values[axis_id].setdefault(value, []).append(exclusion)
|
|
if not proof:
|
|
unproved_exclusions.append(exclusion.get("id"))
|
|
|
|
missing_axis_values = {}
|
|
if not selected_scenario_id:
|
|
for axis_id, values in axis_required_values.items():
|
|
missing_values = []
|
|
for value in sorted(values):
|
|
if covered[axis_id][value] or excluded_values.get(axis_id, {}).get(value):
|
|
continue
|
|
missing_values.append(value)
|
|
if missing_values:
|
|
missing_axis_values[axis_id] = missing_values
|
|
|
|
high_risk_pairings = []
|
|
missing_high_risk_pairings = []
|
|
for pairing in adequacy.get("highRiskPairings", []):
|
|
scenario_ids = [str(item) for item in pairing.get("scenarioIds", [])]
|
|
covered_by = [scenario_id for scenario_id in scenario_ids if scenario_id in passed_results]
|
|
status = "covered" if covered_by else "missing"
|
|
if selected_scenario_id and selected_scenario_id not in scenario_ids:
|
|
status = "not_evaluated_for_selected_scenario"
|
|
elif status == "missing":
|
|
missing_high_risk_pairings.append(str(pairing.get("id")))
|
|
high_risk_pairings.append({
|
|
"id": pairing.get("id"),
|
|
"status": status,
|
|
"axis_values": pairing.get("axisValues", {}),
|
|
"scenario_ids": scenario_ids,
|
|
"covered_by": covered_by,
|
|
})
|
|
|
|
failures = []
|
|
if missing_axis_values:
|
|
failures.append("missing required adequacy axis values")
|
|
if missing_high_risk_pairings:
|
|
failures.append("missing required high-risk pairings")
|
|
if unproved_exclusions:
|
|
failures.append("unproved coverage exclusions")
|
|
|
|
return {
|
|
"schema_version": "student_loop_coverage_adequacy.v1",
|
|
"coverage_universe_version": matrix_policy["coverageUniverseVersion"],
|
|
"scope": "selected_scenario" if selected_scenario_id else "full_matrix",
|
|
"covered_axis_values": covered,
|
|
"missing_axis_values": missing_axis_values,
|
|
"excluded_axis_values": excluded_values,
|
|
"high_risk_pairings": high_risk_pairings,
|
|
"missing_high_risk_pairings": missing_high_risk_pairings,
|
|
"unproved_exclusions": unproved_exclusions,
|
|
"closure_claim_template": adequacy["closureClaimTemplate"],
|
|
"residual_risk_statement": adequacy["residualRiskStatement"],
|
|
"failures": failures,
|
|
}
|
|
|
|
|
|
def build_matrix_summary(
|
|
matrix_policy: dict[str, Any],
|
|
matrix_run_id: str,
|
|
scenario_results: list[dict[str, Any]],
|
|
selected_scenario_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
required_ids = {str(scenario["id"]) for scenario in matrix_policy["scenarios"] if scenario.get("required") is True}
|
|
executed_ids = {str(result["id"]) for result in scenario_results}
|
|
missing_required_scenarios = [] if selected_scenario_id else sorted(required_ids - executed_ids)
|
|
failed_results = [result for result in scenario_results if result["status"] != "passed"]
|
|
planned_required_scenarios = [
|
|
str(scenario["id"])
|
|
for scenario in matrix_policy["scenarios"]
|
|
if scenario.get("required") is True and scenario.get("status") == "planned"
|
|
]
|
|
forbidden_terms = sorted({
|
|
term
|
|
for result in scenario_results
|
|
for term in result.get("forbidden_terms_found", [])
|
|
})
|
|
failed_findings_count = sum(int(result.get("failed_findings_count", 0)) for result in scenario_results)
|
|
failed_steps_count = sum(int(result.get("failed_steps_count", 0)) for result in scenario_results)
|
|
coverage_adequacy = build_coverage_adequacy_summary(
|
|
matrix_policy,
|
|
scenario_results,
|
|
selected_scenario_id=selected_scenario_id,
|
|
)
|
|
summary_failures = []
|
|
if failed_results:
|
|
summary_failures.append("one or more strict scenarios failed")
|
|
if missing_required_scenarios:
|
|
summary_failures.append("required strict scenarios were not executed")
|
|
if planned_required_scenarios:
|
|
summary_failures.append("required strict scenarios cannot be planned")
|
|
if forbidden_terms:
|
|
summary_failures.append("forbidden evidence terms were found")
|
|
if coverage_adequacy["failures"]:
|
|
summary_failures.append("coverage adequacy failures were found")
|
|
|
|
scenario_verification_eligible = not summary_failures and failed_findings_count == 0 and failed_steps_count == 0
|
|
closure_eligible = bool(scenario_verification_eligible and not selected_scenario_id)
|
|
stop_condition = "coverage_closure_achieved" if closure_eligible else "matrix_failed"
|
|
if selected_scenario_id and scenario_verification_eligible:
|
|
stop_condition = "selected_scenario_verified"
|
|
return {
|
|
"schema_version": "student_loop_matrix_summary.v1",
|
|
"matrix_run_id": matrix_run_id,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"coverage_universe_version": matrix_policy["coverageUniverseVersion"],
|
|
"selected_scenario_id": selected_scenario_id,
|
|
"scenario_count": len(scenario_results),
|
|
"required_scenario_count": len(required_ids) if not selected_scenario_id else len(scenario_results),
|
|
"scenario_results": scenario_results,
|
|
"missing_required_scenarios": missing_required_scenarios,
|
|
"planned_required_scenarios": planned_required_scenarios,
|
|
"failed_findings_count": failed_findings_count,
|
|
"failed_steps_count": failed_steps_count,
|
|
"forbidden_terms_found": forbidden_terms,
|
|
"coverage_adequacy": coverage_adequacy,
|
|
"summary_failures": summary_failures,
|
|
"scenario_verification_eligible": scenario_verification_eligible,
|
|
"closure_eligible": closure_eligible,
|
|
"stop_condition": stop_condition,
|
|
}
|
|
|
|
|
|
def run_until_clean() -> dict[str, Any]:
|
|
matrix_policy = load_json(MATRIX_POLICY_PATH)
|
|
validate_matrix_policy(matrix_policy)
|
|
required_clean_cycles = int(matrix_policy["requiredCleanCycles"])
|
|
max_matrix_cycles = int(matrix_policy["maxMatrixCycles"])
|
|
clean_cycles = 0
|
|
cycle_summaries = []
|
|
|
|
for cycle_index in range(1, max_matrix_cycles + 1):
|
|
summary = run_matrix()
|
|
cycle_summaries.append({
|
|
"cycle_index": cycle_index,
|
|
"matrix_run_id": summary["matrix_run_id"],
|
|
"closure_eligible": summary["closure_eligible"],
|
|
"stop_condition": summary["stop_condition"],
|
|
"summary_failures": summary["summary_failures"],
|
|
"coverage_adequacy": summary["coverage_adequacy"],
|
|
})
|
|
if summary["closure_eligible"]:
|
|
clean_cycles += 1
|
|
if clean_cycles >= required_clean_cycles:
|
|
return {
|
|
"stop_condition": "required_clean_cycles_reached",
|
|
"required_clean_cycles": required_clean_cycles,
|
|
"max_matrix_cycles": max_matrix_cycles,
|
|
"max_repair_iterations": matrix_policy["maxRepairIterations"],
|
|
"stop_on_high_risk": matrix_policy["stopOnHighRisk"],
|
|
"clean_cycles": clean_cycles,
|
|
"latest_matrix_summary": summary["matrix_run_id"],
|
|
"coverage_adequacy": summary["coverage_adequacy"],
|
|
"failed_scenario_reports": [],
|
|
"repair_ledger_entries": ".app/student-loop/repair-ledger.jsonl",
|
|
"remaining_risk": summary["coverage_adequacy"]["residual_risk_statement"],
|
|
"cycle_summaries": cycle_summaries,
|
|
}
|
|
else:
|
|
clean_cycles = 0
|
|
return {
|
|
"stop_condition": "matrix_failed",
|
|
"required_clean_cycles": required_clean_cycles,
|
|
"max_matrix_cycles": max_matrix_cycles,
|
|
"max_repair_iterations": matrix_policy["maxRepairIterations"],
|
|
"stop_on_high_risk": matrix_policy["stopOnHighRisk"],
|
|
"clean_cycles": clean_cycles,
|
|
"latest_matrix_summary": summary["matrix_run_id"],
|
|
"coverage_adequacy": summary["coverage_adequacy"],
|
|
"failed_scenario_reports": [
|
|
result["report_path"]
|
|
for result in summary["scenario_results"]
|
|
if result["status"] != "passed"
|
|
],
|
|
"repair_ledger_entries": ".app/student-loop/repair-ledger.jsonl",
|
|
"remaining_risk": summary["coverage_adequacy"]["residual_risk_statement"],
|
|
"cycle_summaries": cycle_summaries,
|
|
}
|
|
|
|
return {
|
|
"stop_condition": "max_matrix_cycles_reached",
|
|
"required_clean_cycles": required_clean_cycles,
|
|
"max_matrix_cycles": max_matrix_cycles,
|
|
"max_repair_iterations": matrix_policy["maxRepairIterations"],
|
|
"stop_on_high_risk": matrix_policy["stopOnHighRisk"],
|
|
"clean_cycles": clean_cycles,
|
|
"latest_matrix_summary": cycle_summaries[-1]["matrix_run_id"] if cycle_summaries else None,
|
|
"coverage_adequacy": cycle_summaries[-1]["coverage_adequacy"] if cycle_summaries else None,
|
|
"failed_scenario_reports": [],
|
|
"repair_ledger_entries": ".app/student-loop/repair-ledger.jsonl",
|
|
"remaining_risk": "required clean cycle count was not reached",
|
|
"cycle_summaries": cycle_summaries,
|
|
}
|
|
|
|
|
|
def write_until_clean_summary(summary: dict[str, Any]) -> Path:
|
|
run_id = allocate_timestamp_id(MATRIX_RUNS_ROOT)
|
|
run_root = MATRIX_RUNS_ROOT / run_id
|
|
run_root.mkdir(parents=True, exist_ok=True)
|
|
path = run_root / "until-clean-summary.json"
|
|
path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def repo_relative_identity(path: Path) -> dict[str, Any]:
|
|
resolved = path.resolve()
|
|
try:
|
|
repo_relative = resolved.relative_to(REPO_ROOT.resolve()).as_posix()
|
|
except ValueError:
|
|
repo_relative = None
|
|
return {
|
|
"name": resolved.name,
|
|
"parent_name": resolved.parent.name,
|
|
"repo_relative": repo_relative,
|
|
"exists": resolved.exists(),
|
|
}
|
|
|
|
|
|
def is_subpath(path: Path, base: Path) -> bool:
|
|
try:
|
|
path.resolve().relative_to(base.resolve())
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def run_self_test() -> None:
|
|
matrix_policy = load_json(MATRIX_POLICY_PATH)
|
|
default_policy = load_json(DEFAULT_POLICY_PATH)
|
|
validate_matrix_policy(matrix_policy)
|
|
validate_default_policy_for_strict(default_policy, matrix_policy)
|
|
|
|
duplicate_policy = json.loads(json.dumps(matrix_policy))
|
|
duplicate_policy["scenarios"][1]["id"] = duplicate_policy["scenarios"][0]["id"]
|
|
assert_policy_fails(duplicate_policy, "duplicate")
|
|
|
|
planned_policy = json.loads(json.dumps(matrix_policy))
|
|
planned_policy["scenarios"][0]["status"] = "planned"
|
|
assert_policy_fails(planned_policy, "planned")
|
|
|
|
missing_metadata_policy = json.loads(json.dumps(matrix_policy))
|
|
del missing_metadata_policy["scenarios"][0]["learnerState"]
|
|
assert_policy_fails(missing_metadata_policy, "missing")
|
|
|
|
missing_adequacy_policy = json.loads(json.dumps(matrix_policy))
|
|
del missing_adequacy_policy["coverageAdequacy"]
|
|
assert_policy_fails(missing_adequacy_policy, "coverageAdequacy")
|
|
|
|
duplicate_axis_policy = json.loads(json.dumps(matrix_policy))
|
|
duplicate_axis_policy["coverageAdequacy"]["scenarioAxes"][1]["id"] = duplicate_axis_policy["coverageAdequacy"]["scenarioAxes"][0]["id"]
|
|
assert_policy_fails(duplicate_axis_policy, "duplicate axis")
|
|
|
|
missing_axis_policy = json.loads(json.dumps(matrix_policy))
|
|
del missing_axis_policy["scenarios"][0]["adequacyAxisCoverage"]["learner_progress_state"]
|
|
assert_policy_fails(missing_axis_policy, "missing axis coverage")
|
|
|
|
missing_pairing_policy = json.loads(json.dumps(matrix_policy))
|
|
missing_pairing_policy["coverageAdequacy"]["highRiskPairings"][0]["scenarioIds"] = ["missing_scenario"]
|
|
assert_policy_fails(missing_pairing_policy, "unknown scenario")
|
|
|
|
invalid_exclusion_policy = json.loads(json.dumps(matrix_policy))
|
|
invalid_exclusion_policy["coverageAdequacy"]["exclusions"] = [{
|
|
"id": "missing-proof",
|
|
"axis": "learner_progress_state",
|
|
"value": "fresh",
|
|
"rationale": "self test",
|
|
"owner": "self test",
|
|
"residualRisk": "self test",
|
|
}]
|
|
assert_policy_fails(invalid_exclusion_policy, "exclusion missing")
|
|
|
|
invalid_clean_policy = json.loads(json.dumps(matrix_policy))
|
|
invalid_clean_policy["requiredCleanCycles"] = 1
|
|
assert_policy_fails(invalid_clean_policy, "at least 2")
|
|
|
|
invalid_mode_policy = json.loads(json.dumps(matrix_policy))
|
|
invalid_mode_policy["scenarios"][1]["env"]["STUDENT_LOOP_SCENARIO_MODE"] = "full"
|
|
assert_policy_fails(invalid_mode_policy, "probe_only mode")
|
|
print("strict matrix policy self-test passed")
|
|
print("coverage adequacy self-test passed")
|
|
|
|
scenario = matrix_policy["scenarios"][0]
|
|
fake_run_id = "20260516T120000Z"
|
|
fake_report = {
|
|
"run": {"run_id": fake_run_id},
|
|
"scenario_coverage": {
|
|
evidence: {"status": "executed", "reason": "self test"}
|
|
for evidence in scenario["requiredEvidence"]
|
|
},
|
|
"findings": [],
|
|
"steps": [{"name": "open_app", "status": "passed"}],
|
|
}
|
|
fake_result = evaluate_scenario_report(
|
|
scenario=scenario,
|
|
run_id=fake_run_id,
|
|
run_root=RUNS_ROOT / fake_run_id,
|
|
report_path=RUNS_ROOT / fake_run_id / "artifacts" / "report.json",
|
|
report=fake_report,
|
|
return_code=0,
|
|
)
|
|
if fake_result["status"] != "passed":
|
|
raise AssertionError(f"strict matrix summary self-test expected passed scenario: {fake_result}")
|
|
missing_evidence_report = json.loads(json.dumps(fake_report))
|
|
missing_evidence_report["scenario_coverage"][scenario["requiredEvidence"][0]]["status"] = "planned"
|
|
missing_result = evaluate_scenario_report(
|
|
scenario=scenario,
|
|
run_id=fake_run_id,
|
|
run_root=RUNS_ROOT / fake_run_id,
|
|
report_path=RUNS_ROOT / fake_run_id / "artifacts" / "report.json",
|
|
report=missing_evidence_report,
|
|
return_code=0,
|
|
)
|
|
if missing_result["status"] != "failed" or not missing_result["missing_required_evidence"]:
|
|
raise AssertionError("strict matrix summary self-test did not fail unexecuted required evidence")
|
|
summary = build_matrix_summary(matrix_policy, "20260516T120100Z", [
|
|
{
|
|
**fake_result,
|
|
"id": scenario["id"],
|
|
}
|
|
for scenario in matrix_policy["scenarios"]
|
|
])
|
|
if not summary["closure_eligible"]:
|
|
raise AssertionError("strict matrix summary self-test expected closure eligible summary")
|
|
selected_summary = build_matrix_summary(matrix_policy, "20260516T120200Z", [fake_result], selected_scenario_id=str(scenario["id"]))
|
|
if not selected_summary["scenario_verification_eligible"] or selected_summary["closure_eligible"]:
|
|
raise AssertionError("selected strict scenario self-test expected targeted verification without closure eligibility")
|
|
if selected_summary["stop_condition"] != "selected_scenario_verified":
|
|
raise AssertionError("selected strict scenario self-test expected selected_scenario_verified stop condition")
|
|
print("strict matrix summary self-test passed")
|
|
|
|
if int(matrix_policy["requiredCleanCycles"]) < 2 or int(matrix_policy["maxMatrixCycles"]) < int(matrix_policy["requiredCleanCycles"]):
|
|
raise AssertionError("until-clean bounds are invalid")
|
|
print("until-clean bounds self-test passed")
|
|
|
|
|
|
def assert_policy_fails(policy: dict[str, Any], expected_text: str) -> None:
|
|
try:
|
|
validate_matrix_policy(policy)
|
|
except AssertionError as error:
|
|
if expected_text.lower() not in str(error).lower():
|
|
raise AssertionError(f"policy failed for the wrong reason: {error}") from error
|
|
return
|
|
raise AssertionError("invalid strict matrix policy unexpectedly passed")
|
|
|
|
|
|
def parse_args(argv: list[str]) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Run the strict student loop coverage matrix.")
|
|
parser.add_argument("--self-test-policy", action="store_true", help="Validate strict policy and summary contracts without launching a browser.")
|
|
parser.add_argument("--strict", action="store_true", help="Run the full strict coverage matrix once.")
|
|
parser.add_argument("--until-clean", action="store_true", help="Run strict matrix cycles until the configured clean cycle count is reached.")
|
|
parser.add_argument("--scenario", help="Run one strict scenario by id. Intended for targeted repair verification.")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv or sys.argv[1:])
|
|
if args.self_test_policy:
|
|
run_self_test()
|
|
return 0
|
|
if args.until_clean:
|
|
summary = run_until_clean()
|
|
path = write_until_clean_summary(summary)
|
|
print(f"until-clean summary: {path}")
|
|
return 0 if summary["stop_condition"] == "required_clean_cycles_reached" else 1
|
|
if args.strict or args.scenario:
|
|
summary = run_matrix(selected_scenario_id=args.scenario)
|
|
return 0 if summary["scenario_verification_eligible"] else 1
|
|
raise SystemExit("choose --strict, --until-clean, --scenario, or --self-test-policy")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|