reach.optimize

Closed-loop skill description optimization using candidate synthesis and empirical probes.

Closed-loop skill description optimization using diagnostic findings and empirical probes.

CandidateOrigin

Bases: StrEnum

Origin source of synthesized description candidate.

Source code in src/reach/optimize.py

class CandidateOrigin(StrEnum):
    """Origin source of synthesized description candidate."""

    DISCLAIMER = "disclaimer"
    HEURISTIC = "heuristic"
    LLM = "llm"

CandidateProbeTally

Bases: BaseModel

Encapsulate empirical probe counts and derived routing metrics for a candidate.

Source code in src/reach/optimize.py

class CandidateProbeTally(BaseModel):
    """Encapsulate empirical probe counts and derived routing metrics for a candidate."""

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

    triggers: int = Field(default=0, ge=0)
    trajectory_triggers: int = Field(default=0, ge=0)
    positive_queries: int = Field(default=0, ge=0)
    correct_count: int = Field(default=0, ge=0)
    misroutes: int = Field(default=0, ge=0)
    total_queries: int = Field(default=0, ge=0)
    failed_queries: tuple[str, ...] = ()
    failed_trajectory_queries: tuple[str, ...] = ()
    misrouted_queries: tuple[str, ...] = ()

    @computed_field  # type: ignore[prop-decorator]
    @property
    def recall(self) -> float:
        """Compute empirical entrypoint recall across positive queries."""
        return round(
            (self.triggers / self.positive_queries) if self.positive_queries > 0 else 1.0,
            4,
        )

    @computed_field  # type: ignore[prop-decorator]
    @property
    def trajectory_recall(self) -> float:
        """Compute empirical trajectory recall across positive queries."""
        effective_traj = max(self.triggers, self.trajectory_triggers)
        return round(
            (effective_traj / self.positive_queries) if self.positive_queries > 0 else 1.0,
            4,
        )

    @computed_field  # type: ignore[prop-decorator]
    @property
    def accuracy(self) -> float:
        """Compute empirical overall accuracy across executed probes."""
        return round(
            (self.correct_count / self.total_queries) if self.total_queries > 0 else 1.0,
            4,
        )

    @computed_field  # type: ignore[prop-decorator]
    @property
    def misroute_rate(self) -> float:
        """Compute empirical misroute rate across executed probes."""
        return round(
            (self.misroutes / self.total_queries) if self.total_queries > 0 else 0.0,
            4,
        )

    def paired_delta_recall(
        self,
        *,
        baseline_recall: float,
        baseline_hits_by_id: dict[str, bool] | None,
        queries_to_run: Sequence[Query],
        target_name: str,
    ) -> float:
        """Compute paired delta recall against baseline hits on the same positive query subset."""
        return _compute_paired_delta(
            self.recall,
            baseline_recall,
            baseline_hits_by_id,
            queries_to_run,
            target_name,
        )

    def paired_delta_trajectory_recall(
        self,
        *,
        baseline_trajectory_recall: float,
        baseline_traj_hits_by_id: dict[str, bool] | None,
        queries_to_run: Sequence[Query],
        target_name: str,
    ) -> float:
        """Compute paired delta trajectory recall against baseline trajectory hits."""
        return _compute_paired_delta(
            self.trajectory_recall,
            baseline_trajectory_recall,
            baseline_traj_hits_by_id,
            queries_to_run,
            target_name,
        )

accuracy property

accuracy: float

Compute empirical overall accuracy across executed probes.

misroute_rate property

misroute_rate: float

Compute empirical misroute rate across executed probes.

recall property

recall: float

Compute empirical entrypoint recall across positive queries.

trajectory_recall property

trajectory_recall: float

Compute empirical trajectory recall across positive queries.

paired_delta_recall

paired_delta_recall(
    *,
    baseline_recall: float,
    baseline_hits_by_id: dict[str, bool] | None,
    queries_to_run: Sequence[Query],
    target_name: str,
) -> float

Compute paired delta recall against baseline hits on the same positive query subset.

