reach.diff

A/B evaluation diffing between two arms against calibrated noise floors.

Cross two recorded evaluation arms and statistically test performance deltas.

Arm

Bases: BaseModel

Hold a single experimental arm and its associated Artifact.

Source code in src/reach/diff.py

class Arm(BaseModel):
    """Hold a single experimental arm and its associated Artifact."""

    model_config = ConfigDict(frozen=True)

    label: str
    artifact: Artifact
    source_queries_digest: str = ""

    @property
    def resident(self) -> tuple[str, ...]:
        """Return the sequence of skill names resident in this arm."""
        return tuple(entry.skill for entry in self.artifact.skills)

    @property
    def query_ids(self) -> frozenset[str]:
        """Return the set of query IDs evaluated in this arm."""
        return frozenset(record.query_id for record in self.artifact.queries)

query_ids property

query_ids: frozenset[str]

Return the set of query IDs evaluated in this arm.

resident property

resident: tuple[str, ...]

Return the sequence of skill names resident in this arm.

ArmSummary

Bases: BaseModel

Summarize provenance digests, catalog sizes, and accuracy metrics for an arm.

Source code in src/reach/diff.py

class ArmSummary(BaseModel):
    """Summarize provenance digests, catalog sizes, and accuracy metrics for an arm."""

    model_config = ConfigDict(frozen=True)

    label: str
    arm: str
    corpus_digest: str
    queries_digest: str
    catalog_id: str
    catalog_size: int = Field(ge=0)
    resident: tuple[str, ...] = ()
    attempts: int = Field(ge=1)
    probes: int = Field(ge=0)
    scored: int = Field(ge=0)
    top1_hits: int = Field(ge=0)
    top1_accuracy: float
    consistency: float
    standard_error: float | None = None
    entrypoint_accuracy: float = 0.0
    trajectory_reachability: float = 0.0
    step_efficiency: float = 0.0
    skill_f1: float = 0.0
    redundancy: float = 0.0

Comparison

Bases: BaseModel

Hold diff analysis between two arms across headline, skill, and query levels.

Source code in src/reach/diff.py

class Comparison(BaseModel):
    """Hold diff analysis between two arms across headline, skill, and query levels."""

    model_config = ConfigDict(frozen=True)

    factor: VaryFactor
    control: ArmSummary
    treatment: ArmSummary
    shared_queries: int = Field(gt=0)
    corroboration: Corroboration
    headline: Headline
    skills: tuple[SkillDelta, ...] = ()
    queries: tuple[QueryDelta, ...] = ()

    @property
    def deltas(self) -> tuple[QueryDelta, ...]:
        """Return query delta records."""
        return self.queries

    @property
    def separated(self) -> tuple[QueryDelta, ...]:
        """Return queries whose intervals showed significant divergence."""
        return tuple(delta for delta in self.queries if delta.real)

    @property
    def verdict(self) -> str:
        """Generate a concise headline summary verdict string for the comparison."""
        points = self.headline.delta * 100
        if self.headline.floor is None:
            return (
                f"top-1 moved {points:+.1f} points, and neither arm reported a "
                "spread to price it against: no verdict"
            )
        floor = self.headline.floor * 100
        if self.headline.real:
            direction = "rose" if points > 0 else "fell"
            return (
                f"top-1 {direction} {abs(points):.1f} points, clearing the "
                f"{floor:.1f}-point noise floor"
            )
        needed = self.headline.needed_probes
        owed = (
            f"; separating a delta this size needs about {needed} probes per arm"
            if needed is not None
            else ""
        )
        if separated := self.separated:
            named = ", ".join(delta.query_id for delta in separated[:ROSTER_SHOWN])
            rest = len(separated) - len(separated[:ROSTER_SHOWN])
            more = f" and {rest} more" if rest else ""
            return (
                f"top-1 moved {points:+.1f} points, inside the {floor:.1f}-point "
                f"noise floor, but {len(separated)} of {len(self.queries)} "
                f"queries separated on their own: {named}{more}{owed}"
            )
        return (
            f"top-1 moved {points:+.1f} points, inside the {floor:.1f}-point "
            f"noise floor: not an improvement{owed}"
        )

