reach.sweep

Execute multi-scale catalog evaluation sweeps to measure reachability decay and capacity knees.

Execute catalog scaling sweeps and detect capacity knees.

PairedTrialOutcomes

Bases: BaseModel

Represent paired McNemar discordant trial outcomes across sweep scales.

Source code in src/reach/sweep.py

class PairedTrialOutcomes(BaseModel):
    """Represent paired McNemar discordant trial outcomes across sweep scales."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    n10: Annotated[
        NonNegativeInt,
        Field(description="Probes successful at baseline but failed at scaled catalog"),
    ]
    n01: Annotated[
        NonNegativeInt,
        Field(description="Probes failed at baseline but successful at scaled catalog"),
    ]
    total_paired: Annotated[
        NonNegativeInt,
        Field(description="Total mutually executed probes in both baseline and scaled arms"),
    ]
    effective_paired: Annotated[
        NonNegativeInt | None,
        Field(description="Effective sample size adjusting for repeated attempts"),
    ] = None

    @model_validator(mode="after")
    def _validate_paired_totals(self) -> Self:
        """Enforce that total paired trials is at least the sum of discordant pairs."""
        if self.total_paired < (self.n10 + self.n01):
            msg = (
                f"total_paired ({self.total_paired}) cannot be less than the sum of "
                f"discordant pairs ({self.n10 + self.n01})"
            )
            raise ValueError(msg)
        return self

ReplicateCollisionDiagnostic

Bases: BaseModel

Identify a query whose routing outcome flips across catalog replicates at scale K.

Source code in src/reach/sweep.py

class ReplicateCollisionDiagnostic(BaseModel):
    """Identify a query whose routing outcome flips across catalog replicates at scale K."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    query_id: NonEmptyStr
    expected_skill: NonEmptyStr | None = None
    scale: PositiveInt
    failed_replicates: tuple[NonNegativeInt, ...]
    passed_replicates: tuple[NonNegativeInt, ...]
    invoked_distractors: tuple[NonEmptyStr, ...] = ()
    suspect_distractors: tuple[NonEmptyStr, ...] = ()

    @model_validator(mode="after")
    def _validate_replicate_partition(self) -> Self:
        """Ensure failed and passed replicate sets are non-empty and disjoint."""
        if not self.failed_replicates or not self.passed_replicates:
            msg = (
                "ReplicateCollisionDiagnostic requires at least one failed and one passed replicate"
            )
            raise ValueError(msg)
        if set(self.failed_replicates) & set(self.passed_replicates):
            msg = "failed_replicates and passed_replicates must be disjoint"
            raise ValueError(msg)
        return self

ScalingPoint

Bases: BaseModel

Represent evaluation outcomes and decomposition for a single catalog scale point.

Source code in src/reach/sweep.py