Source code in src/reach/optimize.py

def paired_delta_recall(
    self,
    *,
    baseline_recall: float,
    baseline_hits_by_id: dict[str, bool] | None,
    queries_to_run: Sequence[Query],
    target_name: str,
) -> float:
    """Compute paired delta recall against baseline hits on the same positive query subset."""
    return _compute_paired_delta(
        self.recall,
        baseline_recall,
        baseline_hits_by_id,
        queries_to_run,
        target_name,
    )

paired_delta_trajectory_recall

paired_delta_trajectory_recall(
    *,
    baseline_trajectory_recall: float,
    baseline_traj_hits_by_id: dict[str, bool] | None,
    queries_to_run: Sequence[Query],
    target_name: str,
) -> float

Compute paired delta trajectory recall against baseline trajectory hits.

Source code in src/reach/optimize.py

def paired_delta_trajectory_recall(
    self,
    *,
    baseline_trajectory_recall: float,
    baseline_traj_hits_by_id: dict[str, bool] | None,
    queries_to_run: Sequence[Query],
    target_name: str,
) -> float:
    """Compute paired delta trajectory recall against baseline trajectory hits."""
    return _compute_paired_delta(
        self.trajectory_recall,
        baseline_trajectory_recall,
        baseline_traj_hits_by_id,
        queries_to_run,
        target_name,
    )

IterationRecord

Bases: BaseModel

Record candidate outcomes and probe scores for a single optimization round.

Source code in src/reach/optimize.py

class IterationRecord(BaseModel):
    """Record candidate outcomes and probe scores for a single optimization round."""

    model_config = ConfigDict(frozen=True)

    iteration: Annotated[int, Field(ge=1)]
    candidates: tuple[OptimizationCandidate, ...]
    best_candidate: OptimizationCandidate
    failed_queries: tuple[str, ...] = ()
    misrouted_queries: tuple[str, ...] = ()
    test_evaluated: bool = False

OptimizationCandidate

Bases: BaseModel

Represent a generated description rewrite and its empirical performance.

Source code in src/reach/optimize.py