deltas property

deltas: tuple[QueryDelta, ...]

Return query delta records.

separated property

separated: tuple[QueryDelta, ...]

Return queries whose intervals showed significant divergence.

verdict property

verdict: str

Generate a concise headline summary verdict string for the comparison.

Corroboration

Bases: BaseModel

Record verification status of provenance changes between arms.

Source code in src/reach/diff.py

class Corroboration(BaseModel):
    """Record verification status of provenance changes between arms."""

    model_config = ConfigDict(frozen=True)

    factor: VaryFactor
    arm_moved: bool
    corpus_moved: bool
    added: tuple[str, ...] = ()
    removed: tuple[str, ...] = ()
    corroborated: bool
    reason: str

Headline

Bases: BaseModel

Summarize run-wide top-1 accuracy change and noise floor significance.

Source code in src/reach/diff.py

class Headline(BaseModel):
    """Summarize run-wide top-1 accuracy change and noise floor significance."""

    model_config = ConfigDict(frozen=True)

    control: float
    treatment: float
    confidence: float = Field(default=DEFAULT_CONFIDENCE, gt=0.0, lt=1.0)
    noise_inflation: float = Field(default=NOISE_INFLATION, gt=0.0)
    floor: float | None = None
    real: bool
    resolvable: float | None = None
    needed_probes: int | None = None

    @computed_field
    @property
    def delta(self) -> float:
        """Calculate treatment top-1 accuracy minus control top-1 accuracy."""
        return self.treatment - self.control

delta property

delta: float

Calculate treatment top-1 accuracy minus control top-1 accuracy.

QueryDelta

Bases: BaseModel

Record per-query hit rate differences and overlap significance between arms.

Source code in src/reach/diff.py

class QueryDelta(BaseModel):
    """Record per-query hit rate differences and overlap significance between arms."""

    model_config = ConfigDict(frozen=True)

    query_id: str
    kind: QueryKind | None = None
    control_hits: int = Field(ge=0)
    control_probes: int = Field(gt=0)
    control_rate: float
    control_interval: Interval
    treatment_hits: int = Field(ge=0)
    treatment_probes: int = Field(gt=0)
    treatment_rate: float
    treatment_interval: Interval
    real: bool

    @computed_field
    @property
    def delta(self) -> float:
        """Calculate treatment hit rate minus control hit rate."""
        return self.treatment_rate - self.control_rate

delta property

delta: float

Calculate treatment hit rate minus control hit rate.

SkillDelta

Bases: BaseModel

Record per-skill recall differences and overlap significance between arms.

Source code in src/reach/diff.py

class SkillDelta(BaseModel):
    """Record per-skill recall differences and overlap significance between arms."""

    model_config = ConfigDict(frozen=True)

    skill: str
    control_reached: int = Field(ge=0)
    control_probes: int = Field(gt=0)
    control_recall: float
    control_interval: Interval
    treatment_reached: int = Field(ge=0)
    treatment_probes: int = Field(gt=0)
    treatment_recall: float
    treatment_interval: Interval
    real: bool

    @computed_field
    @property
    def delta(self) -> float:
        """Calculate treatment recall minus control recall."""
        return self.treatment_recall - self.control_recall

delta property

delta: float

Calculate treatment recall minus control recall.

Survey

Bases: BaseModel

Survey and validate readiness of two runs for pairwise diff comparison.

Source code in src/reach/diff.py

