diff --git a/src/or_audit/eval/cartesian.py b/src/or_audit/eval/cartesian.py index 7b890ae..930cd0b 100644 --- a/src/or_audit/eval/cartesian.py +++ b/src/or_audit/eval/cartesian.py @@ -347,6 +347,7 @@ def run_cartesian_job( n=pair_trials, gym_factory=gym_factory, split=task_split, + independent_case_unit=stage.independent_case_unit if stage is not None else None, ) pairs.append( PairRecord( diff --git a/src/or_audit/eval/job.py b/src/or_audit/eval/job.py index 00dafdc..a7df417 100644 --- a/src/or_audit/eval/job.py +++ b/src/or_audit/eval/job.py @@ -10,7 +10,7 @@ from pathlib import Path from typing import Annotated, Any, Literal, Self -from pydantic import BaseModel, ConfigDict, Field, model_validator +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from or_audit.audit.canonical import digest from or_audit.errors import TaskContractError @@ -18,6 +18,7 @@ from or_audit.eval.contracts import MetricKind from or_audit.eval.enums import WorldKind from or_audit.eval.integrity import tree_digest +from or_audit.eval.split import validate_subgroups from or_audit.eval.task import TaskSpec from or_audit.eval.trace import ProceduralTrace from or_audit.eval.vector import TrialVector @@ -33,6 +34,33 @@ class TrialRecord(BaseModel): trajectory: ProceduralTrace = Field(default_factory=lambda: ProceduralTrace(())) projection: float | None = None projection_spec_digest: str = "" + case_id: str = "" + patient_id: str = "" + site_id: str = "" + episode_id: str = "" + subgroups: dict[str, str] = Field(default_factory=dict) + + @field_validator("subgroups", mode="before") + @classmethod + def _check_subgroups(cls, value: object) -> dict[str, str]: + return validate_subgroups(value) if value is not None else {} + + +class TrialBinding(BaseModel): + """Head-covered trial metadata binding case, patient, site, and cohort subgroups.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + case_id: str = "" + patient_id: str = "" + site_id: str = "" + episode_id: str = "" + subgroups: dict[str, str] = Field(default_factory=dict) + + @field_validator("subgroups", mode="before") + @classmethod + def _check_subgroups(cls, value: object) -> dict[str, str]: + return validate_subgroups(value) if value is not None else {} class WorldEngineProvenance(BaseModel): @@ -97,6 +125,7 @@ class JobResult(BaseModel): independent_cases: Annotated[int, Field(ge=1)] | None = None split_manifest_digest: str = "" split: str = "" + independent_case_unit: str = "" gate_outcome: Literal["passed", "failed", "not-assessable", "unknown"] = "unknown" claim_footer: str = "" head: str = "" @@ -171,6 +200,7 @@ def assemble_job_result( independent_cases: int | None = None, split_manifest_digest: str = "", split: str = "", + independent_case_unit: str = "", ) -> JobResult: """Build a publishable job result and stamp its head.""" assert_publishable(task, trials, claim_footer) @@ -221,6 +251,7 @@ def assemble_job_result( independent_cases=independent_cases, split_manifest_digest=split_manifest_digest, split=split, + independent_case_unit=independent_case_unit, gate_outcome=( "failed" if gate_failed @@ -352,7 +383,7 @@ def write_trial( ) -> None: """Persist one completed trial atomically (crash-safe resume unit). - Writes trajectory, projection, and provenance first, then writes + Writes trajectory, projection, provenance, and binding first, then writes result.json last as the commit marker for the trial. """ trial_dir = out / f"trial-{task_id}-{trial.seed}" @@ -377,12 +408,32 @@ def write_trial( trial_dir / "provenance.json", json.dumps(world_engine, indent=2) + "\n", ) + binding = TrialBinding( + case_id=trial.case_id, + patient_id=trial.patient_id, + site_id=trial.site_id, + episode_id=trial.episode_id, + subgroups=trial.subgroups, + ) + if ( + binding.case_id + or binding.patient_id + or binding.site_id + or binding.episode_id + or binding.subgroups + ): + _atomic_write_text( + trial_dir / "binding.json", + json.dumps(binding.model_dump(mode="json"), indent=2) + "\n", + ) _atomic_write_text( trial_dir / "result.json", json.dumps(_vector_dict(trial.vector), indent=2) + "\n" ) -def read_partial_trials(out: Path, task_id: str) -> tuple[list[TrialRecord], dict[str, Any] | None]: +def read_partial_trials( + out: Path, task_id: str, *, require_binding: bool = False +) -> tuple[list[TrialRecord], dict[str, Any] | None]: """Read committed trials from an interrupted job directory. Reads trial directories created by :func:`write_trial` when the job-level @@ -394,6 +445,8 @@ def read_partial_trials(out: Path, task_id: str) -> tuple[list[TrialRecord], dic observed_provenance: dict[str, Any] | None = None trials_with_prov = 0 trials_without_prov = 0 + trials_with_binding = 0 + trials_without_binding = 0 for trial_dir in sorted(out.glob(f"trial-{task_id}-*")): vector_path = trial_dir / "result.json" trajectory_path = trial_dir / "trajectory.json" @@ -451,6 +504,23 @@ def read_partial_trials(out: Path, task_id: str) -> tuple[list[TrialRecord], dic ) projection = payload.get("projection") projection_spec_digest = str(payload.get("projection_spec_digest", "")) + binding_path = trial_dir / "binding.json" + binding = TrialBinding() + if binding_path.is_file(): + trials_with_binding += 1 + try: + b_raw = json.loads(binding_path.read_text(encoding="utf-8")) + binding = TrialBinding.model_validate(b_raw) + except (json.JSONDecodeError, ValueError) as exc: + raise TaskContractError( + f"trial dir {trial_dir.name} is corrupt: invalid binding.json ({exc})" + ) from exc + else: + trials_without_binding += 1 + if require_binding: + raise TaskContractError( + f"trial dir {trial_dir.name} is missing binding.json required by job split" + ) records.append( TrialRecord( seed=seed, @@ -458,6 +528,11 @@ def read_partial_trials(out: Path, task_id: str) -> tuple[list[TrialRecord], dic trajectory=trajectory, projection=projection, projection_spec_digest=projection_spec_digest, + case_id=binding.case_id, + patient_id=binding.patient_id, + site_id=binding.site_id, + episode_id=binding.episode_id, + subgroups=binding.subgroups, ) ) if trials_with_prov > 0 and trials_without_prov > 0: @@ -465,6 +540,11 @@ def read_partial_trials(out: Path, task_id: str) -> tuple[list[TrialRecord], dic f"job dir {out.name} has mixed provenance: {trials_with_prov} trial(s) have " f"provenance.json while {trials_without_prov} trial(s) are missing it" ) + if trials_with_binding > 0 and trials_without_binding > 0: + raise TaskContractError( + f"job dir {out.name} has mixed binding metadata: {trials_with_binding} trial(s) have " + f"binding.json while {trials_without_binding} trial(s) are missing it" + ) return sorted(records, key=lambda record: record.seed), observed_provenance diff --git a/src/or_audit/eval/leaderboard.py b/src/or_audit/eval/leaderboard.py index 1186268..05d61e0 100644 --- a/src/or_audit/eval/leaderboard.py +++ b/src/or_audit/eval/leaderboard.py @@ -59,6 +59,9 @@ def _metric_display(metric: Mapping[str, Any]) -> str: value = _metric_value(metric) if value is not None: unit = f" {metric['unit']}" if metric.get("unit") else "" + ci = metric.get("ci_95") + if ci: + return f"{value:.4g}{unit} [{ci[0]:.4g}, {ci[1]:.4g}]" return f"{value:.4g}{unit}" if metric["kind"] == "categorical": return ", ".join(f"{category}: {count}" for category, count in metric["counts"].items()) diff --git a/src/or_audit/eval/runner.py b/src/or_audit/eval/runner.py index 9c373b7..c54800b 100644 --- a/src/or_audit/eval/runner.py +++ b/src/or_audit/eval/runner.py @@ -52,7 +52,7 @@ world_kind_key, world_kind_spec, ) -from or_audit.eval.split import load_split_manifest +from or_audit.eval.split import DisjointUnit, load_split_manifest from or_audit.eval.task import TaskSpec from or_audit.eval.trace import ProceduralTrace from or_audit.eval.vector import project @@ -220,6 +220,7 @@ def _resume_trials( n: int, enabled: bool, split: str = "", + split_manifest_digest: str = "", ) -> tuple[dict[int, TrialRecord], dict[str, Any] | None]: """Completed trials to keep when resuming an interrupted job. @@ -248,13 +249,33 @@ def _resume_trials( f"cannot resume {out}: previous result split is {previous.split!r}, " f"requested {split!r}; cross-split resume is refused" ) + if previous.split_manifest_digest != split_manifest_digest: + raise TaskContractError( + f"cannot resume {out}: previous result split manifest digest is " + f"{previous.split_manifest_digest!r}, requested {split_manifest_digest!r}; " + "manifest changed since original run" + ) prev_prov = previous.world_engine.model_dump(mode="json") if previous.world_engine else None return {trial.seed: trial for trial in previous.trials}, prev_prov - return _resume_partial(out, task_id, task_digest, agent_digest, n, split=split) + return _resume_partial( + out, + task_id, + task_digest, + agent_digest, + n, + split=split, + split_manifest_digest=split_manifest_digest, + ) def _resume_partial( - out: Path, task_id: str, task_digest: str, agent_digest: str, n: int, split: str = "" + out: Path, + task_id: str, + task_digest: str, + agent_digest: str, + n: int, + split: str = "", + split_manifest_digest: str = "", ) -> tuple[dict[int, TrialRecord], dict[str, Any] | None]: """Resume a killed run: bundle/config plus completed trial dirs, no result. @@ -278,7 +299,15 @@ def _resume_partial( f"cannot resume {out}: configured n exceeds requested n={n}; " "shrinking a schedule would drop evidence" ) - records, observed_provenance = read_partial_trials(out, task_id) + stored_digest = str(config.get("split_manifest_digest", "")) + if stored_digest != split_manifest_digest: + raise TaskContractError( + f"cannot resume {out}: config split manifest digest is {stored_digest!r}, " + f"requested {split_manifest_digest!r}; manifest changed since original run" + ) + records, observed_provenance = read_partial_trials( + out, task_id, require_binding=bool(stored_digest) + ) foreign = sorted(record.seed for record in records if not 0 <= record.seed < n) if foreign: raise TaskContractError( @@ -299,6 +328,7 @@ def run_job( gym_factory: GymFactory | None = None, resume: bool = False, split: str | None = None, + independent_case_unit: str | None = None, ) -> JobResult: assert_bind(task, agent) task.assert_runnable() @@ -313,6 +343,14 @@ def run_job( episodes = n if n is not None else task.environment.n_eval_episodes if episodes < 1: raise TaskContractError(f"n must be >= 1, got {episodes}") + if ( + task.harness.interaction_mode is InteractionMode.CLOSED_LOOP + and task.environment.splits_path + ): + raise TaskContractError( + f"task {task.id} is closed-loop but declares splits_path; closed-loop tasks do not " + "currently support split manifests without an explicit seed-to-entry mapping" + ) if split and not task.environment.splits_path: raise TaskContractError( f"task {task.id} has no declared splits_path; cannot execute explicit split {split!r}" @@ -328,6 +366,22 @@ def run_job( else "" ) ) + normalized_unit = "" + if independent_case_unit: + u = independent_case_unit.strip().lower() + if u in ("patient", "patient_id"): + normalized_unit = "patient" + elif u in ("site", "site_id"): + normalized_unit = "site" + elif u in ("case", "case_id", "clip", "held-out clip"): + normalized_unit = "case" + else: + normalized_unit = u + split_manifest_digest = "" + if task.environment.splits_path: + manifest_path = task_dir / task.environment.splits_path + manifest = load_split_manifest(manifest_path) + split_manifest_digest = digest(manifest.model_dump(mode="json")) assert_trial_capacity(task, task_dir, episodes, split=target_split) resume_trials, previous_result = _resume_trials( out, @@ -337,6 +391,7 @@ def run_job( n=episodes, enabled=resume, split=target_split, + split_manifest_digest=split_manifest_digest, ) write_job_skeleton( out, @@ -353,6 +408,8 @@ def run_job( "world_pin": task.environment.world_pin, "interface": task.interface.id, "split": target_split, + "split_manifest_digest": split_manifest_digest, + "independent_case_unit": normalized_unit, }, task_dir=task_dir, agent_dir=agent_dir, @@ -388,6 +445,7 @@ def run_job( resume_trials=resume_trials, resume_provenance=previous_result, split=target_split, + independent_case_unit=normalized_unit, ) extra["safety_max_pen"] = safety extra["world_engine"] = provenance @@ -404,6 +462,7 @@ def run_job( resume_trials=resume_trials, resume_provenance=previous_result, split=target_split, + independent_case_unit=normalized_unit, ) elif task.harness.interaction_mode is InteractionMode.INTERACTIVE: result = _run_interactive( @@ -418,6 +477,7 @@ def run_job( resume_trials=resume_trials, resume_provenance=previous_result, split=target_split, + independent_case_unit=normalized_unit, ) elif task.harness.interaction_mode is InteractionMode.COUNTERFACTUAL: result = _run_counterfactual( @@ -432,6 +492,7 @@ def run_job( resume_trials=resume_trials, resume_provenance=previous_result, split=target_split, + independent_case_unit=normalized_unit, ) if previous_result is not None and resume_trials: current_prov = result.world_engine.model_dump(mode="json") if result.world_engine else None @@ -453,6 +514,8 @@ def run_job( "world_pin": task.environment.world_pin, "interface": task.interface.id, "split": result.split, + "split_manifest_digest": result.split_manifest_digest, + "independent_case_unit": result.independent_case_unit, **extra, } write_job(out, config=config, result=result, task_dir=task_dir, agent_dir=agent_dir) @@ -485,6 +548,7 @@ def _run_closed_loop( resume_trials: dict[int, TrialRecord] | None = None, resume_provenance: dict[str, Any] | None = None, split: str | None = None, + independent_case_unit: str = "", ) -> tuple[JobResult, float, dict[str, Any]]: if agent.kind not in {AgentKind.RANDOM.value, AgentKind.POLICY.value}: raise TaskContractError(f"closed-loop runner does not implement kind={agent.kind}") @@ -633,15 +697,6 @@ def action_fn( _close(policy) _close(verifier) _close(env) - split_manifest_digest = "" - independent_cases = None - target_split = split or "" - if task.environment.splits_path: - manifest_path = task_dir / task.environment.splits_path - manifest = load_split_manifest(manifest_path) - split_manifest_digest = digest(manifest.model_dump(mode="json")) - target_split = split or "test" - independent_cases = manifest.independent_case_count(target_split, unit="case") return ( assemble_job_result( task=task, @@ -650,9 +705,10 @@ def action_fn( task_digest=task_digest, world_engine=provenance, agent_digest=agent_digest, - independent_cases=independent_cases, - split_manifest_digest=split_manifest_digest, - split=target_split, + independent_cases=None, + split_manifest_digest="", + split="", + independent_case_unit=independent_case_unit, ), safety, provenance, @@ -673,6 +729,7 @@ def _run_predictions( resume_trials: dict[int, TrialRecord] | None = None, resume_provenance: dict[str, Any] | None = None, split: str | None = None, + independent_case_unit: str = "", ) -> JobResult: if agent_dir is None: raise TaskContractError(f"agent {agent.id} has no package directory") @@ -807,6 +864,11 @@ def _run_predictions( trace_payload[event_name] = item[event_name] trace = ProceduralTrace.from_steps([trace_payload], mode=mode) projection = project(vector, task.projection) if task.projection else None + m_entry = ( + next((e for e in manifest.entries if item_id in e.item_ids), None) + if manifest is not None + else None + ) trials.append( TrialRecord( seed=seed, @@ -814,6 +876,11 @@ def _run_predictions( trajectory=trace, projection=projection, projection_spec_digest=(task.projection.rule_digest if task.projection else ""), + case_id=m_entry.case_id if m_entry else "", + patient_id=m_entry.patient_id or "" if m_entry else "", + site_id=m_entry.site_id or "" if m_entry else "", + episode_id=m_entry.episode_id if m_entry else "", + subgroups=dict(m_entry.subgroups) if m_entry else {}, ) ) write_trial( @@ -832,7 +899,16 @@ def _run_predictions( independent_cases = None if manifest is not None: evaluated_ids = {str(item["id"]) for item in inputs[:n]} - independent_cases = manifest.independent_case_count_for_items(evaluated_ids, unit="case") + count_unit: DisjointUnit = ( + "patient" + if independent_case_unit == "patient" + else "site" + if independent_case_unit == "site" + else "case" + ) + independent_cases = manifest.independent_case_count_for_items( + evaluated_ids, unit=count_unit + ) return assemble_job_result( task=task, agent=agent, @@ -844,6 +920,7 @@ def _run_predictions( independent_cases=independent_cases, split_manifest_digest=split_manifest_digest, split=target_split, + independent_case_unit=independent_case_unit, ) @@ -864,6 +941,7 @@ def _run_interactive( resume_trials: dict[int, TrialRecord] | None = None, resume_provenance: dict[str, Any] | None = None, split: str | None = None, + independent_case_unit: str = "", ) -> JobResult: if agent_dir is None: raise TaskContractError(f"agent {agent.id} has no package directory") @@ -1016,6 +1094,11 @@ def _run_interactive( mode=InteractionMode.INTERACTIVE, ) projection = project(vector, task.projection) if task.projection else None + m_entry = ( + next((e for e in manifest.entries if item_id in e.item_ids), None) + if manifest is not None + else None + ) trials.append( TrialRecord( seed=seed, @@ -1023,6 +1106,11 @@ def _run_interactive( trajectory=trace, projection=projection, projection_spec_digest=(task.projection.rule_digest if task.projection else ""), + case_id=m_entry.case_id if m_entry else "", + patient_id=m_entry.patient_id or "" if m_entry else "", + site_id=m_entry.site_id or "" if m_entry else "", + episode_id=m_entry.episode_id if m_entry else "", + subgroups=dict(m_entry.subgroups) if m_entry else {}, ) ) write_trial( @@ -1041,7 +1129,16 @@ def _run_interactive( independent_cases = None if manifest is not None: evaluated_ids = {str(item["id"]) for item in inputs[:n]} - independent_cases = manifest.independent_case_count_for_items(evaluated_ids, unit="case") + count_unit: DisjointUnit = ( + "patient" + if independent_case_unit == "patient" + else "site" + if independent_case_unit == "site" + else "case" + ) + independent_cases = manifest.independent_case_count_for_items( + evaluated_ids, unit=count_unit + ) return assemble_job_result( task=task, agent=agent, @@ -1053,6 +1150,7 @@ def _run_interactive( independent_cases=independent_cases, split_manifest_digest=split_manifest_digest, split=target_split, + independent_case_unit=independent_case_unit, ) @@ -1100,6 +1198,9 @@ def replay_job( n=int(config["n"]), gym_factory=gym_factory, split=config.get("split") or previous.split or None, + independent_case_unit=( + config.get("independent_case_unit") or previous.independent_case_unit or None + ), ) if rerun.head != previous.head: raise TaskContractError(f"replay head mismatch: stored {previous.head} reran {rerun.head}") diff --git a/src/or_audit/eval/scorecard.py b/src/or_audit/eval/scorecard.py index c5c42a0..a58a24e 100644 --- a/src/or_audit/eval/scorecard.py +++ b/src/or_audit/eval/scorecard.py @@ -9,10 +9,15 @@ from statistics import fmean from typing import Any +from or_audit.errors import TaskContractError from or_audit.eval.contracts import MetricKind from or_audit.eval.job import JobResult, TrialRecord from or_audit.eval.sim.base import BACKEND_SYNTHETIC_STUB, BACKEND_UNKNOWN -from or_audit.eval.uncertainty import bootstrap_mean_ci, wilson_score_interval +from or_audit.eval.uncertainty import ( + bootstrap_mean_ci, + clustered_bootstrap_mean_ci, + wilson_score_interval, +) STUB_HEADLINE = "NOT PHYSICAL EVIDENCE - SYNTHETIC STAND-IN" METRICS_ONLY_HEADLINE = "METRICS-ONLY - NOT SAFETY-ATTESTED" @@ -40,20 +45,6 @@ def _is_abstained(trial: TrialRecord) -> bool: return bool(ab_metric is not None and ab_metric.value is True) -def _trial_subgroup(trial: TrialRecord) -> str: - """Extract declared subgroup from trial trajectory/metadata; never invent seeds.""" - for step in trial.trajectory: - sc = step.get("scenario") if isinstance(step, dict) else getattr(step, "scenario", None) - if sc is not None: - if isinstance(sc, dict): - subgroup = sc.get("subgroup") or sc.get("id") - else: - subgroup = getattr(sc, "subgroup", None) or getattr(sc, "id", None) - if subgroup: - return str(subgroup) - return "" - - def scorecard_data( result: JobResult, *, @@ -99,17 +90,61 @@ def scorecard_data( "false": assessed.count(False), "rate": rate, "ci_95": ci_95, + "ci_method": "wilson", + "confidence": 0.95, } ) elif row["kind"] == "continuous": numeric = [float(value) for value in assessed] - ci_95 = list(bootstrap_mean_ci(numeric)) if numeric else None + declared_unit = result.independent_case_unit.strip().lower() + if declared_unit in ("patient", "patient_id"): + target_attr = "patient_id" + elif declared_unit in ("site", "site_id"): + target_attr = "site_id" + elif declared_unit in ("case", "case_id", "clip", "held-out clip"): + target_attr = "case_id" + else: + target_attr = "" + + clusters: dict[str, list[float]] = defaultdict(list) + ci_method = "bootstrap" + mean_val: float | None = None + + if target_attr: + for trial in result.trials: + m = trial.vector.metric(metric_id) + if m is not None and m.value is not None: + unit_val = getattr(trial, target_attr, "") + if not unit_val: + raise TaskContractError( + f"declared independent_case_unit {declared_unit!r} missing on " + f"trial seed {trial.seed}" + ) + clusters[unit_val].append(float(m.value)) + + if len(clusters) >= 2: + ci_95 = list(clustered_bootstrap_mean_ci(clusters, seed=0)) + ci_method = "clustered_bootstrap" + cluster_level_means = [fmean(v) for v in clusters.values() if v] + mean_val = fmean(cluster_level_means) if cluster_level_means else None + else: + ci_95 = list(bootstrap_mean_ci(numeric, seed=0)) if numeric else None + ci_method = "bootstrap" + mean_val = fmean(numeric) if numeric else None + else: + ci_95 = list(bootstrap_mean_ci(numeric, seed=0)) if numeric else None + ci_method = "bootstrap" + mean_val = fmean(numeric) if numeric else None row.update( { - "mean": fmean(numeric) if numeric else None, + "mean": mean_val, "min": min(numeric) if numeric else None, "max": max(numeric) if numeric else None, "ci_95": ci_95, + "ci_method": ci_method, + "confidence": 0.95, + "resampling_seed": 0, + "draws": 1000, } ) else: @@ -119,7 +154,11 @@ def scorecard_data( metrics.append(row) headline_outcome = next((m for m in result.trials[0].vector.metrics if m.headline), None) headline_kind = headline_outcome.kind if headline_outcome else MetricKind.BOOLEAN - + headline_direction = ( + headline_outcome.direction.value + if headline_outcome and hasattr(headline_outcome.direction, "value") + else (str(headline_outcome.direction) if headline_outcome else "maximize") + ) # Coverage & Risk analysis for abstaining models (Phase D5) abstained_count = sum(1 for trial in result.trials if _is_abstained(trial)) coverage = (result.n - abstained_count) / result.n if result.n > 0 else 0.0 @@ -142,53 +181,107 @@ def scorecard_data( } # Subgroups & Worst-Case Scenario Analysis (Phase D5) - # Subgroups must come from declared manifest metadata only; absent metadata reports unavailable. - scenario_trials: dict[str, list[TrialRecord]] = defaultdict(list) + # Subgroups come strictly from explicit cohort fields (trial.subgroups). + # patient_id, site_id, and case_id are reserved SOLELY for clustering. + cohort_keys: set[str] = set() for trial in result.trials: - sg = _trial_subgroup(trial) - if sg: - scenario_trials[sg].append(trial) - - subgroups = [] + cohort_keys.update(trial.subgroups.keys()) + subgroups: list[dict[str, Any]] = [] worst_case = None - if len(scenario_trials) > 1: - worst_rate = 1.1 - for sc_id, s_trials in sorted(scenario_trials.items()): - s_count = len(s_trials) - if headline_kind is MetricKind.BOOLEAN: - s_pass = sum( - 1 - for t in s_trials - if not t.vector.any_gate_failed and t.vector.headline.value is True - ) - s_rate = s_pass / s_count if s_count > 0 else 0.0 - ci = list(wilson_score_interval(s_pass, s_count)) if s_count > 0 else [0.0, 1.0] - elif headline_kind is MetricKind.CONTINUOUS: - s_vals = [ - float(t.vector.headline.value) - for t in s_trials - if t.vector.headline.value is not None - ] - s_rate = fmean(s_vals) if s_vals else 0.0 - ci = list(bootstrap_mean_ci(s_vals)) if s_vals else [0.0, 0.0] - s_pass = sum(1 for t in s_trials if not t.vector.any_gate_failed) - else: - s_pass = sum(1 for t in s_trials if not t.vector.any_gate_failed) - s_rate = s_pass / s_count if s_count > 0 else 0.0 - ci = list(wilson_score_interval(s_pass, s_count)) if s_count > 0 else [0.0, 1.0] + for axis in sorted(cohort_keys): + axis_groups: dict[str, list[TrialRecord]] = defaultdict(list) + for trial in result.trials: + val = trial.subgroups.get(axis) + if val: + axis_groups[val].append(trial) - entry = { - "subgroup": sc_id, - "count": s_count, - "pass": s_pass, - "rate": round(s_rate, 4), - "ci_95": ci, - "is_underpowered": s_count < 10, - } - subgroups.append(entry) - if s_rate < worst_rate: - worst_rate = s_rate - worst_case = entry + if len(axis_groups) > 1: + for val, s_trials in sorted(axis_groups.items()): + s_count = len(s_trials) + s_rate: float | None = None + ci: list[float] | None = None + if headline_kind is MetricKind.BOOLEAN: + assessed_trials = [ + t + for t in s_trials + if t.vector.headline.value is not None and not _is_abstained(t) + ] + s_assessed = len(assessed_trials) + s_unassessable = s_count - s_assessed + if s_assessed > 0: + s_pass = sum( + 1 + for t in assessed_trials + if not t.vector.any_gate_failed and t.vector.headline.value is True + ) + s_rate = round(s_pass / s_assessed, 4) + ci = list(wilson_score_interval(s_pass, s_assessed)) + else: + s_pass = 0 + elif headline_kind is MetricKind.CONTINUOUS: + assessed_trials = [ + t + for t in s_trials + if t.vector.headline.value is not None and not _is_abstained(t) + ] + s_assessed = len(assessed_trials) + s_unassessable = s_count - s_assessed + if s_assessed > 0: + s_vals = [ + float(t.vector.headline.value) + for t in assessed_trials + if isinstance(t.vector.headline.value, int | float) + ] + s_rate = round(fmean(s_vals), 4) + ci = list(bootstrap_mean_ci(s_vals)) + s_pass = sum(1 for t in assessed_trials if not t.vector.any_gate_failed) + else: + s_pass = 0 + else: + assessed_trials = [ + t + for t in s_trials + if t.vector.headline.value is not None and not _is_abstained(t) + ] + s_assessed = len(assessed_trials) + s_unassessable = s_count - s_assessed + if s_assessed > 0: + s_pass = sum(1 for t in assessed_trials if not t.vector.any_gate_failed) + s_rate = round(s_pass / s_assessed, 4) + ci = list(wilson_score_interval(s_pass, s_assessed)) + else: + s_pass = 0 + + entry = { + "axis": axis, + "subgroup": val, + "count": s_count, + "assessed": s_assessed, + "unassessable": s_unassessable, + "pass": s_pass, + "rate": s_rate, + "ci_95": ci, + "is_underpowered": s_assessed < 10, + } + subgroups.append(entry) + + worst_cases: dict[str, dict[str, Any]] = {} + if headline_direction in ("maximize", "minimize"): + for axis in sorted(cohort_keys): + axis_entries = [sg for sg in subgroups if sg["axis"] == axis and sg["rate"] is not None] + if axis_entries: + if headline_direction == "minimize": + worst_cases[axis] = max( + axis_entries, + key=lambda sg: (float(str(sg["rate"])), -int(str(sg["assessed"]))), + ) + else: + worst_cases[axis] = min( + axis_entries, + key=lambda sg: (float(str(sg["rate"])), int(str(sg["assessed"]))), + ) + + worst_case = next(iter(worst_cases.values())) if len(worst_cases) == 1 else None return { "task_id": result.task_id, "task_version": result.task_version, @@ -210,9 +303,10 @@ def scorecard_data( "head": result.head, "independent_cases": result.independent_cases, "split_manifest_digest": result.split_manifest_digest, + "worst_cases": worst_cases, + "worst_case": worst_case, "coverage": coverage_report, "subgroups": subgroups, - "worst_case": worst_case, } @@ -327,28 +421,38 @@ def render_markdown( ), ] ) - if len(data["subgroups"]) > 1: + if data["subgroups"]: lines.extend( [ "", "## Subgroups and worst-case analysis", "", - "| Subgroup | Count | Pass rate | 95% CI | Power |", - "|---|---:|---:|:---:|:---:|", + "| Cohort axis | Subgroup | Count | Assessed | Unassessable | " + "Estimate | 95% CI | Power |", + "|---|---|---:|---:|---:|---:|:---:|:---:|", ] ) for sg in data["subgroups"]: - ci_str = f"[{sg['ci_95'][0]:.4f}, {sg['ci_95'][1]:.4f}]" + ci_str = f"[{sg['ci_95'][0]:.4f}, {sg['ci_95'][1]:.4f}]" if sg["ci_95"] else "n/a" pwr = "underpowered (<10)" if sg["is_underpowered"] else "adequate" + rate_str = "n/a" if sg["rate"] is None else f"{sg['rate']:.4f}" lines.append( - f"| {sg['subgroup']} | {sg['count']} | {sg['rate']:.4f} | {ci_str} | {pwr} |" + f"| {sg['axis']} | {sg['subgroup']} | {sg['count']} | " + f"{sg['assessed']} | {sg['unassessable']} | " + f"{rate_str} | {ci_str} | {pwr} |" ) - if data["worst_case"]: + if data.get("worst_case") and data["worst_case"]["rate"] is not None: wc = data["worst_case"] lines.append( - f"\n> **Worst-case subgroup:** `{wc['subgroup']}` " - f"with pass rate `{wc['rate']:.4f}`." + f"\n> **Worst-case subgroup:** `{wc['axis']}={wc['subgroup']}` " + f"with estimate `{wc['rate']:.4f}`." ) + elif data.get("worst_cases"): + for axis, wc in sorted(data["worst_cases"].items()): + lines.append( + f"\n> **Worst-case subgroup ({axis}):** `{wc['subgroup']}` " + f"with estimate `{wc['rate']:.4f}`." + ) if data["claim_footer"]: lines.extend(["", "## Claim boundary", "", data["claim_footer"]]) lines.extend( diff --git a/src/or_audit/eval/split.py b/src/or_audit/eval/split.py index fd08f90..476ebac 100644 --- a/src/or_audit/eval/split.py +++ b/src/or_audit/eval/split.py @@ -8,12 +8,20 @@ from __future__ import annotations import json +import re from collections import defaultdict from collections.abc import Iterable from pathlib import Path from typing import Annotated, Literal, Self -from pydantic import BaseModel, ConfigDict, Field, StringConstraints, model_validator +from pydantic import ( + BaseModel, + ConfigDict, + Field, + StringConstraints, + field_validator, + model_validator, +) from or_audit.errors import TaskContractError @@ -21,6 +29,67 @@ Identifier = Annotated[str, StringConstraints(min_length=1, max_length=128)] DisjointUnit = Literal["case", "patient", "site"] +RESERVED_SUBGROUP_KEYS: frozenset[str] = frozenset( + { + "patient", + "patient_id", + "site", + "site_id", + "case", + "case_id", + "episode", + "episode_id", + "seed", + } +) +MAX_SUBGROUP_AXES = 16 +MAX_SUBGROUP_KEY_LENGTH = 64 +MAX_SUBGROUP_VALUE_LENGTH = 128 +SUBGROUP_KEY_PATTERN = re.compile(r"^[a-z0-9][a-z0-9_-]*$") + + +def validate_subgroups(raw: object) -> dict[str, str]: + """Validate bounded, slug-keyed, categorical subgroup mapping.""" + if not isinstance(raw, dict): + raise TaskContractError(f"subgroups must be a dictionary, got {type(raw).__name__}") + if len(raw) > MAX_SUBGROUP_AXES: + raise TaskContractError( + f"subgroups dictionary exceeds maximum of {MAX_SUBGROUP_AXES} axes: {len(raw)}" + ) + validated: dict[str, str] = {} + for key, val in raw.items(): + if not isinstance(key, str) or not isinstance(val, str): + raise TaskContractError( + f"subgroup keys and values must be strings, got " + f"{type(key).__name__}:{type(val).__name__}" + ) + k = key.strip() + v = val.strip() + if not k: + raise TaskContractError("subgroup key cannot be empty") + if not v: + raise TaskContractError(f"subgroup value for key {k!r} cannot be empty") + if len(k) > MAX_SUBGROUP_KEY_LENGTH: + raise TaskContractError( + f"subgroup key {k!r} exceeds maximum length of {MAX_SUBGROUP_KEY_LENGTH}" + ) + if len(v) > MAX_SUBGROUP_VALUE_LENGTH: + raise TaskContractError( + f"subgroup value for key {k!r} exceeds maximum length " + f"of {MAX_SUBGROUP_VALUE_LENGTH}" + ) + if not SUBGROUP_KEY_PATTERN.match(k): + raise TaskContractError( + f"subgroup key {k!r} must match slug pattern {SUBGROUP_KEY_PATTERN.pattern}" + ) + if k in RESERVED_SUBGROUP_KEYS: + raise TaskContractError( + f"subgroup key {k!r} is a reserved identifier; patient/site/case/episode IDs " + "must be specified in their dedicated fields, not subgroups" + ) + validated[k] = v + return validated + class SplitCaseEntry(BaseModel): """Binding for one coherent case/episode in a split.""" @@ -34,6 +103,12 @@ class SplitCaseEntry(BaseModel): site_id: Identifier | None = None scenario_id: Identifier | None = None item_ids: tuple[Identifier, ...] = Field(default_factory=tuple) + subgroups: dict[str, str] = Field(default_factory=dict) + + @field_validator("subgroups", mode="before") + @classmethod + def _check_subgroups(cls, value: object) -> dict[str, str]: + return validate_subgroups(value) if value is not None else {} @model_validator(mode="after") def _validate_items(self) -> Self: diff --git a/src/or_audit/eval/uncertainty.py b/src/or_audit/eval/uncertainty.py index b05b023..b6fa73d 100644 --- a/src/or_audit/eval/uncertainty.py +++ b/src/or_audit/eval/uncertainty.py @@ -107,23 +107,23 @@ def clustered_bootstrap_mean_ci( """Clustered percentile bootstrap resampling by independent cluster key (e.g. patient).""" if not clusters: raise TaskContractError("clustered bootstrap needs at least one cluster") - cluster_keys = sorted(clusters.keys()) - k = len(cluster_keys) + for key, vals in clusters.items(): + if not vals: + raise TaskContractError( + f"cluster {key!r} is empty; clustered bootstrap requires non-empty observations" + ) + cluster_means = {key: (sum(vals) / len(vals)) for key, vals in clusters.items()} + valid_keys = sorted(cluster_means.keys()) + k = len(valid_keys) if k == 1: - single_vals = clusters[cluster_keys[0]] - return bootstrap_mean_ci(single_vals, confidence=confidence, draws=draws, seed=seed) + val = round(cluster_means[valid_keys[0]], 4) + return val, val rng = random.Random(seed) means: list[float] = [] for _ in range(draws): - sampled_keys = rng.choices(cluster_keys, k=k) - sampled_values: list[float] = [] - for key in sampled_keys: - sampled_values.extend(clusters[key]) - if sampled_values: - means.append(sum(sampled_values) / len(sampled_values)) - if not means: - raise TaskContractError("all sampled bootstrap clusters were empty") + sampled_keys = rng.choices(valid_keys, k=k) + means.append(sum(cluster_means[key] for key in sampled_keys) / k) means.sort() lower_idx = int((1.0 - confidence) / 2.0 * draws) diff --git a/tests/test_eval_p3.py b/tests/test_eval_p3.py index e28c4b9..139eabc 100644 --- a/tests/test_eval_p3.py +++ b/tests/test_eval_p3.py @@ -189,7 +189,10 @@ def test_stage_run_enforces_units_supported_axes_and_head_covers_outcome( assert manifest.observed_units == 2 assert manifest.stage.target_units == sum(pair.n for pair in manifest.pairs) assert read_manifest(out).head == manifest.head - + pair_result = json.loads( + (out / manifest.pairs[0].dir / "result.json").read_text(encoding="utf-8") + ) + assert pair_result["independent_case_unit"] == "scenario-target seed" with pytest.raises(TaskContractError, match="schedules 3"): run_cartesian_job(resolved, out=tmp_path / "wrong-n", n=3, gym_factory=_fake) diff --git a/tests/test_split_manifest.py b/tests/test_split_manifest.py index d240fe6..8f9a328 100644 --- a/tests/test_split_manifest.py +++ b/tests/test_split_manifest.py @@ -893,3 +893,52 @@ def test_explicit_split_without_splits_path_refused(tmp_path: Path) -> None: n=1, split="test", ) + + +def test_independent_case_unit_counts_patient_cases_when_declared(tmp_path: Path) -> None: + import shutil + + from or_audit.eval.loader import load_agent, load_task + from or_audit.eval.runner import run_job + + root = Path(__file__).resolve().parents[1] + task_src = root / "docs/examples/tasks/video-nextstep" + agent_src = root / "docs/examples/agents/example-video-predictor" + task_dir = tmp_path / "task-patient-split" + shutil.copytree(task_src, task_dir) + + splits_file = task_dir / "splits.json" + splits_data = json.loads(splits_file.read_text(encoding="utf-8")) + splits_data["entries"][0]["patient_id"] = "pt-shared" + splits_data["entries"][1]["patient_id"] = "pt-shared" + splits_file.write_text(json.dumps(splits_data), encoding="utf-8") + + task = load_task(task_dir) + agent = load_agent(agent_src) + + # 1. When independent_case_unit="patient" -> counts distinct patients (1) + res_patient = run_job( + task=task, + task_dir=task_dir, + agent=agent, + agent_dir=agent_src, + out=tmp_path / "out-patient", + n=2, + split="test", + independent_case_unit="patient", + ) + assert res_patient.independent_case_unit == "patient" + assert res_patient.independent_cases == 1 + + # 2. When default (unclustered / case unit) -> counts distinct cases (2) + res_case = run_job( + task=task, + task_dir=task_dir, + agent=agent, + agent_dir=agent_src, + out=tmp_path / "out-case", + n=2, + split="test", + ) + assert res_case.independent_case_unit == "" + assert res_case.independent_cases == 2 diff --git a/tests/test_uncertainty.py b/tests/test_uncertainty.py index 0a92e77..8eb2a1f 100644 --- a/tests/test_uncertainty.py +++ b/tests/test_uncertainty.py @@ -8,7 +8,7 @@ from or_audit.domain.enums import GateStatus from or_audit.errors import TaskContractError -from or_audit.eval.contracts import MetricKind +from or_audit.eval.contracts import MetricDirection, MetricKind from or_audit.eval.job import JobResult, TrialRecord from or_audit.eval.loader import load_agent, load_task from or_audit.eval.runner import run_job @@ -105,13 +105,23 @@ def test_scorecard_data_includes_uncertainty_intervals(tmp_path: Path) -> None: def test_scorecard_coverage_and_subgroup_analysis() -> None: + from or_audit.eval.contracts import MetricDirection + vector_pass = TrialVector( task_id="test-task", task_version="1", agent_identity="agent", seed=0, gates=(GateOutcome(id="g1", status=GateStatus.PASS),), - metrics=(MetricOutcome(id="m1", kind=MetricKind.BOOLEAN, headline=True, value=True),), + metrics=( + MetricOutcome( + id="m1", + kind=MetricKind.BOOLEAN, + headline=True, + value=True, + direction=MetricDirection.MAXIMIZE, + ), + ), ) vector_abstained = TrialVector( task_id="test-task", @@ -120,7 +130,13 @@ def test_scorecard_coverage_and_subgroup_analysis() -> None: seed=1, gates=(GateOutcome(id="g1", status=GateStatus.NOT_ASSESSABLE, abstained=True),), metrics=( - MetricOutcome(id="m1", kind=MetricKind.BOOLEAN, headline=True, value=None), + MetricOutcome( + id="m1", + kind=MetricKind.BOOLEAN, + headline=True, + value=None, + direction=MetricDirection.MAXIMIZE, + ), MetricOutcome(id="abstained", kind=MetricKind.BOOLEAN, headline=False, value=True), ), ) @@ -130,7 +146,15 @@ def test_scorecard_coverage_and_subgroup_analysis() -> None: agent_identity="agent", seed=2, gates=(GateOutcome(id="g1", status=GateStatus.FAIL),), - metrics=(MetricOutcome(id="m1", kind=MetricKind.BOOLEAN, headline=True, value=False),), + metrics=( + MetricOutcome( + id="m1", + kind=MetricKind.BOOLEAN, + headline=True, + value=False, + direction=MetricDirection.MAXIMIZE, + ), + ), ) step_sc1 = TraceStep.model_validate( @@ -148,9 +172,27 @@ def test_scorecard_coverage_and_subgroup_analysis() -> None: } ) - trial0 = TrialRecord(seed=0, vector=vector_pass, trajectory=ProceduralTrace((step_sc1,))) - trial1 = TrialRecord(seed=1, vector=vector_abstained, trajectory=ProceduralTrace((step_sc1,))) - trial2 = TrialRecord(seed=2, vector=vector_fail, trajectory=ProceduralTrace((step_sc2,))) + trial0 = TrialRecord( + seed=0, + vector=vector_pass, + trajectory=ProceduralTrace((step_sc1,)), + patient_id="patient-1", + subgroups={"anatomy": "bifurcation"}, + ) + trial1 = TrialRecord( + seed=1, + vector=vector_abstained, + trajectory=ProceduralTrace((step_sc1,)), + patient_id="patient-1", + subgroups={"anatomy": "bifurcation"}, + ) + trial2 = TrialRecord( + seed=2, + vector=vector_fail, + trajectory=ProceduralTrace((step_sc2,)), + patient_id="patient-2", + subgroups={"anatomy": "straight"}, + ) result = JobResult( task_id="test-task", @@ -176,15 +218,791 @@ def test_scorecard_coverage_and_subgroup_analysis() -> None: assert data["coverage"]["risk_at_coverage"] == 0.5 assert len(data["subgroups"]) == 2 - sg_a = next(sg for sg in data["subgroups"] if sg["subgroup"] == "scenario-a") - sg_b = next(sg for sg in data["subgroups"] if sg["subgroup"] == "scenario-b") - assert sg_a["count"] == 2 - assert sg_a["is_underpowered"] is True - assert sg_b["count"] == 1 - assert sg_b["rate"] == 0.0 - assert data["worst_case"]["subgroup"] == "scenario-b" + sg_bif = next(sg for sg in data["subgroups"] if sg["subgroup"] == "bifurcation") + sg_str = next(sg for sg in data["subgroups"] if sg["subgroup"] == "straight") + assert sg_bif["count"] == 2 + assert sg_bif["is_underpowered"] is True + assert sg_str["count"] == 1 + assert sg_str["rate"] == 0.0 + assert data["worst_case"]["subgroup"] == "straight" + assert data["worst_case"]["axis"] == "anatomy" + # Patient identifiers must never be exposed as subgroups + subgroup_names = [sg["subgroup"] for sg in data["subgroups"]] + assert "patient-1" not in subgroup_names + assert "patient-2" not in subgroup_names markdown = render_markdown(result) assert "## Risk vs coverage" in markdown assert "## Subgroups and worst-case analysis" in markdown - assert "**Worst-case subgroup:** `scenario-b`" in markdown + assert "**Worst-case subgroup:** `anatomy=straight`" in markdown + + +def test_generated_benchmark_report_fixture(tmp_path: Path) -> None: + import json + import shutil + + from or_audit.eval.scorecard import write_scorecards + + task_src = ROOT / "docs/examples/tasks/video-nextstep" + agent_src = ROOT / "docs/examples/agents/example-video-predictor" + + task_dir = tmp_path / "benchmark-report-task" + shutil.copytree(task_src, task_dir) + + # Manifest defining explicit cohorts and patient clusters + splits_data = { + "format_version": "1", + "dataset_id": "benchmark-video-dataset", + "dataset_revision": "1.0", + "disjoint_by": ["case", "patient"], + "entries": [ + { + "case_id": "case-01", + "episode_id": "ep-01", + "patient_id": "patient-A", + "site_id": "site-1", + "split": "test", + "item_ids": ["clip-001"], + "subgroups": {"anatomy": "tortuous", "scanner": "siemens"}, + }, + { + "case_id": "case-02", + "episode_id": "ep-02", + "patient_id": "patient-A", + "site_id": "site-1", + "split": "test", + "item_ids": ["clip-002"], + "subgroups": {"anatomy": "tortuous", "scanner": "siemens"}, + }, + { + "case_id": "case-03", + "episode_id": "ep-03", + "patient_id": "patient-B", + "site_id": "site-2", + "split": "test", + "item_ids": ["clip-003"], + "subgroups": {"anatomy": "straight", "scanner": "ge"}, + }, + ], + } + (task_dir / "splits.json").write_text(json.dumps(splits_data), encoding="utf-8") + + out = tmp_path / "benchmark-out" + result = run_job( + task=load_task(task_dir), + task_dir=task_dir, + agent=load_agent(agent_src), + agent_dir=agent_src, + out=out, + n=3, + split="test", + ) + + write_scorecards(out, result) + + scorecard_json_path = out / "scorecard.json" + scorecard_md_path = out / "scorecard.md" + scorecard_html_path = out / "scorecard.html" + assert scorecard_json_path.is_file() + assert scorecard_md_path.is_file() + assert scorecard_html_path.is_file() + + report = json.loads(scorecard_json_path.read_text(encoding="utf-8")) + assert report["n"] == 3 + assert report["independent_cases"] == 3 + assert report["split_manifest_digest"] == result.split_manifest_digest + + # Verify uncertainty intervals and methods on metrics + headline_metric = next(m for m in report["metrics"] if m["headline"]) + assert "ci_95" in headline_metric + assert headline_metric["ci_method"] == "wilson" + assert headline_metric["confidence"] == 0.95 + + # Verify subgroup analysis: must be grouped along declared cohort axes, NOT patient IDs! + assert len(report["subgroups"]) >= 2 + axes = {sg["axis"] for sg in report["subgroups"]} + assert "anatomy" in axes or "scanner" in axes + assert "patient" not in axes + + md = scorecard_md_path.read_text(encoding="utf-8") + assert "| 95% CI |" in md + assert "## Subgroups and worst-case analysis" in md + assert "Worst-case subgroup (anatomy):" in md + assert "Worst-case subgroup (scanner):" in md + + +def test_leaderboard_renders_uncertainty_intervals(tmp_path: Path) -> None: + from or_audit.eval.leaderboard import leaderboard_data, render_html + + task_src = ROOT / "docs/examples/tasks/video-nextstep" + agent_src = ROOT / "docs/examples/agents/example-video-predictor" + + out = tmp_path / "lb-job" + run_job( + task=load_task(task_src), + task_dir=task_src, + agent=load_agent(agent_src), + agent_dir=agent_src, + out=out, + n=3, + split="test", + ) + + data = leaderboard_data([out]) + row = data["rows"][0] + headline_metric = row["metrics"][row["headline"]] + assert "ci_95" in headline_metric + assert headline_metric["ci_95"] is not None + + html_text = render_html(data) + assert "[" in html_text + assert "]" in html_text + + +def test_subgroups_validation_constraints() -> None: + from or_audit.eval.split import validate_subgroups + + # Valid mapping passes + valid = validate_subgroups({"anatomy": "bifurcation", "scanner_model": "siemens-1"}) + assert valid == {"anatomy": "bifurcation", "scanner_model": "siemens-1"} + + # Reserved keys rejected + with pytest.raises(TaskContractError, match="reserved identifier"): + validate_subgroups({"patient": "p1"}) + with pytest.raises(TaskContractError, match="reserved identifier"): + validate_subgroups({"site_id": "s1"}) + with pytest.raises(TaskContractError, match="reserved identifier"): + validate_subgroups({"case": "c1"}) + + # Non-slug key rejected + with pytest.raises(TaskContractError, match="must match slug pattern"): + validate_subgroups({"Invalid Key!": "val"}) + + # Non-string value rejected + with pytest.raises(TaskContractError, match="keys and values must be strings"): + validate_subgroups({"anatomy": 123}) + + # Empty key or value rejected + with pytest.raises(TaskContractError, match="cannot be empty"): + validate_subgroups({"": "val"}) + with pytest.raises(TaskContractError, match="cannot be empty"): + validate_subgroups({"anatomy": ""}) + + # Exceeding maximum 16 axes rejected + too_many = {f"axis-{i}": f"val-{i}" for i in range(17)} + with pytest.raises(TaskContractError, match="exceeds maximum of 16 axes"): + validate_subgroups(too_many) + + +def test_trial_binding_model_strictness() -> None: + from pydantic import ValidationError + + from or_audit.eval.job import TrialBinding + + # Extra fields forbidden + with pytest.raises(ValidationError): + TrialBinding.model_validate({"case_id": "c1", "unknown_field": "val"}) + + +def test_crash_resume_restores_bindings_intact(tmp_path: Path) -> None: + from or_audit.eval.job import read_partial_trials + + task_src = ROOT / "docs/examples/tasks/video-nextstep" + agent_src = ROOT / "docs/examples/agents/example-video-predictor" + + out = tmp_path / "job-resume-bindings" + original = run_job( + task=load_task(task_src), + task_dir=task_src, + agent=load_agent(agent_src), + agent_dir=agent_src, + out=out, + n=2, + split="test", + ) + assert original.trials[0].case_id + assert original.trials[0].patient_id + + # Simulate crash by removing result.json + (out / "result.json").unlink() + + # Verify read_partial_trials recovers bindings + records, _ = read_partial_trials(out, "video-nextstep", require_binding=True) + assert len(records) == 2 + assert records[0].case_id == original.trials[0].case_id + assert records[0].patient_id == original.trials[0].patient_id + assert records[0].site_id == original.trials[0].site_id + + # Resume the job + resumed = run_job( + task=load_task(task_src), + task_dir=task_src, + agent=load_agent(agent_src), + agent_dir=agent_src, + out=out, + n=2, + split="test", + resume=True, + ) + assert resumed.head == original.head + assert resumed.trials[0].case_id == original.trials[0].case_id + assert resumed.trials[0].patient_id == original.trials[0].patient_id + + +def test_corrupt_or_missing_binding_json_refuses(tmp_path: Path) -> None: + from or_audit.eval.job import read_partial_trials + + task_src = ROOT / "docs/examples/tasks/video-nextstep" + agent_src = ROOT / "docs/examples/agents/example-video-predictor" + + out = tmp_path / "job-corrupt-binding" + run_job( + task=load_task(task_src), + task_dir=task_src, + agent=load_agent(agent_src), + agent_dir=agent_src, + out=out, + n=2, + split="test", + ) + (out / "result.json").unlink() + + # Corrupt binding.json + b_path = out / "trial-video-nextstep-0" / "binding.json" + assert b_path.is_file() + b_path.write_text("invalid-json{", encoding="utf-8") + with pytest.raises(TaskContractError, match=r"invalid binding\.json"): + read_partial_trials(out, "video-nextstep", require_binding=True) + + # Remove binding.json when required + b_path.unlink() + with pytest.raises(TaskContractError, match=r"missing binding\.json"): + read_partial_trials(out, "video-nextstep", require_binding=True) + + +def test_subgroup_worst_case_respects_minimize_direction() -> None: + from or_audit.eval.contracts import MetricDirection + + m_low = MetricOutcome( + id="max_pen", + kind=MetricKind.CONTINUOUS, + headline=True, + value=0.05, + direction=MetricDirection.MINIMIZE, + ) + m_high = MetricOutcome( + id="max_pen", + kind=MetricKind.CONTINUOUS, + headline=True, + value=0.25, + direction=MetricDirection.MINIMIZE, + ) + + trial_low = TrialRecord( + seed=0, + vector=TrialVector( + task_id="pen-task", + task_version="1", + agent_identity="agent", + seed=0, + gates=(), + metrics=(m_low,), + ), + trajectory=ProceduralTrace(()), + subgroups={"scanner": "scanner-A"}, + ) + trial_high = TrialRecord( + seed=1, + vector=TrialVector( + task_id="pen-task", + task_version="1", + agent_identity="agent", + seed=1, + gates=(), + metrics=(m_high,), + ), + trajectory=ProceduralTrace(()), + subgroups={"scanner": "scanner-B"}, + ) + + result = JobResult( + task_id="pen-task", + task_version="1", + agent_identity="agent", + world_pin="pin", + interface_id="intf", + interaction_mode="single-turn", + task_digest="td", + agent_digest="ad", + n=2, + headline="max_pen", + trials=(trial_low, trial_high), + headline_true=0, + headline_false=0, + headline_unassessable=0, + any_gate_failed=0, + ) + + data = scorecard_data(result) + assert len(data["subgroups"]) == 2 + # For a minimize metric, worst-case is the HIGHEST value (scanner-B: 0.25) + assert data["worst_case"]["subgroup"] == "scanner-B" + assert data["worst_case"]["rate"] == 0.25 + + +def test_subgroup_neutral_metric_produces_no_worst_case() -> None: + from or_audit.eval.contracts import MetricDirection + + m0 = MetricOutcome( + id="neutral_metric", + kind=MetricKind.CONTINUOUS, + headline=True, + value=1.0, + direction=MetricDirection.NEUTRAL, + ) + m1 = MetricOutcome( + id="neutral_metric", + kind=MetricKind.CONTINUOUS, + headline=True, + value=5.0, + direction=MetricDirection.NEUTRAL, + ) + t0 = TrialRecord( + seed=0, + vector=TrialVector( + task_id="t", task_version="1", agent_identity="a", seed=0, gates=(), metrics=(m0,) + ), + subgroups={"anatomy": "bifurcation"}, + ) + t1 = TrialRecord( + seed=1, + vector=TrialVector( + task_id="t", task_version="1", agent_identity="a", seed=1, gates=(), metrics=(m1,) + ), + subgroups={"anatomy": "straight"}, + ) + result = JobResult( + task_id="t", + task_version="1", + agent_identity="a", + world_pin="pin", + interface_id="intf", + interaction_mode="single-turn", + task_digest="td", + agent_digest="ad", + n=2, + headline="neutral_metric", + trials=(t0, t1), + headline_true=0, + headline_false=0, + headline_unassessable=0, + any_gate_failed=0, + ) + data = scorecard_data(result) + assert len(data["subgroups"]) == 2 + # Neutral metric must NOT produce any worst case + assert data["worst_cases"] == {} + assert data["worst_case"] is None + + +def test_subgroups_multi_axis_ranks_within_each_axis() -> None: + m_pass = MetricOutcome( + id="m", + kind=MetricKind.BOOLEAN, + headline=True, + value=True, + direction=MetricDirection.MAXIMIZE, + ) + m_fail = MetricOutcome( + id="m", + kind=MetricKind.BOOLEAN, + headline=True, + value=False, + direction=MetricDirection.MAXIMIZE, + ) + + t0 = TrialRecord( + seed=0, + vector=TrialVector( + task_id="t", task_version="1", agent_identity="a", seed=0, gates=(), metrics=(m_pass,) + ), + subgroups={"anatomy": "bifurcation", "scanner": "siemens"}, + ) + t1 = TrialRecord( + seed=1, + vector=TrialVector( + task_id="t", task_version="1", agent_identity="a", seed=1, gates=(), metrics=(m_fail,) + ), + subgroups={"anatomy": "straight", "scanner": "ge"}, + ) + result = JobResult( + task_id="t", + task_version="1", + agent_identity="a", + world_pin="pin", + interface_id="intf", + interaction_mode="single-turn", + task_digest="td", + agent_digest="ad", + n=2, + headline="m", + trials=(t0, t1), + headline_true=1, + headline_false=1, + headline_unassessable=0, + any_gate_failed=0, + ) + data = scorecard_data(result) + assert "anatomy" in data["worst_cases"] + assert "scanner" in data["worst_cases"] + assert data["worst_cases"]["anatomy"]["subgroup"] == "straight" + assert data["worst_cases"]["scanner"]["subgroup"] == "ge" + # Legacy worst_case is None when multiple axes exist + assert data["worst_case"] is None + + +def test_clustered_bootstrap_rejects_empty_cluster() -> None: + clusters = {"patient-A": [1.0, 2.0], "patient-B": []} + with pytest.raises(TaskContractError, match="requires non-empty observations"): + clustered_bootstrap_mean_ci(clusters) + + +def test_clustered_bootstrap_unequal_size_discriminates_from_pooled() -> None: + # 4 clusters with 1 observation of 100.0 (cluster mean = 100.0) + # 1 cluster with 20 observations of 0.0 each (cluster mean = 0.0) + clusters = { + "patient-A1": [100.0], + "patient-A2": [100.0], + "patient-A3": [100.0], + "patient-A4": [100.0], + "patient-B1": [0.0] * 20, + } + # Cluster-level mean estimand is (100 + 100 + 100 + 100 + 0) / 5 = 80.0 + low_cl, high_cl = clustered_bootstrap_mean_ci(clusters, confidence=0.95, seed=42) + assert low_cl >= 40.0 + assert high_cl <= 100.0 + + # Raw pooled bootstrap on all 24 observations: 4 * 100.0 + 20 * 0.0 -> pooled mean is 16.67! + pooled_values = [100.0] * 4 + [0.0] * 20 + _low_pool, high_pool = bootstrap_mean_ci(pooled_values, confidence=0.95, seed=42) + assert high_pool < 35.0 + assert high_pool < low_cl # Completely discriminates cluster-mean weighting! + + +def test_subgroup_no_assessment_row_reports_na_and_excluded_from_worst_case() -> None: + # Subgroup A: all trials unassessable (value=None) + # Subgroup B: 1 pass, 1 fail + m_unassessable = MetricOutcome( + id="m", + kind=MetricKind.BOOLEAN, + headline=True, + value=None, + direction=MetricDirection.MAXIMIZE, + ) + m_pass = MetricOutcome( + id="m", + kind=MetricKind.BOOLEAN, + headline=True, + value=True, + direction=MetricDirection.MAXIMIZE, + ) + m_fail = MetricOutcome( + id="m", + kind=MetricKind.BOOLEAN, + headline=True, + value=False, + direction=MetricDirection.MAXIMIZE, + ) + + t_un1 = TrialRecord( + seed=0, + vector=TrialVector( + task_id="t", + task_version="1", + agent_identity="a", + seed=0, + gates=(), + metrics=(m_unassessable,), + ), + subgroups={"anatomy": "unassessed-group"}, + ) + t_un2 = TrialRecord( + seed=1, + vector=TrialVector( + task_id="t", + task_version="1", + agent_identity="a", + seed=1, + gates=(), + metrics=(m_unassessable,), + ), + subgroups={"anatomy": "unassessed-group"}, + ) + t_b1 = TrialRecord( + seed=2, + vector=TrialVector( + task_id="t", task_version="1", agent_identity="a", seed=2, gates=(), metrics=(m_pass,) + ), + subgroups={"anatomy": "assessed-group"}, + ) + t_b2 = TrialRecord( + seed=3, + vector=TrialVector( + task_id="t", task_version="1", agent_identity="a", seed=3, gates=(), metrics=(m_fail,) + ), + subgroups={"anatomy": "assessed-group"}, + ) + + result = JobResult( + task_id="t", + task_version="1", + agent_identity="a", + world_pin="pin", + interface_id="intf", + interaction_mode="single-turn", + task_digest="td", + agent_digest="ad", + n=4, + headline="m", + trials=(t_un1, t_un2, t_b1, t_b2), + headline_true=1, + headline_false=1, + headline_unassessable=2, + any_gate_failed=0, + ) + + data = scorecard_data(result) + assert len(data["subgroups"]) == 2 + sg_un = next(sg for sg in data["subgroups"] if sg["subgroup"] == "unassessed-group") + sg_b = next(sg for sg in data["subgroups"] if sg["subgroup"] == "assessed-group") + + assert sg_un["assessed"] == 0 + assert sg_un["unassessable"] == 2 + assert sg_un["rate"] is None + assert sg_un["ci_95"] is None + + assert sg_b["assessed"] == 2 + assert sg_b["unassessable"] == 0 + assert sg_b["rate"] == 0.5 + + # Worst-case must be assessed-group (rate=0.5), NEVER unassessed-group! + assert data["worst_case"]["subgroup"] == "assessed-group" + assert data["worst_case"]["rate"] == 0.5 + + md = render_markdown(result) + assert "| unassessed-group | 2 | 0 | 2 | n/a | n/a |" in md + assert "**Worst-case subgroup:** `anatomy=assessed-group`" in md + + +def test_continuous_subgroup_pass_counts_assessed_only() -> None: + m_val1 = MetricOutcome( + id="metric", + kind=MetricKind.CONTINUOUS, + headline=True, + value=1.0, + direction=MetricDirection.MAXIMIZE, + ) + m_val2 = MetricOutcome( + id="metric", + kind=MetricKind.CONTINUOUS, + headline=True, + value=2.0, + direction=MetricDirection.MAXIMIZE, + ) + m_unassessable = MetricOutcome( + id="metric", + kind=MetricKind.CONTINUOUS, + headline=True, + value=None, + direction=MetricDirection.MAXIMIZE, + ) + + t0 = TrialRecord( + seed=0, + vector=TrialVector( + task_id="t", + task_version="1", + agent_identity="a", + seed=0, + gates=(GateOutcome(id="g", status=GateStatus.FAIL),), + metrics=(m_val1,), + ), + subgroups={"cohort": "group-A"}, + ) + t1 = TrialRecord( + seed=1, + vector=TrialVector( + task_id="t", + task_version="1", + agent_identity="a", + seed=1, + gates=(GateOutcome(id="g", status=GateStatus.PASS),), + metrics=(m_val2,), + ), + subgroups={"cohort": "group-A"}, + ) + t2 = TrialRecord( + seed=2, + vector=TrialVector( + task_id="t", + task_version="1", + agent_identity="a", + seed=2, + gates=(GateOutcome(id="g", status=GateStatus.PASS),), + metrics=(m_unassessable,), + ), + subgroups={"cohort": "group-A"}, + ) + t3 = TrialRecord( + seed=3, + vector=TrialVector( + task_id="t", + task_version="1", + agent_identity="a", + seed=3, + gates=(GateOutcome(id="g", status=GateStatus.PASS),), + metrics=(m_val1,), + ), + subgroups={"cohort": "group-B"}, + ) + + result = JobResult( + task_id="t", + task_version="1", + agent_identity="a", + world_pin="pin", + interface_id="intf", + interaction_mode="single-turn", + task_digest="td", + agent_digest="ad", + n=4, + headline="metric", + trials=(t0, t1, t2, t3), + headline_true=0, + headline_false=0, + headline_unassessable=1, + any_gate_failed=1, + ) + + data = scorecard_data(result) + sg_a = next(sg for sg in data["subgroups"] if sg["subgroup"] == "group-A") + assert sg_a["count"] == 3 + assert sg_a["assessed"] == 2 + assert sg_a["unassessable"] == 1 + # s_pass must be 1 (only trial 1 passed among assessed trials), NOT 2! + assert sg_a["pass"] == 1 + assert sg_a["rate"] == 1.5 + + md = render_markdown(result) + assert "| Estimate |" in md + assert "with estimate" in md + + +def test_scorecard_continuous_metric_cluster_activation_and_seed() -> None: + m0 = MetricOutcome(id="cont", kind=MetricKind.CONTINUOUS, headline=True, value=10.0) + m1 = MetricOutcome(id="cont", kind=MetricKind.CONTINUOUS, headline=True, value=20.0) + + # 1. Without declared patient/case bindings -> must be "bootstrap", NOT "clustered_bootstrap"! + t_unclustered_0 = TrialRecord( + seed=0, + vector=TrialVector( + task_id="t", task_version="1", agent_identity="a", seed=0, gates=(), metrics=(m0,) + ), + ) + t_unclustered_1 = TrialRecord( + seed=1, + vector=TrialVector( + task_id="t", task_version="1", agent_identity="a", seed=1, gates=(), metrics=(m1,) + ), + ) + res_unclustered = JobResult( + task_id="t", + task_version="1", + agent_identity="a", + world_pin="pin", + interface_id="intf", + interaction_mode="single-turn", + task_digest="td", + agent_digest="ad", + n=2, + headline="cont", + trials=(t_unclustered_0, t_unclustered_1), + headline_true=0, + headline_false=0, + headline_unassessable=0, + any_gate_failed=0, + ) + data_unclustered = scorecard_data(res_unclustered) + m_data = next(m for m in data_unclustered["metrics"] if m["headline"]) + assert m_data["ci_method"] == "bootstrap" + assert m_data["resampling_seed"] == 0 + assert m_data["draws"] == 1000 + assert m_data["confidence"] == 0.95 + + # 2. With declared patient clusters -> activates "clustered_bootstrap" + t_clustered_0 = TrialRecord( + seed=0, + vector=TrialVector( + task_id="t", task_version="1", agent_identity="a", seed=0, gates=(), metrics=(m0,) + ), + patient_id="patient-1", + ) + t_clustered_1 = TrialRecord( + seed=1, + vector=TrialVector( + task_id="t", task_version="1", agent_identity="a", seed=1, gates=(), metrics=(m1,) + ), + patient_id="patient-2", + ) + res_clustered = JobResult( + task_id="t", + task_version="1", + agent_identity="a", + world_pin="pin", + interface_id="intf", + interaction_mode="single-turn", + task_digest="td", + agent_digest="ad", + n=2, + headline="cont", + trials=(t_clustered_0, t_clustered_1), + headline_true=0, + headline_false=0, + headline_unassessable=0, + any_gate_failed=0, + independent_case_unit="patient", + ) + data_clustered = scorecard_data(res_clustered) + m_data_cl = next(m for m in data_clustered["metrics"] if m["headline"]) + assert m_data_cl["ci_method"] == "clustered_bootstrap" + assert m_data_cl["resampling_seed"] == 0 + assert m_data_cl["draws"] == 1000 + assert m_data_cl["confidence"] == 0.95 + + # 3. If independent_case_unit="patient" but a trial lacks patient_id -> raises TaskContractError + t_no_patient = TrialRecord( + seed=1, + vector=TrialVector( + task_id="t", task_version="1", agent_identity="a", seed=1, gates=(), metrics=(m1,) + ), + ) + res_missing = JobResult( + task_id="t", + task_version="1", + agent_identity="a", + world_pin="pin", + interface_id="intf", + interaction_mode="single-turn", + task_digest="td", + agent_digest="ad", + n=2, + headline="cont", + trials=(t_clustered_0, t_no_patient), + headline_true=0, + headline_false=0, + headline_unassessable=0, + any_gate_failed=0, + independent_case_unit="patient", + ) + with pytest.raises(TaskContractError, match="missing on trial"): + scorecard_data(res_missing)