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: Annotated[int, Field(ge=0)]
    resident: tuple[str, ...] = ()
    attempts: Annotated[int, Field(ge=1)]
    probes: Annotated[int, Field(ge=0)]
    scored: Annotated[int, Field(ge=0)]
    top1_hits: Annotated[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: Annotated[float | None, Field(ge=0.0, le=1.0)] = None
    skill_f1: Annotated[float | None, Field(ge=0.0, le=1.0)] = None
    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: Annotated[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: Annotated[float, Field(gt=0.0, lt=1.0)] = DEFAULT_CONFIDENCE
    noise_inflation: Annotated[float, Field(gt=0.0)] = NOISE_INFLATION
    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: Annotated[int, Field(ge=0)]
    control_probes: Annotated[int, Field(gt=0)]
    control_rate: float
    control_interval: Interval
    treatment_hits: Annotated[int, Field(ge=0)]
    treatment_probes: Annotated[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: Annotated[int, Field(ge=0)]
    control_probes: Annotated[int, Field(gt=0)]
    control_recall: float
    control_interval: Interval
    treatment_reached: Annotated[int, Field(ge=0)]
    treatment_probes: Annotated[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.

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."""

    AGENT = "agent"
    DESCRIPTION = "description"
    MODEL = "model"
    RIVAL = "rival"
    SCOPE = "scope"

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.query_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,
    )