class OptimizationCandidate(BaseModel):
    """Represent a generated description rewrite and its empirical performance."""

    model_config = ConfigDict(frozen=True)

    accuracy: UnitMetric = 0.0
    delta_recall: DeltaMetric = 0.0
    delta_trajectory_recall: DeltaMetric = 0.0
    description: str
    lint_clean: bool = True
    filtered_out: bool = False
    filter_reason: str = ""
    misroute_rate: UnitMetric = 0.0
    rationale: str = ""
    origin: CandidateOrigin = CandidateOrigin.HEURISTIC
    recall: UnitMetric = 0.0
    trajectory_recall: UnitMetric = 0.0
    test_recall: UnitMetric | None = None
    test_trajectory_recall: UnitMetric | None = None
    test_accuracy: UnitMetric | None = None
    test_misroute_rate: UnitMetric | None = None
    failed_queries: tuple[str, ...] = ()
    failed_trajectory_queries: tuple[str, ...] = ()
    misrouted_queries: tuple[str, ...] = ()

    def mark_filtered(self, reason: str = "") -> OptimizationCandidate:
        """Return a copy marked as filtered out by static lint rules."""
        return self.model_copy(
            update={"lint_clean": False, "filtered_out": True, "filter_reason": reason}
        )

    def unfiltered(self) -> OptimizationCandidate:
        """Return a copy marked as passing static lint rules."""
        return self.model_copy(
            update={"lint_clean": True, "filtered_out": False, "filter_reason": ""}
        )

    def with_train_metrics(
        self,
        tally: CandidateProbeTally,
        *,
        delta_recall: float,
        delta_trajectory_recall: float = 0.0,
    ) -> OptimizationCandidate:
        """Return a copy populated with training probe metrics from a CandidateProbeTally."""
        return self.model_copy(
            update={
                "recall": tally.recall,
                "trajectory_recall": tally.trajectory_recall,
                "accuracy": tally.accuracy,
                "misroute_rate": tally.misroute_rate,
                "delta_recall": round(delta_recall, 4),
                "delta_trajectory_recall": round(delta_trajectory_recall, 4),
                "failed_queries": tally.failed_queries,
                "failed_trajectory_queries": tally.failed_trajectory_queries,
                "misrouted_queries": tally.misrouted_queries,
            }
        )

    def with_test_metrics(self, tally: CandidateProbeTally) -> OptimizationCandidate:
        """Return a copy populated with holdout test probe metrics from a CandidateProbeTally."""
        return self.model_copy(
            update={
                "test_recall": tally.recall,
                "test_trajectory_recall": tally.trajectory_recall,
                "test_accuracy": tally.accuracy,
                "test_misroute_rate": tally.misroute_rate,
            }
        )

    def with_baseline_metrics(
        self,
        *,
        recall: float,
        accuracy: float,
        misroute_rate: float,
        trajectory_recall: float | None = None,
    ) -> OptimizationCandidate:
        """Return a copy populated with baseline scores when description matches baseline."""
        eff_traj = trajectory_recall if trajectory_recall is not None else recall
        return self.model_copy(
            update={
                "recall": round(recall, 4),
                "trajectory_recall": round(eff_traj, 4),
                "accuracy": round(accuracy, 4),
                "misroute_rate": round(misroute_rate, 4),
                "delta_recall": 0.0,
                "delta_trajectory_recall": 0.0,
            }
        )

    def from_cached_train(self, cached: OptimizationCandidate) -> OptimizationCandidate:
        """Return a copy adopting cached training metrics from a prior evaluation."""
        return self.model_copy(
            update={
                "recall": cached.recall,
                "trajectory_recall": cached.trajectory_recall,
                "accuracy": cached.accuracy,
                "misroute_rate": cached.misroute_rate,
                "delta_recall": cached.delta_recall,
                "delta_trajectory_recall": cached.delta_trajectory_recall,
                "failed_queries": cached.failed_queries,
                "failed_trajectory_queries": cached.failed_trajectory_queries,
                "misrouted_queries": cached.misrouted_queries,
            }
        )

    def from_cached_test(self, cached: OptimizationCandidate) -> OptimizationCandidate:
        """Return a copy adopting cached holdout test metrics from a prior evaluation."""
        return self.model_copy(
            update={
                "test_recall": cached.test_recall,
                "test_trajectory_recall": cached.test_trajectory_recall,
                "test_accuracy": cached.test_accuracy,
                "test_misroute_rate": cached.test_misroute_rate,
            }
        )

from_cached_test

from_cached_test(
    cached: OptimizationCandidate,
) -> OptimizationCandidate

Return a copy adopting cached holdout test metrics from a prior evaluation.

Source code in src/reach/optimize.py

def from_cached_test(self, cached: OptimizationCandidate) -> OptimizationCandidate:
    """Return a copy adopting cached holdout test metrics from a prior evaluation."""
    return self.model_copy(
        update={
            "test_recall": cached.test_recall,
            "test_trajectory_recall": cached.test_trajectory_recall,
            "test_accuracy": cached.test_accuracy,
            "test_misroute_rate": cached.test_misroute_rate,
        }
    )

from_cached_train

from_cached_train(
    cached: OptimizationCandidate,
) -> OptimizationCandidate

Return a copy adopting cached training metrics from a prior evaluation.

Source code in src/reach/optimize.py

def from_cached_train(self, cached: OptimizationCandidate) -> OptimizationCandidate:
    """Return a copy adopting cached training metrics from a prior evaluation."""
    return self.model_copy(
        update={
            "recall": cached.recall,
            "trajectory_recall": cached.trajectory_recall,
            "accuracy": cached.accuracy,
            "misroute_rate": cached.misroute_rate,
            "delta_recall": cached.delta_recall,
            "delta_trajectory_recall": cached.delta_trajectory_recall,
            "failed_queries": cached.failed_queries,
            "failed_trajectory_queries": cached.failed_trajectory_queries,
            "misrouted_queries": cached.misrouted_queries,
        }
    )