class Survey(BaseModel):
    """Survey and validate readiness of two runs for pairwise diff comparison."""

    model_config = ConfigDict(frozen=True)

    factor: VaryFactor
    control_path: Path
    treatment_path: Path
    walls: tuple[Wall, ...] = ()
    control: Arm | None = Field(default=None, exclude=True)
    treatment: Arm | None = Field(default=None, exclude=True)

    @property
    def comparable(self) -> bool:
        """Return True if no validation barriers prevent comparison."""
        return not self.walls

    def cross(
        self,
        *,
        confidence: float = DEFAULT_CONFIDENCE,
        noise_inflation: float = NOISE_INFLATION,
        settings: DiffSettings | None = None,
    ) -> Comparison:
        """Perform comparison across validated arms, raising error if barriers exist."""
        if self.control is None or self.treatment is None:
            msg = (
                f"{self.control_path.name} and {self.treatment_path.name} cannot be "
                f"crossed: {len(self.walls)} validation barriers detected"
            )
            raise ValueError(msg)
        return diff_arms(
            self.control,
            self.treatment,
            self.factor,
            confidence=confidence,
            noise_inflation=noise_inflation,
            settings=settings,
        )

comparable property

comparable: bool

Return True if no validation barriers prevent comparison.

cross

cross(
    *,
    confidence: float = DEFAULT_CONFIDENCE,
    noise_inflation: float = NOISE_INFLATION,
    settings: DiffSettings | None = None,
) -> Comparison

Perform comparison across validated arms, raising error if barriers exist.

Source code in src/reach/diff.py

def cross(
    self,
    *,
    confidence: float = DEFAULT_CONFIDENCE,
    noise_inflation: float = NOISE_INFLATION,
    settings: DiffSettings | None = None,
) -> Comparison:
    """Perform comparison across validated arms, raising error if barriers exist."""
    if self.control is None or self.treatment is None:
        msg = (
            f"{self.control_path.name} and {self.treatment_path.name} cannot be "
            f"crossed: {len(self.walls)} validation barriers detected"
        )
        raise ValueError(msg)
    return diff_arms(
        self.control,
        self.treatment,
        self.factor,
        confidence=confidence,
        noise_inflation=noise_inflation,
        settings=settings,
    )

VaryFactor

Bases: StrEnum

Specify the single experimental factor varied between compared runs.

Source code in src/reach/diff.py

class VaryFactor(StrEnum):
    """Specify the single experimental factor varied between compared runs."""

    DESCRIPTION = "description"
    RIVAL = "rival"
    SCOPE = "scope"

Wall

Bases: BaseModel

Represent an incompatibility barrier preventing comparison between two runs.

Source code in src/reach/diff.py

class Wall(BaseModel):
    """Represent an incompatibility barrier preventing comparison between two runs."""

    model_config = ConfigDict(frozen=True)

    where: str
    path: Path | None = None
    reason: str

diff_arms

diff_arms(
    control: Arm,
    treatment: Arm,
    factor: VaryFactor | str,
    *,
    confidence: float = DEFAULT_CONFIDENCE,
    noise_inflation: float = NOISE_INFLATION,
    settings: DiffSettings | None = None,
) -> Comparison

Compare two experimental arms across the specified variation factor.

Source code in src/reach/diff.py

def diff_arms(
    control: Arm,
    treatment: Arm,
    factor: VaryFactor | str,
    *,
    confidence: float = DEFAULT_CONFIDENCE,
    noise_inflation: float = NOISE_INFLATION,
    settings: DiffSettings | None = None,
) -> Comparison:
    """Compare two experimental arms across the specified variation factor."""
    if settings is not None:
        confidence = settings.confidence
        noise_inflation = settings.noise_inflation

    factor = VaryFactor(factor)

    if walls := pairing_walls(control, treatment):
        raise ValueError(walls[0].reason)

    shared = control.query_ids & treatment.query_ids
    return Comparison(
        factor=factor,
        control=_summarize(control),
        treatment=_summarize(treatment),
        shared_queries=len(shared),
        corroboration=_corroborate(factor, control, treatment),
        headline=_headline(control, treatment, confidence, noise_inflation),
        skills=_skill_deltas(control, treatment, confidence),
        queries=_query_deltas(control, treatment, confidence),
    )