class ScalingPoint(BaseModel):
    """Represent evaluation outcomes and decomposition for a single catalog scale point."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    # 1. Identifiers
    scale: PositiveInt
    catalog_id: str

    # 2. Primary rates & intervals (alphabetical pairs)
    f1_score: UnitInterval = 0.0
    f1_interval: Interval = Field(default_factory=Interval.unit)
    pass_rate: UnitInterval
    pass_rate_interval: Interval
    precision: UnitInterval = 0.0
    precision_interval: Interval = Field(default_factory=Interval.unit)
    recall: UnitInterval = 0.0
    recall_interval: Interval = Field(default_factory=Interval.unit)

    # 3. Secondary rates
    abstention_rate: UnitInterval | None = None
    abstention_interval: Interval | None = None
    entrypoint_f1_score: UnitInterval | None = None
    entrypoint_pass_rate: UnitInterval | None = None
    external_distractor_precision: UnitInterval | None = None
    internal_precision: UnitInterval = 0.0

    # 4. Probe accounting counts
    probes_executed: NonNegativeInt
    in_scope_probes: NonNegativeInt = 0
    negative_probes: NonNegativeInt = 0
    probes_failed: NonNegativeInt = 0
    probes_errored: NonNegativeInt = 0

    # 5. Delta attribution
    delta_vs_baseline: float
    delta_abstention: float = 0.0
    delta_collision: float = 0.0
    delta_truncated: float = 0.0

    # 6. Disclosure states & telemetry
    disclosure_states: dict[str, int] = Field(default_factory=dict)
    prompt_tokens_mean: float | None = None
    duration_ms_mean: float = 0.0
    step_efficiency_mean: float = 0.0
    skill_f1_mean: float = 0.0

    @model_validator(mode="after")
    def _validate_probe_accounting(self) -> Self:
        """Ensure runtime errors and routing misses do not exceed executed probe bounds."""
        if self.probes_errored > self.probes_executed:
            msg = (
                f"probes_errored ({self.probes_errored}) cannot exceed "
                f"probes_executed ({self.probes_executed})"
            )
            raise ValueError(msg)
        valid_probes = self.probes_executed - self.probes_errored
        if self.probes_failed > valid_probes:
            msg = (
                f"probes_failed ({self.probes_failed}) cannot exceed "
                f"valid non-errored probes ({valid_probes})"
            )
            raise ValueError(msg)
        return self

    @property
    def all_probes_errored(self) -> bool:
        """Return True when at least one probe ran and every probe failed with a runtime error."""
        return self.probes_executed > 0 and self.probes_errored == self.probes_executed

all_probes_errored property

all_probes_errored: bool

Return True when at least one probe ran and every probe failed with a runtime error.

ScalingStudy

Bases: BaseModel

Represent multi-scale catalog scaling study and knee curvature analysis.

Source code in src/reach/sweep.py

class ScalingStudy(BaseModel):
    """Represent multi-scale catalog scaling study and knee curvature analysis."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    # 1. Corpus & runtime metadata
    agent: NonEmptyStr | None = None
    model: NonEmptyStr | None = None
    target_skill: NonEmptyStr | None = None
    is_corpus_sweep: bool = False
    catalog_replicates: PositiveInt = 1
    total_corpus_skills: NonNegativeInt = 0
    scales: tuple[PositiveInt, ...]

    # 2. Executive knee & drop findings
    knee_scale: PositiveInt | None = None
    knee_scale_interval: tuple[PositiveInt, PositiveInt] | None = None
    entrypoint_knee_scale: PositiveInt | None = None
    knee_metric: Literal["trajectory", "entrypoint"] = "trajectory"
    knee_upper_censored: bool = False
    cliff_probability: UnitInterval | None = None
    drop_probability: UnitInterval | None = None
    steepest_drop_scales: tuple[int, int] | None = None
    steepest_drop_delta: float | None = None

    # 3. Headline pass rates and loss deltas (flattened)
    baseline_pass_rate: UnitInterval = 0.0
    final_pass_rate: UnitInterval = 0.0
    delta_total: float = 0.0
    delta_abstention: float = 0.0
    delta_collision: float = 0.0
    delta_truncated: float = 0.0
    delta_total_interval: Interval = Field(default_factory=Interval.zero)
    delta_abstention_interval: Interval = Field(default_factory=Interval.zero)
    delta_collision_interval: Interval = Field(default_factory=Interval.zero)
    delta_truncated_interval: Interval = Field(default_factory=Interval.zero)
    noise_floor: float = 0.05
    sample_size: NonNegativeInt = 0
    baseline_interval: Interval = Field(default_factory=Interval.zero)
    scaled_interval: Interval = Field(default_factory=Interval.zero)

    # 4. Diagnostics & maps
    anchor_skills: tuple[str, ...] | None = None
    knee_scale_pmf: dict[PositiveInt, UnitInterval] | None = None
    paired_outcomes: PairedTrialOutcomes | None = None
    replicate_collisions: tuple[ReplicateCollisionDiagnostic, ...] = ()
    skill_icc: UnitInterval | None = None

    # 5. Points leaf collection at the bottom
    points: tuple[ScalingPoint, ...]

    @classmethod
    def load(cls, path: Path | str) -> Self:
        """Load and deserialize a ScalingStudy from a JSON file.

        Raises:
            FileNotFoundError: If the target file does not exist.
            IsADirectoryError: If the target path is a directory.
            ValueError: If the target path contains invalid study data.
        """
        p = Path(path).expanduser().resolve()
        if not p.exists():
            msg = f"Scaling study file not found: {p}"
            raise FileNotFoundError(msg)
        if p.is_dir():
            msg = f"Scaling study path is a directory, not a file: {p}"
            raise IsADirectoryError(msg)
        try:
            content = p.read_text(encoding="utf-8")
        except OSError as err:
            msg = f"Failed to read scaling study file at {p}: {err}"
            raise OSError(msg) from err

        try:
            return cls.model_validate_json(content)
        except (ValueError, ValidationError) as err:
            msg = f"Failed to parse scaling study JSON from {p}: {err}"
            raise ValueError(msg) from err

    def save(self, path: Path | str) -> Path:
        """Serialize the scaling study to a formatted JSON file atomically.

        Args:
            path: Target file path for the serialized JSON output.

        Returns:
            The resolved Path where the study was written.

        Raises:
            OSError: If directory creation or file writing fails.
        """
        target = Path(path).expanduser().resolve()
        target.parent.mkdir(parents=True, exist_ok=True)
        tmp_target = target.with_suffix(f"{target.suffix}.tmp.{os.getpid()}")
        try:
            tmp_target.write_text(self.model_dump_json(indent=2) + "\n", encoding="utf-8")
            tmp_target.replace(target)
        finally:
            if tmp_target.exists():
                tmp_target.unlink(missing_ok=True)
        return target

    @model_validator(mode="after")
    def _validate_target_skill_for_mode(self) -> Self:
        """Ensure targeted sweeps specify a target skill and statistical bounds are valid."""
        if not self.is_corpus_sweep and self.target_skill is None:
            msg = "Targeted scaling sweep requires target_skill to be specified."
            raise ValueError(msg)
        if (
            self.knee_scale_interval is not None
            and self.knee_scale_interval[0] > self.knee_scale_interval[1]
        ):
            msg = f"knee_scale_interval bounds are inverted: {self.knee_scale_interval}"
            raise ValueError(msg)
        if self.knee_upper_censored:
            if self.knee_scale_interval is None:
                msg = "knee_upper_censored requires knee_scale_interval to be set"
                raise ValueError(msg)
            if self.scales and self.knee_scale_interval[1] != self.scales[-1]:
                msg = (
                    f"knee_upper_censored upper bound must equal max scale "
                    f"{self.scales[-1]}, got {self.knee_scale_interval[1]}"
                )
                raise ValueError(msg)
        if (
            self.cliff_probability is not None
            and self.drop_probability is not None
            and self.cliff_probability > self.drop_probability + 1e-6
        ):
            msg = (
                f"cliff_probability ({self.cliff_probability}) cannot exceed "
                f"drop_probability ({self.drop_probability})"
            )
            raise ValueError(msg)
        if self.knee_scale_pmf:
            if self.scales and any(k not in self.scales for k in self.knee_scale_pmf):
                msg = f"knee_scale_pmf contains keys outside evaluated scales {self.scales}"
                raise ValueError(msg)
            pmf_sum = sum(self.knee_scale_pmf.values())
            if pmf_sum > _PMF_SUM_MAX:
                msg = f"knee_scale_pmf probabilities sum to {pmf_sum} > 1.0"
                raise ValueError(msg)
        return self