mark_filtered

mark_filtered(reason: str = '') -> OptimizationCandidate

Return a copy marked as filtered out by static lint rules.

Source code in src/reach/optimize.py

def mark_filtered(self, reason: str = "") -> OptimizationCandidate:
    """Return a copy marked as filtered out by static lint rules."""
    return self.model_copy(
        update={"lint_clean": False, "filtered_out": True, "filter_reason": reason}
    )

unfiltered

unfiltered() -> OptimizationCandidate

Return a copy marked as passing static lint rules.

Source code in src/reach/optimize.py

def unfiltered(self) -> OptimizationCandidate:
    """Return a copy marked as passing static lint rules."""
    return self.model_copy(
        update={"lint_clean": True, "filtered_out": False, "filter_reason": ""}
    )

with_baseline_metrics

with_baseline_metrics(
    *,
    recall: float,
    accuracy: float,
    misroute_rate: float,
    trajectory_recall: float | None = None,
) -> OptimizationCandidate

Return a copy populated with baseline scores when description matches baseline.

Source code in src/reach/optimize.py

def with_baseline_metrics(
    self,
    *,
    recall: float,
    accuracy: float,
    misroute_rate: float,
    trajectory_recall: float | None = None,
) -> OptimizationCandidate:
    """Return a copy populated with baseline scores when description matches baseline."""
    eff_traj = trajectory_recall if trajectory_recall is not None else recall
    return self.model_copy(
        update={
            "recall": round(recall, 4),
            "trajectory_recall": round(eff_traj, 4),
            "accuracy": round(accuracy, 4),
            "misroute_rate": round(misroute_rate, 4),
            "delta_recall": 0.0,
            "delta_trajectory_recall": 0.0,
        }
    )

with_test_metrics

with_test_metrics(
    tally: CandidateProbeTally,
) -> OptimizationCandidate

Return a copy populated with holdout test probe metrics from a CandidateProbeTally.

Source code in src/reach/optimize.py

def with_test_metrics(self, tally: CandidateProbeTally) -> OptimizationCandidate:
    """Return a copy populated with holdout test probe metrics from a CandidateProbeTally."""
    return self.model_copy(
        update={
            "test_recall": tally.recall,
            "test_trajectory_recall": tally.trajectory_recall,
            "test_accuracy": tally.accuracy,
            "test_misroute_rate": tally.misroute_rate,
        }
    )

with_train_metrics

with_train_metrics(
    tally: CandidateProbeTally,
    *,
    delta_recall: float,
    delta_trajectory_recall: float = 0.0,
) -> OptimizationCandidate

Return a copy populated with training probe metrics from a CandidateProbeTally.

Source code in src/reach/optimize.py

def with_train_metrics(
    self,
    tally: CandidateProbeTally,
    *,
    delta_recall: float,
    delta_trajectory_recall: float = 0.0,
) -> OptimizationCandidate:
    """Return a copy populated with training probe metrics from a CandidateProbeTally."""
    return self.model_copy(
        update={
            "recall": tally.recall,
            "trajectory_recall": tally.trajectory_recall,
            "accuracy": tally.accuracy,
            "misroute_rate": tally.misroute_rate,
            "delta_recall": round(delta_recall, 4),
            "delta_trajectory_recall": round(delta_trajectory_recall, 4),
            "failed_queries": tally.failed_queries,
            "failed_trajectory_queries": tally.failed_trajectory_queries,
            "misrouted_queries": tally.misrouted_queries,
        }
    )

OptimizationReport

Bases: BaseModel

Represent the full results of closed-loop skill description optimization.

Source code in src/reach/optimize.py