diff_runs

diff_runs(
    control_path: Path | str,
    treatment_path: Path | str,
    factor: VaryFactor | str,
    *,
    queries_root: Path | str | None = None,
    control_corpus: Path | str | None = None,
    treatment_corpus: Path | str | None = None,
    control_label: str | None = None,
    treatment_label: str | None = None,
    queries: Path | str | None = None,
    filter_skill: Sequence[str] = (),
    filter_id: Sequence[str] = (),
    confidence: float = DEFAULT_CONFIDENCE,
    noise_inflation: float = NOISE_INFLATION,
) -> Comparison

Load and execute comparison between two results files across a factor.

Source code in src/reach/diff.py

def diff_runs(
    control_path: Path | str,
    treatment_path: Path | str,
    factor: VaryFactor | str,
    *,
    queries_root: Path | str | None = None,
    control_corpus: Path | str | None = None,
    treatment_corpus: Path | str | None = None,
    control_label: str | None = None,
    treatment_label: str | None = None,
    queries: Path | str | None = None,
    filter_skill: Sequence[str] = (),
    filter_id: Sequence[str] = (),
    confidence: float = DEFAULT_CONFIDENCE,
    noise_inflation: float = NOISE_INFLATION,
) -> Comparison:
    """Load and execute comparison between two results files across a factor."""
    left = Path(control_path).expanduser().resolve()
    right = Path(treatment_path).expanduser().resolve()
    if wall := _same_file(left, right):
        raise ValueError(wall.reason)
    return diff_arms(
        load_arm(
            left,
            queries_root=queries_root,
            corpus=control_corpus,
            label=control_label,
            queries=queries,
            filter_skill=filter_skill,
            filter_id=filter_id,
        ),
        load_arm(
            right,
            queries_root=queries_root,
            corpus=treatment_corpus,
            label=treatment_label,
            queries=queries,
            filter_skill=filter_skill,
            filter_id=filter_id,
        ),
        factor,
        confidence=confidence,
        noise_inflation=noise_inflation,
    )

load_arm

load_arm(
    results_path: Path | str,
    *,
    queries_root: Path | str | None = None,
    corpus: Path | str | None = None,
    label: str | None = None,
    queries: Path | str | None = None,
    filter_skill: Sequence[str] = (),
    filter_id: Sequence[str] = (),
) -> Arm

Load results and reconstruct the Artifact model for an experimental arm.

Source code in src/reach/diff.py