load classmethod

load(path: Path | str) -> Self

Load and deserialize a ScalingStudy from a JSON file.

Raises:

Type	Description
FileNotFoundError	If the target file does not exist.
IsADirectoryError	If the target path is a directory.
ValueError	If the target path contains invalid study data.

Source code in src/reach/sweep.py

@classmethod
def load(cls, path: Path | str) -> Self:
    """Load and deserialize a ScalingStudy from a JSON file.

    Raises:
        FileNotFoundError: If the target file does not exist.
        IsADirectoryError: If the target path is a directory.
        ValueError: If the target path contains invalid study data.
    """
    p = Path(path).expanduser().resolve()
    if not p.exists():
        msg = f"Scaling study file not found: {p}"
        raise FileNotFoundError(msg)
    if p.is_dir():
        msg = f"Scaling study path is a directory, not a file: {p}"
        raise IsADirectoryError(msg)
    try:
        content = p.read_text(encoding="utf-8")
    except OSError as err:
        msg = f"Failed to read scaling study file at {p}: {err}"
        raise OSError(msg) from err

    try:
        return cls.model_validate_json(content)
    except (ValueError, ValidationError) as err:
        msg = f"Failed to parse scaling study JSON from {p}: {err}"
        raise ValueError(msg) from err

save

save(path: Path | str) -> Path