class OptimizationReport(BaseModel):
    """Represent the full results of closed-loop skill description optimization."""

    model_config = ConfigDict(frozen=True)

    skill_name: str
    manifest_path: Path | None = None
    applied: bool = False
    has_probes: bool = False
    rival_name: str = ""
    baseline_accuracy: UnitMetric = 0.0
    baseline_recall: UnitMetric = 0.0
    baseline_trajectory_recall: UnitMetric = 0.0
    baseline_misroute: UnitMetric = 0.0
    baseline_description: str
    handoff: ReciprocalHandoff | None = None
    ceded_terms: tuple[str, ...] = ()
    unclaimed_terms: tuple[str, ...] = ()
    rounds: tuple[IterationRecord, ...] = ()
    candidates: tuple[OptimizationCandidate, ...] = ()

    @property
    def best_candidate(self) -> OptimizationCandidate | None:
        """Return top-ranked candidate if any candidates exist."""
        return self.candidates[0] if self.candidates else None

    @computed_field  # type: ignore[prop-decorator]
    @property
    def has_improvement(self) -> bool:
        """Indicate whether the top-ranked candidate improves over baseline."""
        return self.candidate_has_improvement(self.best_candidate)

    def candidate_has_improvement(self, candidate: OptimizationCandidate | None) -> bool:
        """Determine whether the specified candidate improves over baseline.

        Returns True if there are no empirical probes (heuristic mode), or if the candidate
        achieves positive delta recall, positive delta trajectory recall, or strictly lower
        misroute rate at equal recall.
        """
        if not self.has_probes:
            return True
        if candidate is None:
            return False
        return (
            candidate.delta_recall > 0.0
            or candidate.delta_trajectory_recall > 0.0
            or (candidate.delta_recall == 0.0 and candidate.misroute_rate < self.baseline_misroute)
        )

best_candidate property

best_candidate: OptimizationCandidate | None

Return top-ranked candidate if any candidates exist.

has_improvement property

has_improvement: bool

Indicate whether the top-ranked candidate improves over baseline.

candidate_has_improvement

candidate_has_improvement(
    candidate: OptimizationCandidate | None,
) -> bool

Determine whether the specified candidate improves over baseline.

Returns True if there are no empirical probes (heuristic mode), or if the candidate achieves positive delta recall, positive delta trajectory recall, or strictly lower misroute rate at equal recall.

Source code in src/reach/optimize.py

def candidate_has_improvement(self, candidate: OptimizationCandidate | None) -> bool:
    """Determine whether the specified candidate improves over baseline.

    Returns True if there are no empirical probes (heuristic mode), or if the candidate
    achieves positive delta recall, positive delta trajectory recall, or strictly lower
    misroute rate at equal recall.
    """
    if not self.has_probes:
        return True
    if candidate is None:
        return False
    return (
        candidate.delta_recall > 0.0
        or candidate.delta_trajectory_recall > 0.0
        or (candidate.delta_recall == 0.0 and candidate.misroute_rate < self.baseline_misroute)
    )

ReciprocalHandoff

Bases: BaseModel

Represent reciprocal Layer-2 SKILL.md body routing notes for target and rival.

Source code in src/reach/optimize.py

class ReciprocalHandoff(BaseModel):
    """Represent reciprocal Layer-2 SKILL.md body routing notes for target and rival."""

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

    target_skill: NonEmptyStr
    rival_skill: NonEmptyStr
    target_manifest_path: Path
    rival_manifest_path: Path
    target_note: NonEmptyStr
    rival_note: NonEmptyStr
    target_body_before: str = ""
    rival_body_before: str = ""

    @property
    def target_body_after(self) -> str:
        """Render the target SKILL.md body with its reciprocal Routing Note inserted."""
        return render_body_with_routing_note(
            self.target_body_before, self.rival_skill, self.target_note
        )

    @property
    def rival_body_after(self) -> str:
        """Render the rival SKILL.md body with its reciprocal Routing Note inserted."""
        return render_body_with_routing_note(
            self.rival_body_before, self.target_skill, self.rival_note
        )

rival_body_after property

rival_body_after: str

Render the rival SKILL.md body with its reciprocal Routing Note inserted.

target_body_after property

target_body_after: str

Render the target SKILL.md body with its reciprocal Routing Note inserted.

apply_optimization_candidate

