feat: Phase D5 explicit cohort subgroup validation, head-covered trial bindings, and clustered continuous bootstrap

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(

src/or_audit/eval/job.py

@@ -10,14 +10,15 @@
 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
 from or_audit.eval.agent import AgentPackage
 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,20 +504,47 @@ 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,
                 vector=vector,
                 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:
         raise TaskContractError(
             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
 
 

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())