File size: 3,630 Bytes
d06903c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | import json
from collections import Counter
from pathlib import Path
from nviq.audit.runner import run_audit, write_audit_artifacts
from nviq.execution.csc import run_csc_case
from nviq.execution.canonical_lanes import CANONICAL_LANE_FAMILIES, run_canonical_lane_case
from .evaluators import derive_observation
def derive_observations(cases: list[dict], evidence_records: list[dict]) -> list[dict]:
cases_by_id = {case["case_id"]: case for case in cases}
derived = []
for evidence in evidence_records:
case = cases_by_id[evidence["case_id"]]
family = case.get("family")
if family in CANONICAL_LANE_FAMILIES:
derived.append(run_canonical_lane_case(case, evidence))
elif family == "causal_speech_cognition" and "ledger" in evidence:
result = run_csc_case({"case_id": case["case_id"], "ledger": evidence["ledger"], "expected": case.get("expected", {})})
derived.append({
"case_id": case["case_id"], "track": evidence["track"],
"metrics": result["metrics"],
"catastrophic_failures": result["catastrophic_failures"],
"execution": {"status": "success", "failure_reason": None},
"trace": {
"source": "nviq_csc_engine_v0.1",
"failure_locus": result["failure_locus"],
"earliest_failure_time_ms": result["earliest_failure_time_ms"],
"belief_event_id": result["belief_event_id"],
"belief_time_ms": result["belief_time_ms"],
"decision_event_id": result["decision_event_id"],
"decision_time_ms": result["decision_time_ms"],
},
"evidence_refs": evidence.get("provenance", {}).get("evidence_refs", []),
"adapter": {"kind": "nviq_csc_engine", "provider": evidence.get("provenance", {}).get("provider"), "model": evidence.get("provenance", {}).get("model"), "configuration": evidence.get("provenance", {}).get("configuration", {})},
})
else:
derived.append(derive_observation(case, evidence))
return sorted(derived, key=lambda row: (row["case_id"], row["track"]))
def _execution_summary(cases: list[dict], evidence_records: list[dict], observations: list[dict]) -> dict:
case_map = {case["case_id"]: case for case in cases}
family_counts = Counter(case_map[row["case_id"]]["family"] for row in observations)
track_counts = Counter(row["track"] for row in observations)
return {"execution_version":"0.1","evidence_record_count":len(evidence_records),"derived_observation_count":len(observations),"tracks":dict(sorted(track_counts.items())),"families":dict(sorted(family_counts.items()))}
def execute_audit(manifest: dict, cases: list[dict], evidence_records: list[dict], remediation_registry: list[dict], output_dir: Path) -> dict:
observations = derive_observations(cases, evidence_records)
report = run_audit(manifest, cases, observations, remediation_registry)
paths = write_audit_artifacts(report, output_dir)
derived_path = output_dir / "derived-observations.json"
summary_path = output_dir / "evidence-summary.json"
derived_path.write_text(json.dumps(observations, indent=2, sort_keys=True) + "\n", encoding="utf-8")
summary_path.write_text(json.dumps(_execution_summary(cases, evidence_records, observations), indent=2, sort_keys=True) + "\n", encoding="utf-8")
return {"report":report,"observations":observations,"paths":{**paths,"derived_observations":derived_path,"evidence_summary":summary_path}}
|