apply_optimization_candidate(
    report: OptimizationReport,
    candidate: OptimizationCandidate,
) -> bool

Apply candidate description and optional reciprocal Layer-2 handoffs atomically to disk.

Parameters:

Name	Type	Description	Default
report	OptimizationReport	Optimization report containing target manifest path and staged handoff metadata.	required
candidate	OptimizationCandidate	Candidate proposed for disk application.	required

Returns:

Type	Description
bool	True if all target and rival file updates succeeded cleanly, or False otherwise.

Source code in src/reach/optimize.py

def apply_optimization_candidate(
    report: OptimizationReport,
    candidate: OptimizationCandidate,
) -> bool:
    """Apply candidate description and optional reciprocal Layer-2 handoffs atomically to disk.

    Args:
        report: Optimization report containing target manifest path and staged handoff metadata.
        candidate: Candidate proposed for disk application.

    Returns:
        True if all target and rival file updates succeeded cleanly, or False otherwise.
    """
    handoff = report.handoff
    if (
        report.manifest_path is None
        or not report.manifest_path.is_file()
        or (handoff is not None and not handoff.rival_manifest_path.is_file())
    ):
        return False
    if handoff is None:
        return update_skill_description(report.manifest_path, candidate.description)

    try:
        orig_target_content = report.manifest_path.read_text(encoding="utf-8")
    except OSError:
        return False

    target_body = _read_skill_body(report.manifest_path)
    updated_target_body = render_body_with_routing_note(
        target_body,
        handoff.rival_skill,
        handoff.target_note,
    )
    if not update_skill_description(
        report.manifest_path,
        candidate.description,
        body_override=updated_target_body,
    ):
        return False

    if not upsert_skill_routing_note(
        handoff.rival_manifest_path,
        handoff.target_skill,
        handoff.rival_note,
    ):
        # Rollback target manifest modification to preserve catalog consistency
        with contextlib.suppress(OSError):
            atomic_write_text(report.manifest_path, orig_target_content, encoding="utf-8")
        return False

    return True

optimize_skill

optimize_skill(
    skill_name: str,
    skills_path: Path | str | None = None,
    queries_path: Path | str | None = None,
    agent: str | None = None,
    candidates_count: int = 3,
    budget: int | None = None,
    auto_apply: bool = False,
    runtime_options: Mapping[str, Any] | None = None,
    config: Path | None = None,
    global_scope: bool = False,
    settings: OptimizeSettings | None = None,
    candidate_index: int = 1,
    force: bool = False,
    *,
    with_handoff: bool | None = None,
    workers: int | None = None,
    progress_callback: Callable[[str], None] | None = None,
) -> OptimizationReport

Orchestrate closed-loop skill description optimization and candidate evaluation.

Source code in src/reach/optimize.py