def load_arm(
    results_path: Path | str,
    *,
    queries_root: Path | str | None = None,
    corpus: Path | str | None = None,
    label: str | None = None,
    queries: Path | str | None = None,
    filter_skill: Sequence[str] = (),
    filter_id: Sequence[str] = (),
) -> Arm:
    """Load results and reconstruct the Artifact model for an experimental arm."""
    path = Path(results_path).expanduser()
    if not path.exists():
        msg = f"{path} does not exist"
        raise FileNotFoundError(msg)

    subset_qs = load_query_set(queries) if queries is not None else None
    slicing = bool(subset_qs is not None or filter_skill or filter_id)

    resolved_label = (
        label
        if label is not None
        else (
            path.name[: -len(ARTIFACT_SUFFIX)] if path.name.endswith(ARTIFACT_SUFFIX) else path.stem
        )
    )
    if not slicing and (unsliced := _try_read_unsliced_artifact(path)) is not None:
        return Arm(
            label=resolved_label,
            artifact=unsliced,
            source_queries_digest=unsliced.digests.queries_digest,
        )

    if slicing and not sidecar_path(path).exists():
        sibling = _resolve_sibling_jsonl(path)
        if sibling is None:
            msg = (
                f"{path} has no sibling .jsonl results and .config.json sidecar; "
                "query sub-slicing requires raw probe results to recompute exact scores"
            )
            raise ValueError(msg)
        path = sibling

    recorded = _read_valid_sidecar(path)
    config = _resolve_arm_config(path, recorded.config, queries_root, corpus)
    composed = compose(config)
    raw_results = load_results(path)

    if not slicing:
        return Arm(
            label=resolved_label,
            artifact=Artifact.assemble(
                composed,
                raw_results,
            ),
        )

    expected_q_digest = query_set_digest(composed.query_set)
    _cross_check(
        raw_results,
        Provenance(
            config_fingerprint=config.fingerprint,
            condition_digest=config.condition,
            corpus_digest=corpus_digest(composed.skills),
            queries_digest=expected_q_digest,
            tag=config.study.tag,
        ),
    )

    sliced_qs = filter_query_set(
        composed.query_set,
        subset=subset_qs,
        filter_skill=filter_skill,
        filter_id=filter_id,
    )
    sliced_digest = query_set_digest(sliced_qs)
    kept_ids = {q.id for q in sliced_qs.queries}
    sliced_results = [
        row.model_copy(update={"queries_digest": sliced_digest})
        for row in raw_results
        if row.query_id in kept_ids
    ]
    if not sliced_results:
        msg = f"{path} has no probe results matching the requested query slice"
        raise ValueError(msg)

    sliced_composition = Composition(
        config=config,
        query_set=sliced_qs,
        catalog=composed.catalog,
        skills=composed.skills,
    )
    return Arm(
        label=resolved_label,
        artifact=Artifact.assemble(
            sliced_composition,
            sliced_results,
        ),
        source_queries_digest=expected_q_digest,
    )

noise_floor

noise_floor(
    control: float,
    treatment: float,
    confidence: float = DEFAULT_CONFIDENCE,
    noise_inflation: float = NOISE_INFLATION,
) -> float

Calculate minimum top-1 delta distinguishable from noise at confidence.

Source code in src/reach/diff.py

def noise_floor(
    control: float,
    treatment: float,
    confidence: float = DEFAULT_CONFIDENCE,
    noise_inflation: float = NOISE_INFLATION,
) -> float:
    """Calculate minimum top-1 delta distinguishable from noise at confidence."""
    return critical_value(confidence) * noise_inflation * (control + treatment)

pairing_walls

pairing_walls(
    control: Arm, treatment: Arm
) -> tuple[Wall, ...]

Identify structural barriers preventing comparison between two loaded arms.

Source code in src/reach/diff.py

def pairing_walls(control: Arm, treatment: Arm) -> tuple[Wall, ...]:
    """Identify structural barriers preventing comparison between two loaded arms."""
    walls = []
    shared = control.query_ids & treatment.query_ids
    if control.query_ids != treatment.query_ids:
        only_control = sorted(control.query_ids - treatment.query_ids)
        only_treatment = sorted(treatment.query_ids - control.query_ids)
        walls.append(
            Wall(
                where=PAIRING,
                reason=(
                    f"refusing to compare {control.label} with {treatment.label}: "
                    f"they were scored on different queries ({len(only_control)} "
                    f"only in {control.label}, {len(only_treatment)} only in "
                    f"{treatment.label}). A comparison holds the query set fixed "
                    "and varies one factor; these vary the questions."
                ),
            ),
        )
    elif not shared:
        walls.append(
            Wall(
                where=PAIRING,
                reason=(
                    f"refusing to compare {control.label} with {treatment.label}: "
                    "neither arm was scored on any query"
                ),
            ),
        )

    digests = (
        control.artifact.digests.queries_digest,
        treatment.artifact.digests.queries_digest,
    )
    if not digests[0] or not digests[1]:
        walls.append(
            Wall(
                where=PAIRING,
                reason=(
                    f"refusing to compare {control.label} with {treatment.label}: "
                    "both arms must have recorded queries_digest"
                ),
            ),
        )
    elif digests[0] != digests[1]:
        walls.append(
            Wall(
                where=PAIRING,
                reason=(
                    f"refusing to compare {control.label} with {treatment.label}: "
                    f"their ground truth digests differ ({digests[0]} and "
                    f"{digests[1]}). The queries are the same but their labels are "
                    "not, so the delta would measure the reviewer rather than the "
                    "change."
                ),
            ),
        )
    elif (
        control.source_queries_digest
        and treatment.source_queries_digest
        and control.source_queries_digest != treatment.source_queries_digest
    ):
        walls.append(
            Wall(
                where=PAIRING,
                reason=(
                    f"refusing to compare {control.label} with {treatment.label}: "
                    f"their recorded source query set digests differ "
                    f"({control.source_queries_digest} and {treatment.source_queries_digest})."
                ),
            ),
        )
    return tuple(walls)

