reach.sweep

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

Execute catalog scaling sweeps and detect capacity knees.

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)

    scale: int
    catalog_id: str
    pass_rate: float
    pass_rate_interval: tuple[float, float]
    recall: float = 0.0
    recall_interval: tuple[float, float] = (0.0, 1.0)
    precision: float = 0.0
    precision_interval: tuple[float, float] = (0.0, 1.0)
    internal_precision: float = 0.0
    external_distractor_precision: float | None = None
    abstention_rate: float | None = None
    abstention_interval: tuple[float, float] | None = None
    f1_score: float = 0.0
    f1_interval: tuple[float, float] = (0.0, 1.0)
    in_scope_probes: int = 0
    negative_probes: int = 0
    disclosure_states: dict[str, int] = Field(default_factory=dict)
    delta_vs_baseline: float
    delta_context: float
    delta_shadowing: float
    probes_executed: int
    probes_failed: int = 0
    prompt_tokens_mean: float | None = None
    duration_ms_mean: float = 0.0

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)

    target_skill: str | None = None
    is_corpus_sweep: bool = False
    scales: tuple[int, ...]
    points: tuple[ScalingPoint, ...]
    knee_scale: int | None = None
    sla_90_scale: int | None = None
    sla_90_interpolated: float | None = None
    sla_85_scale: int | None = None
    sla_85_interpolated: float | None = None
    baseline_pass_rate: float
    final_pass_rate: float
    total_delta: float
    total_context_loss: float
    total_shadowing_loss: float
    noise_floor: float = 0.05
    total_corpus_skills: int = 0
    decomposition: DecompositionResult | None = None
    anchor_skills: tuple[str, ...] | None = None

    @model_validator(mode="after")
    def _validate_target_skill_for_mode(self) -> ScalingStudy:
        """Ensure targeted sweeps specify a target skill."""
        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)
        return self

bootstrap_f1_ci

bootstrap_f1_ci(
    results: Sequence[ProbeResult],
    truth: Mapping[str, str | None],
    installed_skills: set[str],
    iterations: int = 1000,
    seed: int = 42,
    target_skill: str | None = None,
) -> tuple[float, float]

Compute empirical bootstrap confidence interval for micro F1 score.

Source code in src/reach/sweep.py

def bootstrap_f1_ci(
    results: Sequence[ProbeResult],
    truth: Mapping[str, str | None],
    installed_skills: set[str],
    iterations: int = 1000,
    seed: int = 42,
    target_skill: str | None = None,
) -> tuple[float, float]:
    """Compute empirical bootstrap confidence interval for micro F1 score."""
    m = len(results)
    if m <= 0 or iterations <= 0:
        return (0.0, 1.0)

    outcomes = [
        _classify_probe_outcome(r, truth.get(r.query_id), installed_skills, target_skill)
        for r in results
    ]
    tp_arr = [int(tp) for tp, _, _ in outcomes]
    fp_arr = [int(fp) for _, fp, _ in outcomes]
    fn_arr = [int(fn) for _, _, fn in outcomes]

    rng = random.Random(seed)  # noqa: S311
    f1_boots: list[float] = []
    indices = list(range(m))

    for _ in range(iterations):
        sample_idx = [rng.choice(indices) for _ in range(m)]
        tp_s = sum(tp_arr[i] for i in sample_idx)
        fp_s = sum(fp_arr[i] for i in sample_idx)
        fn_s = sum(fn_arr[i] for i in sample_idx)
        denom = 2 * tp_s + fp_s + fn_s
        f1_boots.append(2.0 * tp_s / denom if denom > 0 else 0.0)

    f1_boots.sort()
    low_idx = max(0, int(iterations * 0.025))
    high_idx = min(int(iterations * 0.975), iterations - 1)
    return (round(f1_boots[low_idx], 4), round(f1_boots[high_idx], 4))

compute_scaling_noise_floor

compute_scaling_noise_floor(
    baseline_pass_rate: float,
    scaled_pass_rate: float,
    sample_size: int,
    confidence: float = DEFAULT_CONFIDENCE,
    noise_inflation: float = NOISE_INFLATION,
) -> float

Calculate minimum scaling pass-rate drop distinguishable from noise using diff.