Serialize the scaling study to a formatted JSON file atomically.

Parameters:

Name	Type	Description	Default
path	Path | str	Target file path for the serialized JSON output.	required

Returns:

Type	Description
Path	The resolved Path where the study was written.

Raises:

Type	Description
OSError	If directory creation or file writing fails.

Source code in src/reach/sweep.py

def save(self, path: Path | str) -> Path:
    """Serialize the scaling study to a formatted JSON file atomically.

    Args:
        path: Target file path for the serialized JSON output.

    Returns:
        The resolved Path where the study was written.

    Raises:
        OSError: If directory creation or file writing fails.
    """
    target = Path(path).expanduser().resolve()
    target.parent.mkdir(parents=True, exist_ok=True)
    tmp_target = target.with_suffix(f"{target.suffix}.tmp.{os.getpid()}")
    try:
        tmp_target.write_text(self.model_dump_json(indent=2) + "\n", encoding="utf-8")
        tmp_target.replace(target)
    finally:
        if tmp_target.exists():
            tmp_target.unlink(missing_ok=True)
    return target

run_scaling_sweep

run_scaling_sweep(
    config: RunConfig | None = None,
    target_skill: str | None = None,
    scales: Sequence[int] | None = None,
    anchor: int | Sequence[str] | str | None = None,
    runtime: AgentRuntime | None = None,
    rivals_share: float = 0.5,
    noise_floor: float | None = None,
    workers: int = 1,
    skills: Sequence[Skill] | None = None,
    query_set: QuerySet | None = None,
    attempts: int | None = None,
    allow_truncation: bool = True,
    *,
    bootstrap_iterations: int | None = None,
    seed: int | None = None,
    catalog_replicates: int | None = None,
    knee_metric: Literal["trajectory", "entrypoint"]
    | None = None,
    on_scale_complete: Callable[
        [int, int, ScalingPoint, ScalingStudy], None
    ]
    | None = None,
) -> ScalingStudy

Execute multi-scale catalog evaluation sweep and return scaling analysis.

Source code in src/reach/sweep.py