probes_to_resolve

probes_to_resolve(
    delta: float, noise_inflation: float = NOISE_INFLATION
) -> int

Calculate required probes per arm to reliably detect a given accuracy delta.

Source code in src/reach/diff.py

def probes_to_resolve(delta: float, noise_inflation: float = NOISE_INFLATION) -> int:
    """Calculate required probes per arm to reliably detect a given accuracy delta."""
    if not 0.0 < delta <= 1.0:
        msg = f"delta must lie in (0, 1], got {delta}"
        raise ValueError(msg)
    return required_probes(delta / noise_inflation)

survey_runs

survey_runs(
    control_path: Path | str,
    treatment_path: Path | str,
    factor: VaryFactor | str,
    *,
    queries_root: Path | str | None = None,
    control_corpus: Path | str | None = None,
    treatment_corpus: Path | str | None = None,
    control_label: str | None = None,
    treatment_label: str | None = None,
    queries: Path | str | None = None,
    filter_skill: Sequence[str] = (),
    filter_id: Sequence[str] = (),
) -> Survey

Survey and validate two run paths against all comparative requirements.

Source code in src/reach/diff.py

def survey_runs(
    control_path: Path | str,
    treatment_path: Path | str,
    factor: VaryFactor | str,
    *,
    queries_root: Path | str | None = None,
    control_corpus: Path | str | None = None,
    treatment_corpus: Path | str | None = None,
    control_label: str | None = None,
    treatment_label: str | None = None,
    queries: Path | str | None = None,
    filter_skill: Sequence[str] = (),
    filter_id: Sequence[str] = (),
) -> Survey:
    """Survey and validate two run paths against all comparative requirements."""
    factor = VaryFactor(factor)
    left = Path(control_path).expanduser().resolve()
    right = Path(treatment_path).expanduser().resolve()
    if wall := _same_file(left, right):
        return Survey(
            factor=factor,
            control_path=left,
            treatment_path=right,
            walls=(wall,),
        )

    control, control_wall = _survey_arm(
        CONTROL,
        left,
        queries_root=queries_root,
        corpus=control_corpus,
        label=control_label,
        queries=queries,
        filter_skill=filter_skill,
        filter_id=filter_id,
    )
    treatment, treatment_wall = _survey_arm(
        TREATMENT,
        right,
        queries_root=queries_root,
        corpus=treatment_corpus,
        label=treatment_label,
        queries=queries,
        filter_skill=filter_skill,
        filter_id=filter_id,
    )
    walls = [wall for wall in (control_wall, treatment_wall) if wall is not None]

    if control is not None and treatment is not None:
        walls.extend(pairing_walls(control, treatment))
    else:
        standing = "neither arm" if walls[1:] else "one of the two arms"
        walls.append(
            Wall(
                where=PAIRING,
                reason=(
                    f"not reached: {standing} could be rebuilt, so whether these "
                    "two were scored on the same questions is not yet knowable"
                ),
            ),
        )

    return Survey(
        factor=factor,
        control_path=left,
        treatment_path=right,
        walls=tuple(walls),
        control=control,
        treatment=treatment,
    )