Source code in src/reach/sweep.py

def compute_scaling_noise_floor(
    baseline_pass_rate: float,
    scaled_pass_rate: float,
    sample_size: int,
    confidence: float = DEFAULT_CONFIDENCE,
    noise_inflation: float = NOISE_INFLATION,
) -> float:
    """Calculate minimum scaling pass-rate drop distinguishable from noise using diff."""
    n = max(1, sample_size)
    control_se = math.sqrt(max(0.0, baseline_pass_rate * (1.0 - baseline_pass_rate)) / n)
    treatment_se = math.sqrt(max(0.0, scaled_pass_rate * (1.0 - scaled_pass_rate)) / n)
    floor = diff_noise_floor(control_se, treatment_se, confidence, noise_inflation)
    return max(0.01, floor)

compute_sla_crossings

compute_sla_crossings(
    points: Sequence[ScalingPoint], threshold: float
) -> tuple[int | None, float | None]

Compute discrete conservative scale and log-linear continuous crossing for an SLA target.

Source code in src/reach/sweep.py

def compute_sla_crossings(
    points: Sequence[ScalingPoint],
    threshold: float,
) -> tuple[int | None, float | None]:
    """Compute discrete conservative scale and log-linear continuous crossing for an SLA target."""
    if not points:
        return None, None

    ordered = sorted(points, key=lambda p: p.scale)
    discrete_k: int | None = None
    interp_k: float | None = None

    for i in range(len(ordered) - 1):
        p1, p2 = ordered[i], ordered[i + 1]
        if p1.f1_score >= threshold > p2.f1_score:
            discrete_k = p1.scale
            interp_k = _log_interpolate_scale(p1, p2, threshold)
            break

    if discrete_k is None:
        qualifying = [p.scale for p in ordered if p.f1_score >= threshold]
        discrete_k = max(qualifying) if qualifying else None
        for i in range(len(ordered) - 1):
            p1, p2 = ordered[i], ordered[i + 1]
            if p1.f1_score <= threshold < p2.f1_score:
                interp_k = _log_interpolate_scale(p1, p2, threshold)
                break

    if interp_k is None and discrete_k is not None:
        interp_k = float(discrete_k)

    return discrete_k, interp_k

find_kneedle_knee

find_kneedle_knee(
    scales: Sequence[int],
    pass_rates: Sequence[float],
    noise_floor: float = 0.1,
) -> int | None

Identify the inflection knee scale k* using normalized log-scale Kneedle curvature.

Source code in src/reach/sweep.py