def run_scaling_sweep(
    config: RunConfig | None = None,
    target_skill: str | None = None,
    scales: Sequence[int] | None = None,
    anchor: int | Sequence[str] | str | None = None,
    runtime: AgentRuntime | None = None,
    rivals_share: float = 0.5,
    noise_floor: float | None = None,
    workers: int = 1,
    skills: Sequence[Skill] | None = None,
    query_set: QuerySet | None = None,
    attempts: int | None = None,
    allow_truncation: bool = True,
    *,
    bootstrap_iterations: int | None = None,
    seed: int | None = None,
    catalog_replicates: int | None = None,
    knee_metric: Literal["trajectory", "entrypoint"] | None = None,
    on_scale_complete: Callable[[int, int, ScalingPoint, ScalingStudy], None] | None = None,
) -> ScalingStudy:
    """Execute multi-scale catalog evaluation sweep and return scaling analysis."""
    effective_config = _prepare_sweep_config(
        config,
        attempts=attempts,
        bootstrap_iterations=bootstrap_iterations,
        seed=seed,
        catalog_replicates=catalog_replicates,
        knee_metric=knee_metric,
    )
    resolved_skills = (
        list(skills) if skills is not None else load_skills(effective_config.require_skills())
    )
    if len(resolved_skills) <= 1:
        msg = f"Scaling sweep requires at least 2 skills in corpus, got {len(resolved_skills)}"
        raise ValueError(msg)

    raw_query_set = (
        query_set if query_set is not None else load_query_set(effective_config.require_queries())
    )

    is_corpus = target_skill is None
    actual_scales = resolve_sweep_scales(
        len(resolved_skills),
        scales if scales is not None else effective_config.study.scales,
    )

    replicate_setups = [
        _setup_sweep_execution(
            is_corpus=is_corpus,
            target_skill=target_skill,
            anchor=anchor,
            resolved_skills=resolved_skills,
            actual_scales=actual_scales,
            raw_query_set=raw_query_set,
            effective_config=effective_config,
            rivals_share=rivals_share,
            seed_override=(
                effective_config.catalog.seed
                if rep_idx == 0
                else (effective_config.catalog.seed or effective_config.study.bootstrap_seed or 42)
                + rep_idx
            ),
        )
        for rep_idx in range(effective_config.study.catalog_replicates)
    ]
    primary_setup = replicate_setups[0]
    resolved_runtime = runtime if runtime is not None else build_runtime(effective_config.runtime)
    try:
        if not allow_truncation and resolved_runtime.rations_catalog:
            for setup in replicate_setups:
                for cat in setup.catalogs:
                    validate_catalog_fit(
                        resolved_runtime,
                        cat,
                        resolved_skills,
                        allow_truncation=False,
                    )
        baseline_results: tuple[ProbeResult, ...] = ()
        latest_results: tuple[ProbeResult, ...] = ()
        points: list[ScalingPoint] = []
        final_decomp: DecompositionResult | None = None
        work_dir, temp_dir_obj = _resolve_sweep_work_dir(effective_config.study.workdir)

        ctx = _SweepContext(
            effective_config=effective_config,
            work_dir=work_dir,
            corpus_plan=primary_setup.corpus_plan,
            raw_query_set=raw_query_set,
            resolved_query_set=primary_setup.query_set,
            resolved_skills=resolved_skills,
            raw_query_map={q.query_id: q for q in raw_query_set.queries},
            is_corpus=is_corpus,
        )

        shared_outcome_cache: dict[Any, Any] = {}
        scale_query_sums: dict[int, dict[str, _QueryOutcome]] = {}
        replicates_by_scale: dict[int, tuple[_ScaleReplicateRecord, ...]] = {}
        scoped_target = primary_setup.target if not is_corpus else None

        def _take_snapshot() -> ScalingStudy:
            return _snapshot_scaling_study(
                ctx=ctx,
                primary_setup=primary_setup,
                agent=(resolved_runtime.name or "").strip() or None,
                model=(resolved_runtime.model or "").strip() or None,
                actual_scales=actual_scales,
                points=points,
                noise_floor=noise_floor,
                baseline_results=baseline_results,
                latest_results=latest_results,
                decomp=final_decomp,
                scale_query_sums=scale_query_sums,
                replicates_by_scale=replicates_by_scale,
            )

        for step_idx, scale in enumerate(actual_scales, start=1):
            (
                combined_results,
                installed_union,
                primary_query_set,
                scale_replicates,
            ) = _execute_scale_replicates(
                step_idx=step_idx,
                scale_workers=max(1, workers),
                replicate_setups=replicate_setups,
                ctx=ctx,
                resolved_runtime=resolved_runtime,
                allow_truncation=allow_truncation,
                shared_outcome_cache=shared_outcome_cache,
                target=primary_setup.target,
            )
            replicates_by_scale[scale] = scale_replicates
            point, decomp = _build_scaling_point(
                scale=scale,
                catalog_id=primary_setup.catalogs[step_idx - 1].id,
                results=combined_results,
                resolved_query_set=primary_query_set,
                baseline_results=baseline_results,
                installed_skills=installed_union,
                target_skill=scoped_target,
                seed=effective_config.catalog.seed,
            )
            points.append(point)
            latest_results = combined_results
            scale_query_map = {q.query_id: q for q in primary_query_set.queries}
            truth_expected = {q.query_id: q.expected_skill for q in primary_query_set.queries}
            scale_query_sums[scale] = _extract_query_outcomes(
                combined_results,
                truth_expected,
                installed_union,
                target_skill=scoped_target,
                queries_by_id=scale_query_map,
                trajectory=(effective_config.study.knee_metric == "trajectory"),
            )

            if scale == actual_scales[0]:
                baseline_results = combined_results
            elif decomp is not None:
                final_decomp = decomp

            if on_scale_complete is not None:
                on_scale_complete(step_idx, len(actual_scales), point, _take_snapshot())

            if step_idx == 1 and point.all_probes_errored:
                break

        result = _take_snapshot()
        if temp_dir_obj is not None:
            temp_dir_obj.cleanup()
        return result
    finally:
        if runtime is None:
            resolved_runtime.cleanup()