def optimize_skill(
    skill_name: str,
    skills_path: Path | str | None = None,
    queries_path: Path | str | None = None,
    agent: str | None = None,
    candidates_count: int = 3,
    budget: int | None = None,
    auto_apply: bool = False,
    runtime_options: Mapping[str, Any] | None = None,
    config: Path | None = None,
    global_scope: bool = False,
    settings: OptimizeSettings | None = None,
    candidate_index: int = 1,
    force: bool = False,
    *,
    with_handoff: bool | None = None,
    workers: int | None = None,
    progress_callback: Callable[[str], None] | None = None,
) -> OptimizationReport:
    """Orchestrate closed-loop skill description optimization and candidate evaluation."""
    settings = settings or OptimizeSettings()
    updates: dict[str, Any] = {}
    if budget is not None:
        updates["budget"] = budget
    if with_handoff is not None:
        updates["with_handoff"] = with_handoff
    if workers is not None:
        updates["workers"] = workers
    if updates:
        settings = settings.model_copy(update=updates)

    rt_settings = RuntimeSettings.resolve_for_optimize(config, agent=agent, options=runtime_options)
    context = _resolve_target_and_rivals(skill_name, skills_path, config, agent, global_scope)

    handoff: ReciprocalHandoff | None = None
    if settings.with_handoff and context.rival_skills:
        handoff = build_reciprocal_handoff(
            target=context.target_skill,
            rival=context.rival_skills[0],
            ceded_terms=context.ceded_terms,
            unclaimed_terms=context.unclaimed_terms,
        )

    exec_ctx = _OptimizeExecutionContext(
        runtime_settings=rt_settings,
        workers=settings.workers,
        handoff=handoff,
        progress_callback=progress_callback,
    )

    queries, train_queries, test_queries = _prepare_optimization_queries(
        target_skill=context.target_skill,
        all_skills=context.all_skills,
        rival_skills=context.rival_skills,
        queries_path=queries_path,
        settings=settings,
        agent=agent,
        config=config,
        ceded_terms=context.ceded_terms,
        unclaimed=context.unclaimed_terms,
        runtime_options=runtime_options,
    )

    baseline = _evaluate_baseline_performance(
        target_skill=context.target_skill,
        rival_skills=context.rival_skills,
        train_queries=train_queries,
        remaining_budget=settings.budget,
        candidates_count=candidates_count,
        agent=agent,
        config=config,
        skills_corpus=context.all_skills,
        exec_ctx=exec_ctx,
        runtime_options=runtime_options,
    )
    remaining_budget = baseline.remaining_budget

    lint_config = _resolve_lint_settings(config)
    driver = _setup_driver(agent, runtime_options, config=config)

    rounds_history: list[IterationRecord] = []
    all_candidates: list[OptimizationCandidate] = []
    global_best: OptimizationCandidate | None = None
    eval_cache = _CandidateEvalCache()

    for iter_idx in range(1, settings.iterations + 1):
        prev_desc = (
            global_best.description if global_best is not None else context.target_skill.description
        )
        failed_triggers = global_best.failed_queries if global_best is not None else ()
        false_triggers = global_best.misrouted_queries if global_best is not None else ()

        outcome = _run_optimization_round(
            context=context,
            train_queries=train_queries,
            test_queries=test_queries,
            agent=agent,
            driver=driver,
            lint_config=lint_config,
            config=config,
            candidates_count=candidates_count,
            failed_triggers=failed_triggers,
            false_triggers=false_triggers,
            prev_description=prev_desc,
            iter_idx=iter_idx,
            iterations=settings.iterations,
            remaining_budget=remaining_budget,
            holdout=settings.holdout,
            baseline=baseline,
            eval_cache=eval_cache,
            exec_ctx=exec_ctx,
            runtime_options=runtime_options,
        )
        remaining_budget = max(0, remaining_budget - outcome.probes_spent)

        if outcome.round_best is not None:
            has_test = bool(test_queries)
            if global_best is None or _candidate_rank_key(
                outcome.round_best, has_test=has_test
            ) >= _candidate_rank_key(global_best, has_test=has_test):
                global_best = outcome.round_best

            rounds_history.append(
                IterationRecord(
                    iteration=iter_idx,
                    candidates=outcome.evaluated_candidates,
                    best_candidate=outcome.round_best,
                    failed_queries=outcome.round_best.failed_queries,
                    misrouted_queries=outcome.round_best.misrouted_queries,
                    test_evaluated=outcome.test_evaluated,
                ),
            )
            all_candidates.extend(outcome.evaluated_candidates)

            if (
                global_best.recall >= 1.0
                and global_best.misroute_rate <= 0.0
                and (not test_queries or (global_best.test_recall or 0.0) >= 1.0)
            ):
                break

    return _build_optimization_report(
        skill_name=skill_name,
        target_skill=context.target_skill,
        rival_name=context.rival_name,
        ceded_terms=context.ceded_terms,
        unclaimed_terms=context.unclaimed_terms,
        baseline=baseline,
        all_candidates=all_candidates,
        global_best=global_best,
        rounds_history=rounds_history,
        has_test=bool(test_queries),
        has_probes=bool(queries),
        auto_apply=auto_apply,
        candidate_index=candidate_index,
        force=force,
        handoff=handoff,
    )