def find_kneedle_knee(
    scales: Sequence[int],
    pass_rates: Sequence[float],
    noise_floor: float = 0.10,
) -> int | None:
    """Identify the inflection knee scale k* using normalized log-scale Kneedle curvature."""
    if (
        len(scales) < _MIN_KNEE_POINTS
        or len(scales) != len(pass_rates)
        or (max(pass_rates) - min(pass_rates)) <= noise_floor
    ):
        return None

    points = sorted(zip(scales, pass_rates, strict=True), key=lambda p: p[0])
    k_vals = [p[0] for p in points]
    y_vals = [p[1] for p in points]

    if (y_vals[0] - y_vals[-1]) <= noise_floor:
        return None

    log_k = [math.log(k) for k in k_vals]
    min_log = log_k[0]
    max_log = log_k[-1]
    log_range = max_log - min_log
    min_y = min(y_vals)
    max_y = max(y_vals)
    y_range = max_y - min_y

    if log_range <= 0.0 or y_range <= 0.0:
        return None

    norm_x = [(lk - min_log) / log_range for lk in log_k]
    norm_y = [(y - min_y) / y_range for y in y_vals]

    y0 = norm_y[0]
    y_end = norm_y[-1]
    diffs_below = [(y0 + x * (y_end - y0)) - y for x, y in zip(norm_x, norm_y, strict=True)]
    diffs_above = [y - (y0 + x * (y_end - y0)) for x, y in zip(norm_x, norm_y, strict=True)]

    max_below = max(diffs_below[1:-1])
    max_above = max(diffs_above[1:-1])
    min_prominence = max(0.05, (noise_floor * 0.5) / y_range)

    if max_above >= max_below and max_above > min_prominence:
        knee_idx = 1 + diffs_above[1:-1].index(max_above)
        return k_vals[knee_idx]
    if max_below > min_prominence:
        knee_idx = 1 + diffs_below[1:-1].index(max_below)
        return k_vals[knee_idx]
    return None

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,
    early_stop: bool | 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,
    early_stop: bool | None = None,
) -> ScalingStudy:
    """Execute multi-scale catalog evaluation sweep and return scaling analysis."""
    effective_config = _prepare_sweep_config(config, attempts, early_stop)
    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
    requested_scales = scales if scales is not None else effective_config.study.scales
    actual_scales = resolve_sweep_scales(len(resolved_skills), requested_scales)

    target, catalogs, resolved_query_set, corpus_plan, resolved_anchors = _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,
    )

    resolved_runtime = runtime or build_runtime(effective_config.runtime)
    baseline_results: tuple[ProbeResult, ...] = ()
    points: list[ScalingPoint] = []
    final_decomp: DecompositionResult | None = None

    work_dir = effective_config.study.workdir
    temp_dir_obj: tempfile.TemporaryDirectory[str] | None = None
    if work_dir is None:
        temp_dir_obj = tempfile.TemporaryDirectory(prefix="reach_sweep_")
        work_dir = Path(temp_dir_obj.name)

    shared_outcome_cache: dict[Any, Any] = {}
    try:
        for scale, catalog in zip(actual_scales, catalogs, strict=True):
            safe_cat_id = catalog.id.replace(":", "_").replace("/", "_")
            scale_out = work_dir / f"sweep_{safe_cat_id}.jsonl"
            scale_config = effective_config.model_copy(
                update={
                    "study": effective_config.study.model_copy(
                        update={
                            "catalog": catalog.id,
                            "rescope": True,
                            "partial": True,
                            "workdir": work_dir,
                            "out": scale_out,
                        }
                    ),
                    "catalog": effective_config.catalog.model_copy(
                        update={"mode": CatalogMode.SWEEP}
                    ),
                }
            )

            scale_query_set = (
                corpus_plan.queries_for_scale(
                    catalog=catalog,
                    raw_query_set=raw_query_set,
                )
                if is_corpus and corpus_plan is not None
                else resolved_query_set
            )

            composed = Composition(
                config=scale_config,
                query_set=scale_query_set,
                catalog=catalog,
                skills=tuple(resolved_skills),
            )

            outcome = conduct(
                config=scale_config,
                runtime=resolved_runtime,
                composed=composed,
                allow_truncation=True,
                append_across_arms=True,
                workers=workers,
                outcome_cache=shared_outcome_cache,
            )

            point, decomp = _build_scaling_point(
                scale=scale,
                catalog_id=catalog.id,
                results=outcome.results,
                resolved_query_set=scale_query_set,
                baseline_results=baseline_results,
                installed_skills=set(catalog.skills),
                target_skill=target if not is_corpus else None,
                seed=effective_config.catalog.seed,
            )
            points.append(point)

            if scale == actual_scales[0]:
                baseline_results = outcome.results
            elif decomp is not None:
                final_decomp = decomp

            if (
                effective_config.study.early_stop
                and is_corpus
                and len(points) >= _MIN_EARLY_STOP_POINTS
                and point.f1_interval[1] < _EARLY_STOP_F1_THRESHOLD
            ):
                break

        effective_noise_floor = _compute_effective_noise_floor(
            noise_floor, points, len(baseline_results)
        )
        evaluated_scales = actual_scales[: len(points)]
        rate_curve = [p.f1_score for p in points] if is_corpus else [p.pass_rate for p in points]
        knee = find_kneedle_knee(evaluated_scales, rate_curve, noise_floor=effective_noise_floor)

        return _build_study_result(
            target=target,
            is_corpus=is_corpus,
            evaluated_scales=evaluated_scales,
            points=points,
            knee=knee,
            noise_floor=effective_noise_floor,
            total_skills=len(resolved_skills),
            decomp=final_decomp,
            anchor_skills=resolved_anchors,
        )
    finally:
        if temp_dir_obj is not None:
            temp_dir_obj.cleanup()