from __future__ import annotations

import json
import hashlib
from pathlib import Path
from typing import Any

from .config_signature import (
    effective_config_signature as _shared_effective_config_signature,
    normalized_config_patch as _shared_normalized_config_patch,
)
from .engine_adapters.vllm import default_vllm_descriptors
from .lca import EPSILON, WorkloadProfile
from .mechanism_planner import (
    CoordinateSearchPolicy,
    coordinate_line_search_candidates,
)
from .spec import ConfigPatch, Proposal, StudySpec, StudyState, TrialSummary


_TOPOLOGY_KEYS = {
    "tensor-parallel-size",
    "data-parallel-size",
    "expert-parallel-size",
    "enable-expert-parallel",
}
_RUNTIME_KEYS = {
    "max-num-seqs",
    "max-num-batched-tokens",
    "block-size",
    "gpu-memory-utilization",
    "enable-chunked-prefill",
}
_STRONG_INCUMBENT_MIN_GAIN = 1.8
_MIN_POST_INCUMBENT_VALIDATION_TRIALS = 2
_VALIDATION_TRIALS_WITHOUT_FAMILY_COVERAGE = 3
_STATEFUL_HISTORY_LIMIT = 8
# Decode-bound throughput is frequently KV-cache limited, so more gpu-memory-utilization
# yields more KV blocks and more concurrent decode. Hill-climb in small steps toward a
# safe ceiling and let measurement find the real peak: a too-high target regresses or
# fails to launch and is rejected by the incumbent guard, and its tested signature then
# blocks re-proposal so the climb terminates.
# Pathological starts below the nominal floor jump back into the normal operating range
# before this small-step climb begins.
_GMU_STEP = 0.02
_GMU_NOMINAL_FLOOR = 0.9
_GMU_SAFE_CEILING = 0.97
_PREFILL_QUANTUM_HEAD_OF_LINE_RATIO = 1.0
_PREFILL_QUANTUM_FRAGMENTATION_RATIO = 0.5
_ADMISSION_PRESSURE_STEP_UP = 1.5
_FRONTIER_DELTA_MIN_ABS_GAIN = 0.02
_FRONTIER_DELTA_MIN_REL_GAIN = 0.03
_FRONTIER_DELTA_PROJECTED_INCUMBENT_FLOOR = 0.98
_FRONTIER_DELTA_TARGET_BASELINE_COMPETITIVE_FLOOR = 0.75


def build_harness_context(
    *,
    study: StudySpec,
    window_summary: dict[str, Any],
    state: StudyState,
    workload_profile: WorkloadProfile | None = None,
) -> dict[str, Any]:
    recent_diagnostics = _recent_trial_diagnostics(state)
    trial_profiles = _trial_profiles(study, recent_diagnostics)
    bottleneck_hypotheses = _rank_bottleneck_hypotheses(
        study,
        window_summary,
        trial_profiles,
    )
    experiment_plan = _experiment_plan(
        study,
        window_summary,
        state,
        recent_diagnostics,
        trial_profiles,
        bottleneck_hypotheses,
    )
    harness_stop = _harness_stop_decision(
        study,
        state,
        recent_diagnostics,
        experiment_plan=experiment_plan,
    )
    return {
        "paper_alignment": {
            "goal": "Use workload-feature-to-knob harnesses to reduce wasted trials and avoid regressing after a good configuration is found.",
            "feature_model": "L-C-A: request lengths, inter-request KV-cache reuse, and arrival dynamics.",
            "trial_policy": "Profile measured trials, rank bottleneck hypotheses, score generic candidate actions, and stop only when no useful measured hypothesis remains.",
        },
        "workload_lca_profile": _workload_lca_profile(window_summary, workload_profile),
        "recent_trial_diagnostics": recent_diagnostics,
        "trial_profiles": trial_profiles,
        "bottleneck_hypotheses": bottleneck_hypotheses,
        "candidate_actions": experiment_plan["candidate_actions"],
        "experiment_plan": experiment_plan,
        "convergence_guard": _convergence_guard(state, recent_diagnostics),
        "harness_stop": harness_stop,
        "stop_authority": _stop_authority(
            study,
            state,
            recent_diagnostics,
            experiment_plan,
            harness_stop,
        ),
        "harness_proposal": _harness_proposal_decision(
            study,
            window_summary,
            state,
            recent_diagnostics,
            experiment_plan=experiment_plan,
        ),
        "knob_harnesses": _knob_harnesses(study, window_summary, recent_diagnostics),
        "proposal_rules": _proposal_rules(),
    }


def build_harness_stop_proposal(context: dict[str, Any]) -> Proposal | None:
    stop = context.get("harness_stop")
    if not isinstance(stop, dict) or not stop.get("should_stop"):
        return None
    reason = str(stop.get("reason") or "harness_converged")
    evidence = stop.get("evidence") if isinstance(stop.get("evidence"), dict) else {}
    observation = (
        "Harness convergence guard triggered before requesting another proposal: "
        f"{reason}."
    )
    diagnosis = str(evidence.get("summary") or reason)
    return Proposal(
        observation=observation,
        diagnosis=diagnosis,
        config_patch=ConfigPatch(env_patch={}, flag_patch={}),
        expected_effects=[
            "stop without spending another GPU trial",
            "preserve the current best observed configuration",
        ],
        why_not_previous_failures=(
            "The stop decision is based on completed validation evidence and does not "
            "repeat any failed configuration."
        ),
        should_stop=True,
    )


def build_harness_guided_proposal(context: dict[str, Any]) -> Proposal | None:
    proposal = context.get("harness_proposal")
    if not isinstance(proposal, dict) or not proposal.get("should_propose"):
        return None
    patch = proposal.get("config_patch")
    if not isinstance(patch, dict):
        return None
    flag_patch = patch.get("flag_patch")
    env_patch = patch.get("env_patch")
    if not isinstance(flag_patch, dict) or not isinstance(env_patch, dict):
        return None
    reason = str(proposal.get("reason") or "harness_guided_probe")
    diagnosis = str(proposal.get("diagnosis") or reason)
    return Proposal(
        observation=(
            "Harness selected the highest-scoring deterministic candidate-set action: "
            f"{reason}."
        ),
        diagnosis=diagnosis,
        config_patch=ConfigPatch(env_patch=dict(env_patch), flag_patch=dict(flag_patch)),
        expected_effects=[
            str(item)
            for item in proposal.get("expected_effects", [])
            if isinstance(item, str)
        ],
        why_not_previous_failures=(
            "The proposal is based on the first completed baseline trial and does not "
            "repeat a prior failed configuration."
        ),
        should_stop=False,
    )


def render_harness_context(context: dict[str, Any]) -> str:
    return json.dumps(context, ensure_ascii=False, indent=2)


def _workload_lca_profile(
    window_summary: dict[str, Any],
    workload_profile: WorkloadProfile | None = None,
) -> dict[str, Any]:
    if workload_profile is not None:
        return _canonical_lca_profile(workload_profile)
    prefix_cache = window_summary.get("prefix_cache")
    if not isinstance(prefix_cache, dict):
        prefix_cache = {}
    prompt_p50 = _as_float(window_summary.get("prompt_tokens_p50"))
    prompt_p95 = _as_float(window_summary.get("prompt_tokens_p95"))
    prompt_p99 = _as_float(window_summary.get("prompt_tokens_p99"))
    tail_ratio = _as_float(window_summary.get("prompt_tail_ratio_p95_p50"))
    repeated_token_ratio = _as_float(prefix_cache.get("repeated_token_ratio_estimate"))
    burst_ratio = _as_float(window_summary.get("arrival_burst_ratio_p95_to_mean"))
    interarrival_cv = _as_float(window_summary.get("arrival_interarrival_cv"))
    return {
        "L_request_lengths": {
            "prompt_tokens_p50": prompt_p50,
            "prompt_tokens_p95": prompt_p95,
            "prompt_tokens_p99": prompt_p99,
            "tail_ratio_p95_p50": tail_ratio,
            "regime": _length_regime(prompt_p95, tail_ratio),
        },
        "C_prefix_cache": {
            "repeated_token_ratio_estimate": repeated_token_ratio,
            "repeated_block_ratio": _as_float(prefix_cache.get("repeated_block_ratio")),
            "multi_turn_request_ratio": _as_float(prefix_cache.get("multi_turn_request_ratio")),
            "rows_with_hash_ids": int(prefix_cache.get("rows_with_hash_ids") or 0),
            "regime": _cache_regime(repeated_token_ratio),
        },
        "A_arrivals": {
            "request_rate": _as_float(window_summary.get("request_rate")),
            "qps_1s_p50": _as_float(window_summary.get("arrival_qps_1s_p50")),
            "qps_1s_p95": _as_float(window_summary.get("arrival_qps_1s_p95")),
            "burst_ratio_p95_to_mean": burst_ratio,
            "interarrival_cv": interarrival_cv,
            "regime": _arrival_regime(burst_ratio, interarrival_cv),
        },
    }


def _canonical_lca_profile(profile: WorkloadProfile) -> dict[str, Any]:
    """Authoritative L-C-A block: the paper's 10-dim RobustScaler vector.

    Sourced from lca.WorkloadProfile so the prompt's L-C-A is the same metric
    used for the workload-similarity computations, not an ad-hoc re-derivation.
    The regime labels reuse the heuristic classifiers but are fed from the
    canonical stats.
    """
    stats = profile.stats if isinstance(profile.stats, dict) else {}
    length = stats.get("length") if isinstance(stats.get("length"), dict) else {}
    cache = stats.get("cache") if isinstance(stats.get("cache"), dict) else {}
    arrival = stats.get("arrival") if isinstance(stats.get("arrival"), dict) else {}
    length_p95 = _as_float(length.get("p95"))
    length_p50 = _as_float(length.get("p50"))
    tail_ratio = float(length_p95 / max(length_p50, EPSILON)) if length_p95 else 0.0
    repeated_token_ratio = _as_float(cache.get("input_hit_rate"))
    fano_1s = _as_float(arrival.get("fano_1s"))
    interarrival_cv = _as_float(arrival.get("interarrival_cv"))
    return {
        "metric": "paper L-C-A (10-dim, RobustScaler-normalized) from lca.WorkloadProfile",
        "length_mode": profile.length_mode,
        "feature_names": list(profile.feature_names),
        "vector": list(profile.vector),
        "L_request_lengths": {
            "mean": _as_float(length.get("mean")),
            "p50": length_p50,
            "p95": length_p95,
            "cv": _as_float(length.get("cv")),
            "tail_ratio_p95_p50": tail_ratio,
            "regime": _length_regime(length_p95, tail_ratio),
        },
        "C_prefix_cache": {
            "hit_rate": _as_float(cache.get("hit_rate")),
            "input_hit_rate": repeated_token_ratio,
            "repeated_block_ratio": _as_float(cache.get("repeated_block_ratio")),
            "rows_with_hash_ids": int(cache.get("rows_with_hash_ids") or 0),
            "regime": _cache_regime(repeated_token_ratio),
        },
        "A_arrivals": {
            "request_rate": _as_float(arrival.get("request_rate")),
            "request_rate_per_gpu": _as_float(arrival.get("request_rate_per_gpu")),
            "interarrival_cv": interarrival_cv,
            "fano_1s": fano_1s,
            "regime": _arrival_regime(fano_1s, interarrival_cv),
        },
    }


def _knob_harnesses(
    study: StudySpec,
    window_summary: dict[str, Any],
    recent_diagnostics: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    tunable = set(study.engine.tunable_flags)
    latest = recent_diagnostics[-1] if recent_diagnostics else {}
    active_bottleneck = latest.get("active_bottleneck") or _workload_default_bottleneck(
        study, window_summary
    )
    harnesses: list[dict[str, Any]] = []
    if "tensor-parallel-size" in tunable:
        harnesses.append(
            {
                "knob_family": "tensor-parallel-size",
                "use_when": [
                    "TTFT failures dominate, especially on long prompt windows.",
                    "The L profile has a heavy tail and prefill service time is the likely bottleneck.",
                    "For decode-only studies, TP/DP redistribution can be the primary way to trade per-request decode latency against GPU-parallel serving replicas.",
                ],
                "procedure": [
                    "Probe only adjacent legal TP choices around the incumbent topology.",
                    "Prefer TP up when it lowers long-prefill latency and the projected request rate remains left of the high-TP queueing knee.",
                    "For decode_tpot bottlenecks, compare legal TP/DP redistributions using request_rate_per_gpu and SLO pass rate; prefer smaller TP with larger DP only when the model fits and observed TPOT remains within the configured SLO.",
                    "Prefer TP down or keep moderate TP when communication overhead or concurrency loss becomes visible.",
                ],
                "guards": [
                    "Do not jump across multiple TP values without a launch-safe reason.",
                    "Do not raise TP for a short-prompt/cache-heavy window if TTFT is already passing and TPOT or queueing is the active bottleneck, unless decode-only TPOT evidence points to under-parallelized decoding.",
                    "Keep TP/DP/EP inside topology_constraints.",
                ],
                "active_now": active_bottleneck in {"ttft_prefill", "decode_tpot"},
            }
        )
    if "data-parallel-size" in tunable:
        harnesses.append(
            {
                "knob_family": "data-parallel-size",
                "use_when": [
                    "Decode-only or cache-heavy workloads need more independent replicas after TTFT is no longer the limiting objective.",
                    "The active bottleneck is decode_tpot or admission/queueing and topology_constraints allow a legal TP/DP redistribution.",
                ],
                "procedure": [
                    "Change DP only as part of a legal adjacent TP/DP topology move when the product constraints require coupling.",
                    "Evaluate the candidate by request_rate_per_gpu, configured TPOT/TTFT pass rate, and launch stability; do not optimize raw request rate alone.",
                    "Use previous same-topology or adjacent-topology history to decide whether DP improved admission without making token latency infeasible.",
                ],
                "guards": [
                    "Do not reduce TP so far that decode_tpot failures dominate the configured SLO.",
                    "Do not repeat a topology whose launch failure implicated TP/DP/EP.",
                    "Keep TP/DP/EP inside topology_constraints.",
                ],
                "active_now": active_bottleneck in {"decode_tpot", "admission_or_queueing"},
            }
        )
    if "max-num-seqs" in tunable:
        harnesses.append(
            {
                "knob_family": "max-num-seqs",
                "use_when": [
                    "Prefix-cache reuse is high, requests are short-to-moderate after cache hits, and queueing/admission is limiting throughput.",
                    "TTFT is mostly passing but offered load stalls below the target.",
                    "For decode_tpot bottlenecks, decode concurrency is too high or too low relative to the configured TPOT SLO and observed pass-rate headroom.",
                ],
                "procedure": [
                    "Increase max-num-seqs one step at a time to exploit cache-created parallelism.",
                    "For decode_tpot, lower max-num-seqs when TPOT failures dominate; raise it only when the incumbent has SLO headroom and admission/queueing limits throughput.",
                    "Decrease it if p95 TTFT worsens, prefill queueing appears, TPOT failures dominate, or memory pressure causes launch/runtime failures.",
                ],
                "guards": [
                    "Avoid large max-num-seqs increases on low-cache or heavy-tail windows.",
                    "Do not combine a max-num-seqs jump with a TP jump unless the history clearly isolates both bottlenecks.",
                ],
                "active_now": active_bottleneck in {"decode_tpot", "admission_or_queueing"},
            }
        )
    if "max-num-batched-tokens" in tunable:
        harnesses.append(
            {
                "knob_family": "max-num-batched-tokens",
                "use_when": [
                    "Prefill batching is too small for the L profile or TTFT is hurt by excessive chunking overhead.",
                    "GPU work appears fragmented and the incumbent is stable but under-utilized.",
                    "For decode-only workloads, decode batching pressure affects TPOT pass rate after topology is stable.",
                ],
                "procedure": [
                    "Raise MBT for long prompts when memory headroom and SLO permit.",
                    "Lower MBT if long requests monopolize batches and short-request TTFT regresses.",
                    "For decode_tpot, lower MBT when token-latency failures dominate and raise it only when SLO headroom and under-utilization are both observed.",
                ],
                "guards": [
                    "Keep MBT changes within a conservative trust region.",
                    "Do not raise MBT after OOM or launch failures involving memory-related knobs.",
                    "Do not raise MBT when the incumbent MBT already covers prompt p99 unless same-topology history proves prefill fragmentation is the bottleneck.",
                ],
                "active_now": active_bottleneck in {"ttft_prefill", "decode_tpot"},
            }
        )
    if "enable-chunked-prefill" in tunable:
        prefill_scheduler_active = (
            active_bottleneck in {"ttft_prefill", "admission_or_queueing"}
            and _prefill_scheduler_workload_applies(study, window_summary)
        )
        harnesses.append(
            {
                "knob_family": "enable-chunked-prefill",
                "use_when": [
                    "The L profile has a long tail and long prefills block shorter requests.",
                    "Treat chunked prefill as part of a scheduler interaction with prefill quantum and admission pressure, not as a standalone magic flag.",
                ],
                "procedure": [
                    "Enable chunked prefill when the measured bottleneck indicates prefill head-of-line blocking.",
                    "Move max-num-batched-tokens by relative trust-region steps in normalized prefill_quantum_ratio space.",
                    "Move max-num-seqs only as a relative admission-pressure correction, and preserve topology while testing this scheduler hypothesis.",
                ],
                "guards": [
                    "Do not disable chunked prefill on a heavy-tail workload without direct evidence from a nearby trial.",
                    "Do not use fixed absolute MBT/MNS tables; derive the next concrete flag values from the incumbent and workload scale.",
                ],
                "active_now": prefill_scheduler_active,
            }
        )
    if "expert-parallel-size" in tunable or "enable-expert-parallel" in tunable:
        harnesses.append(
            {
                "knob_family": "expert-parallel",
                "use_when": [
                    "Only when history or a capability profile identifies expert communication or MoE dispatch as the active bottleneck.",
                    "For MoE decode-only studies, retain an already valid EP topology when it is part of the baseline constraints or prior best, but change EP only with direct adjacent evidence.",
                ],
                "procedure": [
                    "Keep expert parallel disabled for pure TTFT/prefill tuning unless there is direct positive evidence for EP on this stack.",
                    "For decode_tpot, preserve the current effective EP setting unless the nearby history shows EP-specific token-latency or launch behavior is the limiting factor.",
                    "If EP is tested, change only EP-related knobs and treat launch/runtime failure as hard negative evidence.",
                ],
                "guards": [
                    "Do not introduce EP as the first follow-up after a successful TP increase.",
                    "Do not use EP to address generic TTFT-prefill bottlenecks; TP and batching harnesses are the relevant families.",
                    "Do not change EP size for a decode-only run just to search broader topology space; require EP-specific evidence or a topology constraint that forces the current EP.",
                    "Do not enable EP after any launch failure involving expert-parallel knobs.",
                ],
                "active_now": False,
            }
        )
    if "gpu-memory-utilization" in tunable:
        harnesses.append(
            {
                "knob_family": "gpu-memory-utilization",
                "use_when": [
                    "The engine launches cleanly but memory headroom limits batching.",
                    "A decode-bound incumbent (decode_tpot) is KV-cache limited and could sustain more concurrent decode with more KV blocks.",
                ],
                "procedure": [
                    "Make small adjustments only after topology and batching knobs are stable.",
                    "Raise gpu-memory-utilization one small step at a time and keep the step only if request_rate_per_gpu improves and the engine still launches.",
                ],
                "guards": [
                    "Treat launch OOM as hard negative evidence and back off immediately.",
                    "Do not exceed a safe utilization ceiling; stop climbing once a higher target regresses or fails to launch.",
                ],
                "active_now": active_bottleneck in {"decode_tpot", "admission_or_queueing"},
            }
        )
    return harnesses


def _recent_trial_diagnostics(
    state: StudyState,
    *,
    limit: int | None = _STATEFUL_HISTORY_LIMIT,
) -> list[dict[str, Any]]:
    diagnostics: list[dict[str, Any]] = []
    trials = state.trials if limit is None else state.trials[-limit:]
    for trial in trials:
        item: dict[str, Any] = {
            "trial_id": trial.trial_id,
            "status": trial.status,
            "parallel_size": trial.parallel_size,
            "best_request_rate": trial.best_request_rate,
            "best_request_rate_per_gpu": trial.best_request_rate_per_gpu,
            "best_pass_rate": trial.best_pass_rate,
            "diagnosis": trial.diagnosis,
            "config_patch": trial.config_patch,
            "failure_stage": trial.failure_stage,
            "failure_reason": trial.failure_reason,
        }
        result = _load_result(trial)
        if result:
            measurement = result.get("measurement")
            if isinstance(measurement, dict):
                item["measurement_evidence"] = measurement
            probes = result.get("probes")
            if isinstance(probes, list) and probes:
                best_probe = _best_feasible_probe(probes)
                last_probe = probes[-1] if isinstance(probes[-1], dict) else None
                item["probe_summary"] = {
                    "best_feasible_probe": _compact_probe(best_probe),
                    "last_probe": _compact_probe(last_probe),
                }
                item["active_bottleneck"] = _active_bottleneck_from_probe_sequence(
                    probes,
                    fallback=best_probe or last_probe,
                )
            elif result.get("all_infeasible_diagnostics"):
                diag = result["all_infeasible_diagnostics"]
                item["probe_summary"] = {"all_infeasible": diag}
                item["active_bottleneck"] = _active_bottleneck({"payload": diag})
        else:
            item["active_bottleneck"] = _failure_bottleneck(trial)
        diagnostics.append(item)
    return diagnostics


def _trial_profiles(
    study: StudySpec,
    recent_diagnostics: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    profiles: list[dict[str, Any]] = []
    for item in recent_diagnostics:
        flags = _effective_flags_for_item(study, item)
        probe_summary = item.get("probe_summary")
        best_probe = None
        last_probe = None
        all_infeasible = None
        if isinstance(probe_summary, dict):
            best_probe = probe_summary.get("best_feasible_probe")
            last_probe = probe_summary.get("last_probe")
            all_infeasible = probe_summary.get("all_infeasible")
        limiting_probe = (
            last_probe
            if isinstance(last_probe, dict)
            else all_infeasible
            if isinstance(all_infeasible, dict)
            else best_probe
            if isinstance(best_probe, dict)
            else None
        )
        latency = limiting_probe.get("latency_summary") if isinstance(limiting_probe, dict) else {}
        if not isinstance(latency, dict):
            latency = {}
        failed_counts = latency.get("failed_reason_counts")
        if not isinstance(failed_counts, dict):
            failed_counts = {}
        profile = {
            "trial_id": item.get("trial_id"),
            "status": item.get("status"),
            "config_patch": item.get("config_patch") if isinstance(item.get("config_patch"), dict) else {},
            "topology": {
                "tensor_parallel_size": _parse_int_like(
                    flags.get("tensor-parallel-size"),
                    default=1,
                ),
                "data_parallel_size": _parse_int_like(
                    flags.get("data-parallel-size"),
                    default=1,
                ),
                "expert_parallel_size": _parse_int_like(
                    flags.get("expert-parallel-size"),
                    default=1,
                ),
                "enable_expert_parallel": bool(flags.get("enable-expert-parallel", False)),
            },
            "performance": {
                "best_request_rate": item.get("best_request_rate"),
                "best_request_rate_per_gpu": item.get("best_request_rate_per_gpu"),
                "best_pass_rate": item.get("best_pass_rate"),
            },
            "probe_profile": {
                "best_feasible_probe": best_probe,
                "limiting_probe": limiting_probe,
                "active_bottleneck": item.get("active_bottleneck"),
                "failed_reason_counts": failed_counts,
                "latency_quantiles": {
                    "ttft_ms": latency.get("ttft_ms") if isinstance(latency.get("ttft_ms"), dict) else {},
                    "tpot_ms": latency.get("tpot_ms") if isinstance(latency.get("tpot_ms"), dict) else {},
                },
            },
            "failure_profile": {
                "failure_stage": item.get("failure_stage"),
                "failure_reason": item.get("failure_reason"),
            },
        }
        profiles.append(profile)
    return profiles


def _rank_bottleneck_hypotheses(
    study: StudySpec,
    window_summary: dict[str, Any],
    trial_profiles: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    scores = {
        "ttft_prefill": 0.0,
        "decode_tpot": 0.0,
        "admission_or_queueing": 0.0,
        "launch_or_memory": 0.0,
    }
    evidence: dict[str, list[str]] = {name: [] for name in scores}

    default = _workload_default_bottleneck(study, window_summary)
    if default in scores:
        scores[default] += 0.18
        evidence[default].append(f"workload default bottleneck is {default}")

    if study.trace.request_mode == "decode_only" and study.slo.tpot_rule is not None:
        scores["decode_tpot"] += 0.22
        evidence["decode_tpot"].append("decode_only study with configured TPOT SLO")
    if study.slo.ttft_rule is not None:
        prompt_p95 = _as_float(window_summary.get("prompt_tokens_p95"))
        if prompt_p95 >= 4096:
            scores["ttft_prefill"] += 0.14
            evidence["ttft_prefill"].append("long prompt p95 makes TTFT/prefill plausible")

    for profile in trial_profiles[-stateful_history_limit() :]:
        active = str(profile.get("probe_profile", {}).get("active_bottleneck") or "")
        if active in scores:
            weight = 0.34 if profile is trial_profiles[-1] else 0.18
            scores[active] += weight
            evidence[active].append(
                f"{profile.get('trial_id')} probe diagnosis is {active}"
            )
        failed = profile.get("probe_profile", {}).get("failed_reason_counts")
        if not isinstance(failed, dict):
            failed = {}
        ttft_count = sum(int(v) for k, v in failed.items() if str(k).startswith("ttft"))
        tpot_count = sum(int(v) for k, v in failed.items() if str(k).startswith("tpot"))
        elapsed_count = sum(
            int(v)
            for k, v in failed.items()
            if str(k).startswith("probe_elapsed_s>")
            or str(k).startswith("arrival_lag_s>")
        )
        total = max(ttft_count + tpot_count + elapsed_count, 1)
        if ttft_count:
            scores["ttft_prefill"] += min(0.24, 0.24 * ttft_count / total)
            evidence["ttft_prefill"].append(
                f"{profile.get('trial_id')} TTFT failures={ttft_count}"
            )
        if tpot_count:
            scores["decode_tpot"] += min(0.24, 0.24 * tpot_count / total)
            evidence["decode_tpot"].append(
                f"{profile.get('trial_id')} TPOT failures={tpot_count}"
            )
        if elapsed_count:
            scores["admission_or_queueing"] += min(0.18, 0.18 * elapsed_count / total)
            evidence["admission_or_queueing"].append(
                f"{profile.get('trial_id')} queue/elapsed failures={elapsed_count}"
            )
        failure_stage = str(profile.get("failure_profile", {}).get("failure_stage") or "")
        failure_reason = str(profile.get("failure_profile", {}).get("failure_reason") or "")
        if failure_stage == "engine_launch" or "out of memory" in failure_reason.lower():
            scores["launch_or_memory"] += 0.4
            evidence["launch_or_memory"].append(
                f"{profile.get('trial_id')} launch or memory failure"
            )

    ranked = []
    for name, score in scores.items():
        if score <= 0:
            continue
        ranked.append(
            {
                "name": name,
                "confidence": min(0.99, round(score, 4)),
                "evidence": evidence[name][:6],
            }
        )
    ranked.sort(key=lambda item: item["confidence"], reverse=True)
    return ranked


def stateful_history_limit() -> int:
    return _STATEFUL_HISTORY_LIMIT


def _state_completed_trials_with_rates(state: StudyState) -> list[TrialSummary]:
    return [
        trial
        for trial in state.trials
        if trial.status == "completed"
        and isinstance(trial.best_request_rate_per_gpu, (int, float))
    ]


def _load_result(trial: TrialSummary) -> dict[str, Any] | None:
    if not trial.result_path:
        return None
    path = Path(trial.result_path)
    if not path.exists():
        return None
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    return payload if isinstance(payload, dict) else None


def _best_feasible_probe(probes: list[Any]) -> dict[str, Any] | None:
    best: dict[str, Any] | None = None
    for probe in probes:
        if not isinstance(probe, dict) or not probe.get("feasible"):
            continue
        payload = probe.get("payload")
        if not isinstance(payload, dict):
            continue
        if best is None:
            best = probe
            continue
        best_payload = best.get("payload") if isinstance(best.get("payload"), dict) else {}
        if _as_float(payload.get("request_rate")) > _as_float(best_payload.get("request_rate")):
            best = probe
    return best


def _compact_probe(probe: dict[str, Any] | None) -> dict[str, Any] | None:
    if not isinstance(probe, dict):
        return None
    payload = probe.get("payload")
    if not isinstance(payload, dict):
        payload = probe
    return {
        "threshold": probe.get("threshold", payload.get("threshold")),
        "feasible": probe.get("feasible", payload.get("feasible")),
        "request_count": payload.get("request_count"),
        "pass_rate": payload.get("pass_rate"),
        "request_rate": payload.get("request_rate"),
        "early_stopped": payload.get("early_stopped"),
        "early_stop_reason": payload.get("early_stop_reason"),
        "latency_summary": payload.get("latency_summary"),
    }


def _active_bottleneck(probe: dict[str, Any] | None) -> str:
    compact = _compact_probe(probe)
    if not compact:
        return "unknown"
    early_stop_reason = str(compact.get("early_stop_reason") or "")
    if early_stop_reason.startswith("arrival_lag_s>"):
        return "admission_or_queueing"
    latency_summary = compact.get("latency_summary")
    if not isinstance(latency_summary, dict):
        if early_stop_reason == "slo_pass_rate_unrecoverable":
            return "admission_or_queueing"
        return "unknown"
    failed = latency_summary.get("failed_reason_counts")
    if not isinstance(failed, dict) or not failed:
        if early_stop_reason == "slo_pass_rate_unrecoverable":
            return "admission_or_queueing"
        return "none_obvious"
    ttft_count = sum(int(v) for k, v in failed.items() if str(k).startswith("ttft"))
    tpot_count = sum(int(v) for k, v in failed.items() if str(k).startswith("tpot"))
    request_failed_count = sum(
        int(v)
        for k, v in failed.items()
        if not str(k).startswith("ttft")
        and not str(k).startswith("tpot")
        and not str(k).startswith("probe_elapsed_s>")
        and str(k) != "slo_pass_rate_unrecoverable"
    )
    if ttft_count == 0 and tpot_count == 0 and request_failed_count == 0:
        return "none_obvious"
    if ttft_count >= max(tpot_count, request_failed_count):
        return "ttft_prefill"
    if tpot_count >= max(ttft_count, request_failed_count):
        return "decode_tpot"
    return "admission_or_queueing"


def _active_bottleneck_from_probe_sequence(
    probes: list[Any],
    *,
    fallback: dict[str, Any] | None,
) -> str:
    for probe in reversed(probes):
        if not isinstance(probe, dict):
            continue
        bottleneck = _active_bottleneck(probe)
        if bottleneck not in {"unknown", "none_obvious"}:
            return bottleneck
    return _active_bottleneck(fallback)


def _failure_bottleneck(trial: TrialSummary) -> str:
    if trial.failure_stage == "engine_launch":
        return "launch_or_memory"
    if trial.failure_reason:
        return "runtime_failure"
    return "unknown"


def _convergence_guard(
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
) -> dict[str, Any]:
    infeasible_progress = _infeasible_progress_guard(recent_diagnostics)
    strong_incumbent = _strong_incumbent_guard(state, recent_diagnostics)
    completed = [
        item
        for item in recent_diagnostics
        if item.get("status") == "completed"
        and isinstance(item.get("best_request_rate_per_gpu"), (int, float))
    ]
    should_stop = False
    reason = "insufficient_completed_trials"
    if len(completed) >= 3 and state.best_request_rate_per_gpu is not None:
        recent_rates = [float(item["best_request_rate_per_gpu"]) for item in completed[-3:]]
        best = float(state.best_request_rate_per_gpu)
        near_best = [rate >= best * 0.98 for rate in recent_rates]
        if all(near_best):
            should_stop = True
            reason = "last_three_completed_trials_are_within_2_percent_of_incumbent"
        else:
            reason = "recent_trials_still_show_material_variation"
    elif state.best_request_rate_per_gpu is not None:
        reason = "need_more_evidence_before_stop"
    if not should_stop and infeasible_progress["plateau_detected"]:
        reason = str(infeasible_progress["reason"])
    if (
        not should_stop
        and not infeasible_progress["plateau_detected"]
        and strong_incumbent["guard_active"]
    ):
        reason = "strong_incumbent_requires_validation_probes"
    return {
        "deterministic_stop": should_stop,
        "should_stop_if_no_harness_can_justify_a_new_adjacent_probe": (
            should_stop
            or bool(infeasible_progress["stop_if_next_probe_repeats_family"])
        ),
        "reason": reason,
        "infeasible_progress": infeasible_progress,
        "strong_incumbent": strong_incumbent,
        "incumbent": {
            "trial_id": state.best_trial_id,
            "parallel_size": state.best_parallel_size,
            "best_sampling_u": state.best_sampling_u,
            "best_request_rate": state.best_request_rate,
            "best_request_rate_per_gpu": state.best_request_rate_per_gpu,
        },
    }


def _harness_stop_decision(
    study: StudySpec,
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
    *,
    experiment_plan: dict[str, Any] | None = None,
) -> dict[str, Any]:
    high_saturation = _search_high_saturation_guard(study, state, recent_diagnostics)
    if high_saturation["saturated"] and _parallel_size_can_vary(study):
        return {
            "should_stop": False,
            "reason": "search_high_saturation_requires_parallel_size_evidence",
            "evidence": {
                "summary": (
                    "search_high_saturation is measurement evidence only; "
                    "request_rate_per_gpu studies with variable topology need a "
                    "parallel-size/topology comparison before stop can be authorized."
                ),
                "objective": "request_rate_per_gpu",
                "search_high_saturation": high_saturation,
            },
        }
    topology_frontier = _topology_frontier_status(study, state, recent_diagnostics)
    if topology_frontier["frontier_open"]:
        return {
            "should_stop": False,
            "reason": "topology_frontier_requires_probe",
            "evidence": topology_frontier,
        }
    if experiment_plan is not None and experiment_plan.get("next_action"):
        action = experiment_plan["next_action"]
        if isinstance(action, dict) and _as_float(action.get("score")) >= 0.35:
            return {
                "should_stop": False,
                "reason": "experiment_plan_has_high_value_candidate",
                "evidence": {
                    "summary": "The profile-driven planner still has a useful measured hypothesis to test.",
                    "next_action": action,
                },
            }
    guard = _convergence_guard(state, recent_diagnostics)
    if guard["deterministic_stop"]:
        return {
            "should_stop": True,
            "reason": guard["reason"],
            "evidence": {
                "summary": "The convergence guard fired and no further adjacent probe is required.",
                "convergence_guard": guard,
            },
        }
    validation = _validation_exhausted_guard(state, recent_diagnostics)
    if validation["exhausted"]:
        return {
            "should_stop": True,
            "reason": validation["reason"],
            "evidence": validation,
        }
    return {
        "should_stop": False,
        "reason": "continue_harness_guided_search",
        "evidence": {
            "summary": "No deterministic harness stop condition is satisfied.",
            "search_high_saturation": high_saturation,
            "convergence_guard": guard,
            "validation_exhausted": validation,
        },
    }


def _stop_authority(
    study: StudySpec,
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
    experiment_plan: dict[str, Any] | None,
    harness_stop: dict[str, Any],
) -> dict[str, Any]:
    """Stop-B authority: the deterministic validator decides if stopping is justified.

    ``authorized`` mirrors the deterministic harness stop decision. The LLM's
    should_stop is only a corroborating signal: the tuning loop honors an
    LLM-originated stop only when this validator authorizes it (or when the
    harness is disabled). ``opportunity_remains`` flags that a concrete adjacent
    probe (open topology frontier or a high-value planned candidate) still exists,
    so an early stop would leave measured headroom on the table.
    """
    frontier = _topology_frontier_status(study, state, recent_diagnostics)
    next_action = (
        experiment_plan.get("next_action") if isinstance(experiment_plan, dict) else None
    )
    has_candidate = (
        isinstance(next_action, dict) and _as_float(next_action.get("score")) >= 0.35
    )
    opportunity_remains = bool(frontier.get("frontier_open")) or has_candidate
    authorized = bool(harness_stop.get("should_stop"))
    return {
        "authorized": authorized,
        "reason": harness_stop.get("reason"),
        "opportunity_remains": opportunity_remains,
        "summary": (
            "Deterministic validator authorizes stop; no adjacent bottleneck probe remains."
            if authorized
            else "Validator does not authorize stop; LLM should_stop is advisory only."
        ),
    }


def _harness_proposal_decision(
    study: StudySpec,
    window_summary: dict[str, Any],
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
    *,
    experiment_plan: dict[str, Any] | None = None,
) -> dict[str, Any]:
    default = {
        "should_propose": False,
        "reason": "no_deterministic_harness_proposal",
        "diagnosis": (
            "The profile-driven candidate set has no untested action above the "
            "proposal threshold; defer to the LLM proposal policy."
        ),
        "config_patch": {"env_patch": {}, "flag_patch": {}},
        "expected_effects": [],
    }
    tested_signatures = {
        _effective_config_signature(
            study,
            item.get("config_patch") if isinstance(item, dict) else None,
        )
        for item in recent_diagnostics
    }
    tested_signatures.update(_state_tested_signatures(study, state))
    if experiment_plan is not None:
        next_action = experiment_plan.get("next_action")
        if isinstance(next_action, dict) and _as_float(next_action.get("score")) >= 0.35:
            patch = next_action.get("config_patch")
            if isinstance(patch, dict):
                signature = _effective_config_signature(study, patch)
                candidate_fingerprint = next_action.get("effective_config_fingerprint")
                if signature not in tested_signatures:
                    return {
                        "should_propose": True,
                        "reason": str(next_action.get("action_id") or "profile_driven_candidate"),
                        "diagnosis": str(next_action.get("hypothesis") or "Profile-driven harness candidate."),
                        "config_patch": patch,
                        "expected_effects": [
                            str(item)
                            for item in next_action.get("expected_effects", [])
                            if isinstance(item, str)
                        ],
                        "candidate_score": next_action.get("score"),
                        "candidate_id": next_action.get("candidate_id"),
                        "effective_config_fingerprint": candidate_fingerprint,
                        "bottleneck_hypotheses": experiment_plan.get("bottleneck_hypotheses", []),
                    }
    return default


def _topology_frontier_proposal(
    study: StudySpec,
    window_summary: dict[str, Any],
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
    *,
    tested_signatures: set[str],
) -> dict[str, Any]:
    default = {
        "should_propose": False,
        "reason": "topology_frontier_not_applicable",
        "diagnosis": "No untested topology frontier is justified by the current SLO/workload profile.",
        "config_patch": {"env_patch": {}, "flag_patch": {}},
        "expected_effects": [],
    }
    frontier = _topology_frontier_status(study, state, recent_diagnostics, window_summary)
    if not frontier["frontier_open"]:
        return {**default, "reason": frontier["reason"]}
    flag_patch = frontier.get("flag_patch")
    if not isinstance(flag_patch, dict):
        return {**default, "reason": "topology_frontier_patch_missing"}
    signature = _effective_config_signature(study, {"env_patch": {}, "flag_patch": flag_patch})
    if signature in tested_signatures:
        return {**default, "reason": "topology_frontier_already_tested"}
    return {
        "should_propose": True,
        "reason": "topology_frontier_probe_for_slo_pressure",
        "diagnosis": (
            "The current SLO/workload profile still shows latency pressure and a higher "
            "legal TP frontier has not been measured. Validate that topology before "
            "declaring the incumbent converged or spending more trials on local runtime knobs."
        ),
        "config_patch": {"env_patch": {}, "flag_patch": flag_patch},
        "expected_effects": [
            "test whether extra model parallelism improves the configured TTFT/TPOT pass rate",
            "cover the next legal topology frontier before local runtime refinement",
            "avoid overfitting the harness to the first strong incumbent",
        ],
        "frontier": frontier,
    }


def _experiment_plan(
    study: StudySpec,
    window_summary: dict[str, Any],
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
    trial_profiles: list[dict[str, Any]],
    bottleneck_hypotheses: list[dict[str, Any]],
) -> dict[str, Any]:
    tested_signatures = {
        _effective_config_signature(
            study,
            item.get("config_patch") if isinstance(item, dict) else None,
        )
        for item in recent_diagnostics
    }
    tested_signatures.update(_state_tested_signatures(study, state))
    candidate_set = _candidate_action_set(
        study,
        window_summary,
        state,
        recent_diagnostics,
        trial_profiles,
        bottleneck_hypotheses,
        tested_signatures=tested_signatures,
    )
    candidates = list(candidate_set["eligible_candidates"])
    candidates.sort(key=lambda item: _as_float(item.get("score")), reverse=True)
    next_action = candidates[0] if candidates else None
    return {
        "planner_version": "profile-driven-v1",
        "bottleneck_hypotheses": bottleneck_hypotheses,
        "candidate_actions": candidates[:8],
        "candidate_set": {
            **candidate_set,
            "eligible_candidates": candidates[:8],
            "blocked_candidates": candidate_set["blocked_candidates"][:16],
        },
        "next_action": next_action,
        "stop_ready": next_action is None or _as_float(next_action.get("score")) < 0.35,
        "stop_rationale": (
            "no untested high-value candidate remains"
            if not candidates or _as_float(candidates[0].get("score")) < 0.35
            else "continue with the highest-scoring measured hypothesis"
        ),
    }


def _candidate_action_set(
    study: StudySpec,
    window_summary: dict[str, Any],
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
    trial_profiles: list[dict[str, Any]],
    bottleneck_hypotheses: list[dict[str, Any]],
    *,
    tested_signatures: set[str],
) -> dict[str, Any]:
    blocked_candidates: list[dict[str, Any]] = []
    eligible_candidates = _candidate_actions(
        study,
        window_summary,
        state,
        recent_diagnostics,
        trial_profiles,
        bottleneck_hypotheses,
        tested_signatures=tested_signatures,
        blocked_candidates=blocked_candidates,
    )
    _annotate_candidate_signatures(study, eligible_candidates)
    eligible_candidates.sort(key=lambda item: _as_float(item.get("score")), reverse=True)
    blocked_summary = _blocked_reason_summary(blocked_candidates)
    candidate_set_hash = _candidate_set_hash(eligible_candidates, blocked_candidates)
    return {
        "version": "candidate-set-v1",
        "signature_semantics": (
            "effective_config_fingerprint = sha256(normalized effective full-config JSON)"
        ),
        "candidate_set_hash": candidate_set_hash,
        "tested_signature_count": len(tested_signatures),
        "eligible_count": len(eligible_candidates),
        "blocked_count": len(blocked_candidates),
        "blocked_reason_summary": blocked_summary,
        "eligible_candidates": eligible_candidates,
        "blocked_candidates": blocked_candidates,
    }


def _candidate_actions(
    study: StudySpec,
    window_summary: dict[str, Any],
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
    trial_profiles: list[dict[str, Any]],
    bottleneck_hypotheses: list[dict[str, Any]],
    *,
    tested_signatures: set[str],
    blocked_candidates: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    if not recent_diagnostics:
        return []
    anchor = _anchor_profile(study, state, recent_diagnostics, trial_profiles)
    if anchor is None:
        return []
    top_bottleneck = (
        str(bottleneck_hypotheses[0]["name"])
        if bottleneck_hypotheses
        else str(anchor.get("probe_profile", {}).get("active_bottleneck") or "")
    )
    candidates: list[dict[str, Any]] = []
    candidates.extend(
        _topology_candidate_actions(
            study,
            anchor,
            top_bottleneck,
            bottleneck_hypotheses,
            tested_signatures,
            blocked_candidates,
        )
    )
    candidates.extend(
        _frontier_delta_projection_actions(
            study,
            state,
            trial_profiles,
            top_bottleneck,
            bottleneck_hypotheses,
            tested_signatures,
            blocked_candidates,
        )
    )
    candidates.extend(
        _runtime_candidate_actions(
            study,
            window_summary,
            anchor,
            top_bottleneck,
            bottleneck_hypotheses,
            recent_diagnostics,
            tested_signatures,
            blocked_candidates,
        )
    )
    return candidates


def _anchor_profile(
    study: StudySpec,
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
    trial_profiles: list[dict[str, Any]],
) -> dict[str, Any] | None:
    if state.best_trial_id:
        for profile in reversed(trial_profiles):
            if profile.get("trial_id") == state.best_trial_id:
                return profile
    for profile in reversed(trial_profiles):
        if profile.get("status") == "completed":
            return profile
    return trial_profiles[-1] if trial_profiles else None


def _topology_candidate_actions(
    study: StudySpec,
    anchor: dict[str, Any],
    top_bottleneck: str,
    bottleneck_hypotheses: list[dict[str, Any]],
    tested_signatures: set[str],
    blocked_candidates: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    if not ({"tensor-parallel-size", "data-parallel-size"} & set(study.engine.tunable_flags)):
        return []
    anchor_flags = _effective_flags_for_item(study, anchor)
    current_tp = _parse_int_like(anchor_flags.get("tensor-parallel-size"), default=1)
    current_dp = _parse_int_like(anchor_flags.get("data-parallel-size"), default=1)
    current_ep = _parse_int_like(anchor_flags.get("expert-parallel-size"), default=1)
    current_enable_ep = bool(anchor_flags.get("enable-expert-parallel", False))
    legal = _legal_topology_points(
        study,
        current_tp=current_tp,
        current_dp=current_dp,
        current_ep=current_ep,
        current_enable_ep=current_enable_ep,
    )
    adjacent_lower_tp = None
    if (
        study.trace.request_mode != "decode_only"
        and not _anchor_has_topology_patch(anchor)
        and not _has_unmeasured_higher_tp_candidate(
            study,
            legal,
            current_tp=current_tp,
            current_dp=current_dp,
            tested_signatures=tested_signatures,
        )
    ):
        adjacent_lower_tp = _adjacent_lower_tp_candidate(
            legal,
            current_tp=current_tp,
            current_dp=current_dp,
        )
    actions: list[dict[str, Any]] = []
    for point in legal:
        if point["tensor-parallel-size"] == current_tp and point["data-parallel-size"] == current_dp:
            continue
        patch = _topology_patch(study, point)
        signature = _effective_config_signature(study, {"env_patch": {}, "flag_patch": patch})
        if signature in tested_signatures:
            blocked_candidates.append(
                _blocked_candidate(
                    action_id=_topology_action_id(current_tp, current_dp, point),
                    knob_family="topology",
                    config_patch={"env_patch": {}, "flag_patch": patch},
                    blocked_reason="blocked_noop_or_repeat_effective_full_config",
                    effective_config_signature=signature,
                )
            )
            continue
        score, factors = _score_topology_candidate(
            top_bottleneck,
            bottleneck_hypotheses,
            request_mode=study.trace.request_mode,
            current_tp=current_tp,
            current_dp=current_dp,
            candidate_tp=point["tensor-parallel-size"],
            candidate_dp=point["data-parallel-size"],
        )
        if (
            adjacent_lower_tp is not None
            and current_tp > 2
            and point["tensor-parallel-size"] == adjacent_lower_tp
            and point["data-parallel-size"] == current_dp
        ):
            score = max(score, 0.74)
            factors["bad_start_topology_bracket"] = 0.74
        if score <= 0:
            blocked_candidates.append(
                _blocked_candidate(
                    action_id=_topology_action_id(current_tp, current_dp, point),
                    knob_family="topology",
                    config_patch={"env_patch": {}, "flag_patch": patch},
                    blocked_reason="blocked_non_positive_candidate_score",
                    effective_config_signature=signature,
                    score=round(score, 4),
                    score_factors=factors,
                )
            )
            continue
        action_id = _topology_action_id(current_tp, current_dp, point)
        actions.append(
            {
                "action_id": action_id,
                "knob_family": "topology",
                "score": round(score, 4),
                "score_factors": factors,
                "config_patch": {"env_patch": {}, "flag_patch": patch},
                "hypothesis": _topology_hypothesis(
                    top_bottleneck,
                    current_tp=current_tp,
                    current_dp=current_dp,
                    candidate_tp=point["tensor-parallel-size"],
                    candidate_dp=point["data-parallel-size"],
                ),
                "expected_effects": [
                    "measure whether topology changes relieve the ranked bottleneck",
                    "compare request_rate_per_gpu under the configured SLO, not raw throughput alone",
                    "reject this hypothesis if latency improves but per-GPU throughput regresses materially",
                ],
            }
        )
    return actions


def _runtime_candidate_actions(
    study: StudySpec,
    window_summary: dict[str, Any],
    anchor: dict[str, Any],
    top_bottleneck: str,
    bottleneck_hypotheses: list[dict[str, Any]],
    recent_diagnostics: list[dict[str, Any]],
    tested_signatures: set[str],
    blocked_candidates: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    tunable = set(study.engine.tunable_flags)
    anchor_flags = _effective_flags_for_item(study, anchor)
    topology_patch = _preserve_topology_patch(study, anchor_flags)
    runtime_base_patch = {**topology_patch, **_preserve_runtime_patch(study, anchor_flags)}
    actions: list[dict[str, Any]] = []
    seen_signatures = set(tested_signatures)

    cur_tp = _parse_int_like(anchor_flags.get("tensor-parallel-size"), default=1)
    cur_dp = _parse_int_like(anchor_flags.get("data-parallel-size"), default=1)
    # Topology-before-runtime: gpu-mem-util / raising max-num-seqs are micro-tuning that is
    # only justified once no untested TP increase remains. At an intermediate TP (e.g. TP2
    # while TP4 is still reachable and untried) a latency bottleneck must still be answered
    # by climbing TP, not a runtime tweak -- otherwise runtime tuning preempts the frontier.
    higher_tp_patch = _higher_tp_frontier_patch(
        study,
        current_tp=cur_tp,
        current_dp=cur_dp,
    )
    tp_frontier_open = (
        higher_tp_patch is not None
        and _effective_config_signature(study, {"env_patch": {}, "flag_patch": higher_tp_patch})
        not in tested_signatures
    )
    topology_settled = not tp_frontier_open

    prefill_scheduler_actions = _prefill_scheduler_candidate_actions(
        study,
        window_summary,
        anchor_flags,
        runtime_base_patch,
        top_bottleneck,
        bottleneck_hypotheses,
        topology_settled=topology_settled,
        seen_signatures=seen_signatures,
        blocked_candidates=blocked_candidates,
    )
    actions.extend(prefill_scheduler_actions)
    prefill_scheduler_candidate_available = bool(prefill_scheduler_actions)

    if (
        "max-num-batched-tokens" in tunable
        and _anchor_has_topology_patch(anchor)
        and recent_diagnostics
        and recent_diagnostics[-1].get("trial_id") == anchor.get("trial_id")
        and cur_tp > 1
        and not bottleneck_hypotheses
        and not prefill_scheduler_candidate_available
    ):
        current_mbt = _parse_int_like(anchor_flags.get("max-num-batched-tokens"), default=0)
        target_mbt = (
            _initial_mbt_from_window(window_summary)
            if current_mbt <= 0
            else _next_mbt_step(current_mbt)
        )
        if target_mbt is not None:
            action_id = (
                "same_topology_runtime_seed_after_tp_incumbent"
                if current_mbt <= 0
                else "same_topology_mbt_growth_after_feasible_runtime_incumbent"
            )
            patch = {**runtime_base_patch, "max-num-batched-tokens": target_mbt}
            if "enable-chunked-prefill" in tunable:
                patch["enable-chunked-prefill"] = True
            signature = _effective_config_signature(
                study,
                {"env_patch": {}, "flag_patch": patch},
            )
            if signature in seen_signatures:
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id=action_id,
                        knob_family="same-topology-runtime",
                        config_patch={"env_patch": {}, "flag_patch": patch},
                        blocked_reason="blocked_noop_or_repeat_effective_full_config",
                        effective_config_signature=signature,
                    )
                )
            else:
                actions.append(
                    _runtime_action(
                        action_id=action_id,
                        knob_family="same-topology-runtime",
                        score=0.46 + _information_gain(bottleneck_hypotheses, "runtime"),
                        patch=patch,
                        hypothesis=(
                            "A measured TP incumbent improved request_rate_per_gpu; seed "
                            "same-topology batching controls before changing another topology axis."
                        ),
                        expected_effects=[
                            "preserve the incumbent topology while increasing batching headroom",
                            "reject if the runtime seed fails to improve request_rate_per_gpu",
                        ],
                    )
                )
                seen_signatures.add(signature)

    if "max-num-batched-tokens" in tunable and not prefill_scheduler_candidate_available:
        current_mbt = _parse_int_like(anchor_flags.get("max-num-batched-tokens"), default=0)
        mbt_targets: list[tuple[str, int]] = []
        if top_bottleneck == "ttft_prefill":
            target = (
                _initial_mbt_from_window(window_summary)
                if current_mbt <= 0
                else _next_mbt_step(current_mbt)
            )
            if target is not None:
                mbt_targets.append(("raise_mbt", target))
        elif top_bottleneck == "decode_tpot" and current_mbt > 8192:
            mbt_targets.append(("lower_mbt", max(8192, current_mbt // 2)))
        for action_id, target in mbt_targets:
            patch = {**runtime_base_patch, "max-num-batched-tokens": target}
            signature = _effective_config_signature(study, {"env_patch": {}, "flag_patch": patch})
            if signature in seen_signatures:
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id=action_id,
                        knob_family="max-num-batched-tokens",
                        config_patch={"env_patch": {}, "flag_patch": patch},
                        blocked_reason="blocked_noop_or_repeat_effective_full_config",
                        effective_config_signature=signature,
                    )
                )
                continue
            relief = 0.24 if top_bottleneck == "ttft_prefill" else 0.14
            actions.append(
                _runtime_action(
                    action_id=action_id,
                    knob_family="max-num-batched-tokens",
                    score=relief + _information_gain(bottleneck_hypotheses, "runtime"),
                    patch=patch,
                    hypothesis=(
                        "Adjust max-num-batched-tokens to test whether batching, not topology, "
                        "is limiting the active latency objective."
                    ),
                    expected_effects=[
                        "change prefill/decode batching pressure on the incumbent topology",
                        "confirm if the latency knee moves without requiring another topology change",
                    ],
                )
            )
            seen_signatures.add(signature)

    if "max-num-seqs" in tunable:
        current_mns = _parse_int_like(anchor_flags.get("max-num-seqs"), default=0)
        max_num_seqs_tested = any(
            "max-num-seqs" in (
                ((item.get("config_patch") or {}).get("flag_patch") or {})
                if isinstance(item.get("config_patch"), dict)
                else {}
            )
            for item in recent_diagnostics
        )
        mns_targets: list[tuple[str, int]] = []
        if top_bottleneck == "admission_or_queueing":
            target = max(8, int(current_mns * 1.5)) if current_mns > 0 else 64
            mns_targets.append(("raise_max_num_seqs", _round_up_to_multiple(target, 8)))
        elif top_bottleneck == "decode_tpot":
            if current_mns > 8:
                mns_targets.append(("lower_max_num_seqs", max(8, current_mns // 2)))
            # Decode concurrency can also be too low: once topology is settled, raising
            # max-num-seqs exploits decode parallelism when the incumbent has SLO headroom.
            # The incumbent guard keeps it only if per-GPU rate improves.
            if topology_settled:
                raise_target = _round_up_to_multiple(
                    max(16, int(current_mns * 1.5)) if current_mns > 0 else 48, 8
                )
                mns_targets.append(("raise_max_num_seqs", raise_target))
        elif top_bottleneck == "ttft_prefill" and topology_settled and not max_num_seqs_tested:
            # Prefill-heavy TTFT can still be admission/concurrency limited after TP and
            # max-num-batched-tokens probes settle. Try a modest same-topology seq cap
            # increase before letting convergence guards declare the incumbent final.
            target = _round_up_to_multiple(
                max(16, int(current_mns * 1.5)) if current_mns > 0 else 64, 8
            )
            mns_targets.append(("raise_max_num_seqs", target))
        for action_id, target in mns_targets:
            patch = {**runtime_base_patch, "max-num-seqs": target}
            signature = _effective_config_signature(study, {"env_patch": {}, "flag_patch": patch})
            if signature in seen_signatures:
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id=action_id,
                        knob_family="max-num-seqs",
                        config_patch={"env_patch": {}, "flag_patch": patch},
                        blocked_reason="blocked_noop_or_repeat_effective_full_config",
                        effective_config_signature=signature,
                    )
                )
                continue
            if top_bottleneck in {"decode_tpot", "admission_or_queueing"}:
                relief = 0.25
            elif top_bottleneck == "ttft_prefill":
                relief = 0.3
            else:
                relief = 0.08
            actions.append(
                _runtime_action(
                    action_id=action_id,
                    knob_family="max-num-seqs",
                    score=relief + _information_gain(bottleneck_hypotheses, "runtime"),
                    patch=patch,
                    hypothesis=(
                        "Adjust max-num-seqs to test whether concurrency/admission pressure "
                        "is the limiting factor under the configured SLO."
                    ),
                    expected_effects=[
                        "change prefill/decode admission concurrency on the incumbent topology",
                        "confirm if latency or queueing pressure is caused by sequence concurrency",
                    ],
                )
            )
            seen_signatures.add(signature)

    if (
        top_bottleneck == "ttft_prefill"
        and topology_settled
        and "max-num-batched-tokens" in tunable
        and "max-num-seqs" in tunable
        and max_num_seqs_tested
        and not prefill_scheduler_candidate_available
    ):
        current_mbt = _parse_int_like(anchor_flags.get("max-num-batched-tokens"), default=0)
        current_mns = _parse_int_like(anchor_flags.get("max-num-seqs"), default=0)
        if current_mbt > 0:
            window_target = _initial_mbt_from_window(window_summary)
            step_target = _next_mbt_step(current_mbt) or current_mbt
            mbt_target = min(
                32768,
                max(
                    step_target,
                    min(window_target, _round_up_to_multiple(current_mbt * 2, 1024)),
                ),
            )
        else:
            mbt_target = _initial_mbt_from_window(window_summary)
        mns_target = _round_up_to_multiple(
            max(16, int(current_mns * 1.5)) if current_mns > 0 else 64, 8
        )
        if mbt_target > 0 and (mbt_target != current_mbt or mns_target != current_mns):
            patch = {
                **runtime_base_patch,
                "max-num-batched-tokens": mbt_target,
                "max-num-seqs": mns_target,
            }
            signature = _effective_config_signature(study, {"env_patch": {}, "flag_patch": patch})
            if signature not in seen_signatures:
                actions.append(
                    _runtime_action(
                        action_id="raise_mbt_and_max_num_seqs",
                        knob_family="prefill-runtime-interaction",
                        score=0.38
                        + _information_gain(bottleneck_hypotheses, "runtime"),
                        patch=patch,
                        hypothesis=(
                            "Jointly raise max-num-batched-tokens and max-num-seqs to test "
                            "whether prefill batching headroom and admission concurrency only "
                            "help when adjusted together."
                        ),
                        expected_effects=[
                            "preserve the incumbent topology while changing coupled prefill runtime limits",
                            "confirm whether separate MBT or sequence-cap probes masked an interaction",
                        ],
                    )
                )
                seen_signatures.add(signature)
            else:
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id="raise_mbt_and_max_num_seqs",
                        knob_family="prefill-runtime-interaction",
                        config_patch={"env_patch": {}, "flag_patch": patch},
                        blocked_reason="blocked_noop_or_repeat_effective_full_config",
                        effective_config_signature=signature,
                    )
                )

    if (
        "enable-chunked-prefill" in tunable
        and top_bottleneck == "ttft_prefill"
        and not prefill_scheduler_candidate_available
    ):
        current = bool(anchor_flags.get("enable-chunked-prefill", False))
        if not current:
            patch = {**runtime_base_patch, "enable-chunked-prefill": True}
            signature = _effective_config_signature(study, {"env_patch": {}, "flag_patch": patch})
            if signature not in seen_signatures:
                actions.append(
                    _runtime_action(
                        action_id="enable_chunked_prefill",
                        knob_family="enable-chunked-prefill",
                        score=0.2 + _information_gain(bottleneck_hypotheses, "runtime"),
                        patch=patch,
                        hypothesis=(
                            "Enable chunked prefill to test whether long-prefill head-of-line "
                            "blocking is driving TTFT failures."
                        ),
                        expected_effects=[
                            "reduce long-prefill interference for mixed-length chat windows",
                            "reject if chunking overhead worsens request_rate_per_gpu",
                        ],
                    )
                )
                seen_signatures.add(signature)
            else:
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id="enable_chunked_prefill",
                        knob_family="enable-chunked-prefill",
                        config_patch={"env_patch": {}, "flag_patch": patch},
                        blocked_reason="blocked_noop_or_repeat_effective_full_config",
                        effective_config_signature=signature,
                    )
                )

    if (
        "gpu-memory-utilization" in tunable
        and topology_settled
        and top_bottleneck in {"decode_tpot", "admission_or_queueing", "ttft_prefill"}
    ):
        target = _next_gpu_memory_utilization_target(
            study,
            anchor_flags,
            recent_diagnostics,
            anchor_rate_per_gpu=_profile_request_rate_per_gpu(anchor),
        )
        if target is not None:
            patch = {**runtime_base_patch, "gpu-memory-utilization": target}
            signature = _effective_config_signature(study, {"env_patch": {}, "flag_patch": patch})
            if signature not in seen_signatures:
                actions.append(
                    _runtime_action(
                        action_id="raise_gpu_memory_utilization",
                        knob_family="gpu-memory-utilization",
                        score=0.5 + _information_gain(bottleneck_hypotheses, "runtime"),
                        patch=patch,
                        hypothesis=(
                            "Raise gpu-memory-utilization on the settled incumbent topology "
                            "to test whether extra KV-cache headroom moves the SLO frontier."
                        ),
                        expected_effects=[
                            "add KV-cache blocks for higher concurrency on the incumbent topology",
                            "reject if the higher memory target regresses request_rate_per_gpu or fails to launch",
                        ],
                    )
                )
                seen_signatures.add(signature)
            else:
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id="raise_gpu_memory_utilization",
                        knob_family="gpu-memory-utilization",
                        config_patch={"env_patch": {}, "flag_patch": patch},
                        blocked_reason="blocked_noop_or_repeat_effective_full_config",
                        effective_config_signature=signature,
                    )
                )
    production_covered_knobs = _candidate_action_runtime_knobs(actions)
    actions.extend(
        _descriptor_runtime_candidate_actions(
            study,
            window_summary,
            anchor_flags,
            runtime_base_patch,
            top_bottleneck,
            bottleneck_hypotheses,
            recent_diagnostics,
            production_covered_knobs=production_covered_knobs,
            topology_settled=topology_settled,
            seen_signatures=seen_signatures,
            blocked_candidates=blocked_candidates,
        )
    )
    return actions


def _descriptor_runtime_candidate_actions(
    study: StudySpec,
    window_summary: dict[str, Any],
    anchor_flags: dict[str, Any],
    runtime_base_patch: dict[str, Any],
    top_bottleneck: str,
    bottleneck_hypotheses: list[dict[str, Any]],
    recent_diagnostics: list[dict[str, Any]],
    *,
    production_covered_knobs: set[str],
    topology_settled: bool,
    seen_signatures: set[str],
    blocked_candidates: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    descriptors = _engine_knob_descriptors(study)
    if not descriptors:
        return []
    evidence_weights = _mechanism_evidence_weights(
        study,
        window_summary,
        top_bottleneck,
        bottleneck_hypotheses,
    )
    candidates = coordinate_line_search_candidates(
        current_config=anchor_flags,
        descriptors=tuple(
            descriptor
            for descriptor in descriptors
            if descriptor.name in anchor_flags and anchor_flags.get(descriptor.name) is not None
        ),
        evidence_weights=evidence_weights,
        policy=CoordinateSearchPolicy(
            initial_relative_step=1.0,
            initial_fraction_step=0.1,
            step_multipliers=(1.0, 2.0, 4.0),
            min_score=0.25,
        ),
    )
    actions: list[dict[str, Any]] = []
    for candidate in candidates:
        patch = {**runtime_base_patch, **candidate.patch}
        signature = _effective_config_signature(study, {"env_patch": {}, "flag_patch": patch})
        if candidate.knob in production_covered_knobs:
            blocked_candidates.append(
                _blocked_candidate(
                    action_id=candidate.action_id,
                    knob_family=f"descriptor:{candidate.mechanism}",
                    config_patch={"env_patch": {}, "flag_patch": patch},
                    blocked_reason="blocked_descriptor_knob_has_production_candidate",
                    effective_config_signature=signature,
                    score=candidate.score,
                    score_factors=candidate.score_factors,
                )
            )
            continue
        if _recent_runtime_knob_probe_count(candidate.knob, recent_diagnostics) >= 2:
            blocked_candidates.append(
                _blocked_candidate(
                    action_id=candidate.action_id,
                    knob_family=f"descriptor:{candidate.mechanism}",
                    config_patch={"env_patch": {}, "flag_patch": patch},
                    blocked_reason="blocked_descriptor_knob_already_covered_by_recent_probes",
                    effective_config_signature=signature,
                    score=candidate.score,
                    score_factors=candidate.score_factors,
                )
            )
            continue
        if signature in seen_signatures:
            blocked_candidates.append(
                _blocked_candidate(
                    action_id=candidate.action_id,
                    knob_family=f"descriptor:{candidate.mechanism}",
                    config_patch={"env_patch": {}, "flag_patch": patch},
                    blocked_reason="blocked_noop_or_repeat_effective_full_config",
                    effective_config_signature=signature,
                    score=candidate.score,
                    score_factors=candidate.score_factors,
                )
            )
            continue
        score_cap = 0.52 if topology_settled else 0.45
        score = min(
            score_cap,
            0.22
            + candidate.score * 0.65
            + _information_gain(bottleneck_hypotheses, "runtime") * 0.5,
        )
        actions.append(
            _runtime_action(
                action_id=candidate.action_id,
                knob_family=f"descriptor:{candidate.mechanism}",
                score=score,
                score_factors={
                    **candidate.score_factors,
                    "planner_score": candidate.score,
                    "information_gain": round(
                        _information_gain(bottleneck_hypotheses, "runtime"), 4
                    ),
                    "regression_risk": round(
                        candidate.score_factors.get("direction_risk", 0.0)
                        + candidate.score_factors.get("step_risk", 0.0),
                        4,
                    ),
                },
                patch=patch,
                hypothesis=(
                    "Descriptor-driven coordinate line search over a serving mechanism "
                    f"({candidate.mechanism}) selected {candidate.knob} "
                    f"{candidate.direction} without using a knob-specific candidate rule."
                ),
                expected_effects=[
                    "test one declarative knob mechanism while preserving the anchor topology",
                    "treat a positive result as transferable mechanism evidence, not a fixed recipe",
                    "reject or shrink this direction if the measured per-GPU SLO objective does not improve",
                ],
            )
        )
        seen_signatures.add(signature)
    return actions[:8]


def _candidate_action_runtime_knobs(actions: list[dict[str, Any]]) -> set[str]:
    knobs: set[str] = set()
    for action in actions:
        patch = action.get("config_patch")
        if not isinstance(patch, dict):
            continue
        flag_patch = patch.get("flag_patch")
        if not isinstance(flag_patch, dict):
            continue
        knobs.update(set(flag_patch) & _RUNTIME_KEYS)
    return knobs


def _recent_runtime_knob_probe_count(
    knob: str,
    recent_diagnostics: list[dict[str, Any]],
) -> int:
    count = 0
    for item in recent_diagnostics:
        patch = item.get("config_patch")
        if not isinstance(patch, dict):
            continue
        flag_patch = patch.get("flag_patch")
        if isinstance(flag_patch, dict) and knob in flag_patch:
            count += 1
    return count


def _engine_knob_descriptors(study: StudySpec):
    engine_name = str(study.engine.engine_name or "").lower()
    if engine_name == "vllm":
        return default_vllm_descriptors(tunable_flags=study.engine.tunable_flags)
    return ()


def _mechanism_evidence_weights(
    study: StudySpec,
    window_summary: dict[str, Any],
    top_bottleneck: str,
    bottleneck_hypotheses: list[dict[str, Any]],
) -> dict[str, float]:
    confidence = _hypothesis_confidence(bottleneck_hypotheses, top_bottleneck)
    if confidence <= 0:
        confidence = 0.35
    weights: dict[str, float] = {}

    def add(mechanism: str, weight: float) -> None:
        weights[mechanism] = max(weights.get(mechanism, 0.0), round(weight, 4))

    if top_bottleneck == "ttft_prefill":
        add("prefill_scheduling", 0.78 * confidence)
        add("prefill_tail_latency", 0.58 * confidence)
        add("admission_capacity", 0.64 * confidence)
        add("kv_memory_capacity", 0.58 * confidence)
    elif top_bottleneck == "admission_or_queueing":
        add("admission_capacity", 0.82 * confidence)
        add("kv_memory_capacity", 0.62 * confidence)
        add("decode_batching", 0.48 * confidence)
        add("prefill_scheduling", 0.36 * confidence)
    elif top_bottleneck == "decode_tpot":
        add("kv_memory_pressure", 0.74 * confidence)
        add("kv_memory_capacity", 0.56 * confidence)
        add("decode_batching", 0.48 * confidence)
    elif top_bottleneck == "launch_or_memory":
        add("launch_feasibility", 0.82 * confidence)
        add("kv_memory_pressure", 0.56 * confidence)
    else:
        add("admission_capacity", 0.32)
        add("kv_memory_capacity", 0.28)

    prompt_p95 = _as_float(window_summary.get("prompt_tokens_p95"))
    tail_ratio = _as_float(window_summary.get("prompt_tail_ratio_p95_p50"))
    if _length_regime(prompt_p95, tail_ratio) != "short_or_moderate":
        add("prefill_scheduling", 0.36)
    if study.trace.request_mode == "decode_only":
        add("decode_batching", 0.36)
    return weights


def _frontier_delta_projection_actions(
    study: StudySpec,
    state: StudyState,
    trial_profiles: list[dict[str, Any]],
    top_bottleneck: str,
    bottleneck_hypotheses: list[dict[str, Any]],
    tested_signatures: set[str],
    blocked_candidates: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    if not (set(study.engine.tunable_flags) & _RUNTIME_KEYS):
        return []
    projection_profiles = _frontier_projection_profiles(study, state, trial_profiles)
    anchors = _pareto_frontier_anchor_profiles(study, projection_profiles)
    if len(anchors) < 2:
        return []
    deltas = _positive_runtime_delta_records(study, projection_profiles)
    if not deltas:
        return []

    actions: list[dict[str, Any]] = []
    seen_signatures = set(tested_signatures)
    incumbent_rate = max(
        (
            _profile_request_rate_per_gpu(profile)
            for profile in projection_profiles
            if profile.get("status") == "completed"
        ),
        default=0.0,
    )
    for delta in deltas:
        source_topology_key = delta["source_topology_key"]
        source_parallel = int(delta["source_parallel_size"])
        source_rate = _as_float(delta["source_rate_per_gpu"])
        source_gain = _as_float(delta["gain"])
        for target in anchors:
            target_flags = _effective_flags_for_item(study, target)
            target_topology_key = _topology_key(target_flags)
            if target_topology_key == source_topology_key:
                continue
            target_parallel = _topology_parallel_size(target_flags)
            target_rate = _profile_request_rate_per_gpu(target)
            runtime_delta = {
                key: value
                for key, value in delta["runtime_delta"].items()
                if key in study.engine.tunable_flags and target_flags.get(key) != value
            }
            action_id = (
                f"project_runtime_delta_from_{delta['source_trial_id']}"
                f"_to_{target.get('trial_id')}"
            )
            if not runtime_delta:
                target_topology_patch = _explicit_topology_patch(study, target_flags)
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id=action_id,
                        knob_family="frontier-delta-projection",
                        config_patch={
                            "env_patch": {},
                            "flag_patch": {
                                **target_topology_patch,
                                **_preserve_runtime_patch(study, target_flags),
                            },
                        },
                        blocked_reason="blocked_noop_or_repeat_effective_full_config",
                        effective_config_signature=_effective_config_signature(
                            study,
                            {
                                "env_patch": {},
                                "flag_patch": {
                                    **target_topology_patch,
                                    **_preserve_runtime_patch(study, target_flags),
                                },
                            },
                        ),
                    )
                )
                continue

            patch = {
                **_explicit_topology_patch(study, target_flags),
                **_preserve_runtime_patch(study, target_flags),
                **runtime_delta,
            }
            signature = _effective_config_signature(
                study,
                {"env_patch": {}, "flag_patch": patch},
            )
            if source_rate + EPSILON < target_rate + max(
                _FRONTIER_DELTA_MIN_ABS_GAIN,
                target_rate * _FRONTIER_DELTA_MIN_REL_GAIN,
            ):
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id=action_id,
                        knob_family="frontier-delta-projection",
                        config_patch={"env_patch": {}, "flag_patch": patch},
                        blocked_reason="blocked_delta_source_not_above_target_anchor",
                        effective_config_signature=signature,
                    )
                )
                continue
            if (
                target_parallel < source_parallel
                and delta["baseline_rate_per_gpu"] > 0
                and target_rate + EPSILON
                < delta["baseline_rate_per_gpu"]
                * _FRONTIER_DELTA_TARGET_BASELINE_COMPETITIVE_FLOOR
            ):
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id=action_id,
                        knob_family="frontier-delta-projection",
                        config_patch={"env_patch": {}, "flag_patch": patch},
                        blocked_reason="blocked_target_anchor_not_competitive_with_source_baseline",
                        effective_config_signature=signature,
                    )
                )
                continue
            if (
                incumbent_rate > 0
                and target_parallel >= source_parallel
                and target_rate + source_gain + EPSILON
                < incumbent_rate * _FRONTIER_DELTA_PROJECTED_INCUMBENT_FLOOR
            ):
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id=action_id,
                        knob_family="frontier-delta-projection",
                        config_patch={"env_patch": {}, "flag_patch": patch},
                        blocked_reason="blocked_projected_gain_below_incumbent_floor",
                        effective_config_signature=signature,
                    )
                )
                continue
            if signature in seen_signatures:
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id=action_id,
                        knob_family="frontier-delta-projection",
                        config_patch={"env_patch": {}, "flag_patch": patch},
                        blocked_reason="blocked_noop_or_repeat_effective_full_config",
                        effective_config_signature=signature,
                    )
                )
                continue

            score, factors = _score_frontier_delta_projection(
                top_bottleneck=top_bottleneck,
                source_gain=delta["gain"],
                source_baseline_rate=delta["baseline_rate_per_gpu"],
                source_rate=delta["source_rate_per_gpu"],
                source_parallel=source_parallel,
                target_rate=target_rate,
                target_parallel=target_parallel,
                bottleneck_hypotheses=bottleneck_hypotheses,
            )
            if score <= 0:
                blocked_candidates.append(
                    _blocked_candidate(
                        action_id=action_id,
                        knob_family="frontier-delta-projection",
                        config_patch={"env_patch": {}, "flag_patch": patch},
                        blocked_reason="blocked_non_positive_candidate_score",
                        effective_config_signature=signature,
                        score=round(score, 4),
                        score_factors=factors,
                    )
                )
                continue

            actions.append(
                _runtime_action(
                    action_id=action_id,
                    knob_family="frontier-delta-projection",
                    score=score,
                    score_factors=factors,
                    patch=patch,
                    hypothesis=(
                        "A measured runtime-only delta improved one topology anchor; "
                        "apply the same delta to another Pareto anchor to test whether "
                        "the mechanism transfers without inheriting the source topology."
                    ),
                    expected_effects=[
                        "preserve the target anchor topology and only apply measured runtime changes",
                        "separate runtime-mechanism gain from topology-resource tradeoffs",
                        "reject if the projected delta fails to improve request_rate_per_gpu",
                    ],
                )
            )
            seen_signatures.add(signature)

    actions.sort(key=lambda item: _as_float(item.get("score")), reverse=True)
    return actions[:8]


def _frontier_projection_profiles(
    study: StudySpec,
    state: StudyState,
    trial_profiles: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    if len(state.trials) <= len(trial_profiles):
        return trial_profiles
    return _trial_profiles(
        study,
        _recent_trial_diagnostics(state, limit=None),
    )


def _pareto_frontier_anchor_profiles(
    study: StudySpec,
    trial_profiles: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    best_by_topology: dict[tuple[int, int, int, bool], dict[str, Any]] = {}
    for profile in trial_profiles:
        if profile.get("status") != "completed":
            continue
        rate = _profile_request_rate_per_gpu(profile)
        if rate <= 0:
            continue
        flags = _effective_flags_for_item(study, profile)
        topology_key = _topology_key(flags)
        incumbent = best_by_topology.get(topology_key)
        if incumbent is None:
            best_by_topology[topology_key] = profile
            continue
        incumbent_rate = _profile_request_rate_per_gpu(incumbent)
        raw_rate = _as_float(profile.get("performance", {}).get("best_request_rate"))
        incumbent_raw_rate = _as_float(
            incumbent.get("performance", {}).get("best_request_rate")
        )
        if (rate, raw_rate) > (incumbent_rate, incumbent_raw_rate):
            best_by_topology[topology_key] = profile

    anchors = list(best_by_topology.values())
    frontier: list[dict[str, Any]] = []
    for candidate in anchors:
        candidate_rate = _profile_request_rate_per_gpu(candidate)
        candidate_parallel = _topology_parallel_size(
            _effective_flags_for_item(study, candidate)
        )
        dominated = False
        for other in anchors:
            if other is candidate:
                continue
            other_rate = _profile_request_rate_per_gpu(other)
            other_parallel = _topology_parallel_size(_effective_flags_for_item(study, other))
            if (
                other_parallel <= candidate_parallel
                and other_rate + EPSILON >= candidate_rate
                and (other_parallel < candidate_parallel or other_rate > candidate_rate + EPSILON)
            ):
                dominated = True
                break
        if not dominated:
            frontier.append(candidate)

    frontier.sort(
        key=lambda item: (
            _topology_parallel_size(_effective_flags_for_item(study, item)),
            -_profile_request_rate_per_gpu(item),
            str(item.get("trial_id") or ""),
        )
    )
    return frontier


def _positive_runtime_delta_records(
    study: StudySpec,
    trial_profiles: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    completed = [
        profile
        for profile in trial_profiles
        if profile.get("status") == "completed"
        and _profile_request_rate_per_gpu(profile) > 0
    ]
    records: list[dict[str, Any]] = []
    for source in completed:
        source_patch = source.get("config_patch")
        if not isinstance(source_patch, dict):
            continue
        source_flag_patch = source_patch.get("flag_patch")
        if not isinstance(source_flag_patch, dict):
            continue
        source_runtime_keys = set(source_flag_patch) & _RUNTIME_KEYS & set(study.engine.tunable_flags)
        if not source_runtime_keys:
            continue

        source_flags = _effective_flags_for_item(study, source)
        source_topology_key = _topology_key(source_flags)
        source_rate = _profile_request_rate_per_gpu(source)
        best_baseline: dict[str, Any] | None = None
        best_baseline_delta: dict[str, Any] = {}
        best_baseline_gain = 0.0
        for baseline in completed:
            if baseline is source:
                continue
            baseline_flags = _effective_flags_for_item(study, baseline)
            if _topology_key(baseline_flags) != source_topology_key:
                continue
            baseline_rate = _profile_request_rate_per_gpu(baseline)
            runtime_delta = _runtime_delta_patch(
                study,
                baseline_flags=baseline_flags,
                source_flags=source_flags,
                source_flag_patch=source_flag_patch,
            )
            if not runtime_delta:
                continue
            gain = source_rate - baseline_rate
            min_gain = max(
                _FRONTIER_DELTA_MIN_ABS_GAIN,
                baseline_rate * _FRONTIER_DELTA_MIN_REL_GAIN,
            )
            if gain + EPSILON < min_gain:
                continue
            if (
                best_baseline is None
                or baseline_rate > _profile_request_rate_per_gpu(best_baseline)
            ):
                best_baseline = baseline
                best_baseline_delta = runtime_delta
                best_baseline_gain = gain

        if best_baseline is None:
            continue
        records.append(
            {
                "source_trial_id": source.get("trial_id"),
                "baseline_trial_id": best_baseline.get("trial_id"),
                "source_topology_key": source_topology_key,
                "source_parallel_size": _topology_parallel_size(source_flags),
                "runtime_delta": best_baseline_delta,
                "gain": best_baseline_gain,
                "source_rate_per_gpu": source_rate,
                "baseline_rate_per_gpu": _profile_request_rate_per_gpu(best_baseline),
            }
        )

    records.sort(
        key=lambda item: (
            _as_float(item.get("gain")),
            _as_float(item.get("source_rate_per_gpu")),
        ),
        reverse=True,
    )
    return records[:8]


def _runtime_delta_patch(
    study: StudySpec,
    *,
    baseline_flags: dict[str, Any],
    source_flags: dict[str, Any],
    source_flag_patch: dict[str, Any],
) -> dict[str, Any]:
    tunable = set(study.engine.tunable_flags)
    patch: dict[str, Any] = {}
    for key in sorted(_RUNTIME_KEYS & tunable & set(source_flag_patch)):
        if baseline_flags.get(key) == source_flags.get(key):
            continue
        patch[key] = source_flags.get(key)
    return patch


def _score_frontier_delta_projection(
    *,
    top_bottleneck: str,
    source_gain: float,
    source_baseline_rate: float,
    source_rate: float,
    source_parallel: int,
    target_rate: float,
    target_parallel: int,
    bottleneck_hypotheses: list[dict[str, Any]],
) -> tuple[float, dict[str, Any]]:
    relative_gain = source_gain / max(source_baseline_rate, EPSILON)
    measured_delta_evidence = min(0.24, max(0.0, relative_gain) * 0.8)
    if target_parallel < source_parallel:
        resource_transfer = 0.16
    elif target_parallel == source_parallel:
        resource_transfer = 0.06
    else:
        resource_transfer = -0.04
    anchor_quality = min(0.12, max(0.0, target_rate) / max(source_rate, target_rate, EPSILON) * 0.12)
    information_gain = _information_gain(bottleneck_hypotheses, "runtime")
    score = 0.42 + measured_delta_evidence + resource_transfer + anchor_quality + information_gain
    factors = {
        "active_bottleneck": top_bottleneck,
        "measured_delta_gain_per_gpu": round(source_gain, 4),
        "measured_delta_relative_gain": round(relative_gain, 4),
        "measured_delta_evidence": round(measured_delta_evidence, 4),
        "source_parallel_size": source_parallel,
        "target_parallel_size": target_parallel,
        "resource_transfer": round(resource_transfer, 4),
        "target_anchor_rate_per_gpu": round(target_rate, 4),
        "source_delta_rate_per_gpu": round(source_rate, 4),
        "anchor_quality": round(anchor_quality, 4),
        "information_gain": round(information_gain, 4),
        "regression_risk": 0.05,
    }
    return score, factors


def _next_gpu_memory_utilization_target(
    study: StudySpec,
    anchor_flags: dict[str, Any],
    recent_diagnostics: list[dict[str, Any]],
    *,
    anchor_rate_per_gpu: float = 0.0,
) -> float | None:
    current_gmu = _parse_float_like(
        anchor_flags.get("gpu-memory-utilization"), default=0.9
    )
    if current_gmu <= 0 or current_gmu >= _GMU_SAFE_CEILING:
        return None
    anchor_topology = _normalized_topology_flags(anchor_flags)
    successful_gmus: list[float] = [current_gmu]
    failed_gmus: list[float] = []
    for item in recent_diagnostics:
        patch = item.get("config_patch")
        if not isinstance(patch, dict):
            continue
        flag_patch = patch.get("flag_patch")
        if not isinstance(flag_patch, dict) or "gpu-memory-utilization" not in flag_patch:
            continue
        flags = _effective_flags_for_item(study, item)
        if _normalized_topology_flags(flags) != anchor_topology:
            continue
        gmu = _parse_float_like(flag_patch.get("gpu-memory-utilization"), default=0.0)
        if gmu <= 0:
            continue
        if abs(gmu - current_gmu) <= EPSILON:
            continue
        if item.get("status") == "completed":
            rate = _as_float(item.get("best_request_rate_per_gpu"))
            if anchor_rate_per_gpu > 0 and rate <= anchor_rate_per_gpu + EPSILON:
                failed_gmus.append(gmu)
            else:
                successful_gmus.append(gmu)
        elif item.get("status") == "failed":
            failed_gmus.append(gmu)
    climb_from = max(successful_gmus)
    if climb_from < _GMU_NOMINAL_FLOOR:
        target = min(_GMU_SAFE_CEILING, _GMU_NOMINAL_FLOOR)
    else:
        target = min(_GMU_SAFE_CEILING, climb_from + _GMU_STEP)
    target = round(target, 4)
    if target <= climb_from:
        return None
    if any(failed <= target + EPSILON for failed in failed_gmus):
        return None
    return target


def _profile_request_rate_per_gpu(profile: dict[str, Any]) -> float:
    performance = profile.get("performance")
    if isinstance(performance, dict):
        return _as_float(performance.get("best_request_rate_per_gpu"))
    return _as_float(profile.get("best_request_rate_per_gpu"))


def _prefill_scheduler_candidate_actions(
    study: StudySpec,
    window_summary: dict[str, Any],
    anchor_flags: dict[str, Any],
    runtime_base_patch: dict[str, Any],
    top_bottleneck: str,
    bottleneck_hypotheses: list[dict[str, Any]],
    *,
    topology_settled: bool,
    seen_signatures: set[str],
    blocked_candidates: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    tunable = set(study.engine.tunable_flags)
    if not topology_settled:
        return []
    if top_bottleneck not in {"ttft_prefill", "admission_or_queueing"}:
        return []
    if "enable-chunked-prefill" not in tunable or "max-num-batched-tokens" not in tunable:
        return []
    if not _prefill_scheduler_workload_applies(study, window_summary):
        return []

    prompt_scale = _prefill_prompt_scale(window_summary)
    if prompt_scale <= 0:
        return []

    current_mbt = _parse_int_like(anchor_flags.get("max-num-batched-tokens"), default=0)
    current_mns = _parse_int_like(anchor_flags.get("max-num-seqs"), default=0)
    current_chunked = bool(anchor_flags.get("enable-chunked-prefill", False))
    quantum_step = _next_prefill_quantum_step(
        current_mbt,
        prompt_scale,
        top_bottleneck=top_bottleneck,
    )
    admission_step = (
        _next_admission_pressure_step(
            study,
            current_mns,
            top_bottleneck=top_bottleneck,
            quantum_direction=quantum_step["direction"],
        )
        if "max-num-seqs" in tunable
        else None
    )

    admission_target = admission_step["target"] if admission_step is not None else None
    if current_chunked and quantum_step["target"] is None and admission_target is None:
        return []

    patch = {**runtime_base_patch, "enable-chunked-prefill": True}
    if quantum_step["target"] is not None:
        patch["max-num-batched-tokens"] = quantum_step["target"]
    if admission_target is not None:
        patch["max-num-seqs"] = admission_target

    signature = _effective_config_signature(study, {"env_patch": {}, "flag_patch": patch})
    action_id = _prefill_scheduler_action_id(quantum_step["direction"], admission_step)
    if signature in seen_signatures:
        blocked_candidates.append(
            _blocked_candidate(
                action_id=action_id,
                knob_family="prefill-scheduler-interaction",
                config_patch={"env_patch": {}, "flag_patch": patch},
                blocked_reason="blocked_noop_or_repeat_effective_full_config",
                effective_config_signature=signature,
            )
        )
        return []

    current_ratio = current_mbt / prompt_scale if current_mbt > 0 else None
    target_mbt = quantum_step["target"] if quantum_step["target"] is not None else current_mbt
    target_ratio = target_mbt / prompt_scale if target_mbt > 0 else None
    confidence = _hypothesis_confidence(bottleneck_hypotheses, top_bottleneck)
    relief = 0.56 if quantum_step["direction"] == "lower" else 0.42
    if quantum_step["direction"] == "seed":
        relief = 0.38
    if admission_target is not None:
        relief += 0.08 if admission_step and admission_step["direction"] == "lower" else 0.06
    coverage_bonus = 0.0
    if quantum_step["direction"] == "seed" or not current_chunked:
        coverage_bonus = 0.28
    elif quantum_step["target"] is not None or admission_target is not None:
        coverage_bonus = 0.14
    information_gain = _information_gain(bottleneck_hypotheses, "runtime")
    score = relief * max(confidence, 0.35) + information_gain + 0.08 + coverage_bonus
    factors = {
        "expected_bottleneck_relief": round(relief, 4),
        "bottleneck_confidence": round(confidence, 4),
        "information_gain": round(information_gain, 4),
        "uncovered_scheduler_dimension_bonus": round(coverage_bonus, 4),
        "launch_safety": 0.08,
        "regression_risk": 0.06 if current_chunked else 0.1,
        "prefill_quantum_ratio_current": (
            round(current_ratio, 4) if current_ratio is not None else None
        ),
        "prefill_quantum_ratio_target": (
            round(target_ratio, 4) if target_ratio is not None else None
        ),
        "admission_pressure_current": current_mns or None,
        "admission_pressure_target": admission_target,
        "admission_pressure_direction": admission_step["direction"] if admission_step else "hold",
        "admission_pressure_ratio_current": (
            round(admission_step["current_ratio"], 4) if admission_step else None
        ),
        "admission_pressure_ratio_target": (
            round(admission_step["target_ratio"], 4) if admission_step else None
        ),
    }
    actions = [
        _runtime_action(
            action_id=action_id,
            knob_family="prefill-scheduler-interaction",
            score=score,
            score_factors=factors,
            patch=patch,
            hypothesis=(
                "Test the prefill scheduler hypothesis in normalized control space: "
                "chunked prefill changes the scheduler mode, max-num-batched-tokens "
                "controls prefill_quantum_ratio, and max-num-seqs controls admission pressure."
            ),
            expected_effects=[
                "preserve the settled topology while perturbing scheduler controls",
                "reduce long-prefill head-of-line blocking when the prefill quantum is too large",
                "reject this scheduler hypothesis if request_rate_per_gpu does not improve under the configured SLO",
            ],
        )
    ]
    seen_signatures.add(signature)
    return actions


def _prefill_scheduler_workload_applies(
    study: StudySpec,
    window_summary: dict[str, Any],
) -> bool:
    if study.trace.request_mode == "decode_only":
        return False
    prompt_p95 = _as_float(window_summary.get("prompt_tokens_p95"))
    prompt_p99 = _as_float(window_summary.get("prompt_tokens_p99"))
    tail_ratio = _as_float(window_summary.get("prompt_tail_ratio_p95_p50"))
    if prompt_p95 <= 0 and prompt_p99 > 0:
        prompt_p95 = prompt_p99
    if _length_regime(prompt_p95, tail_ratio) == "short_or_moderate":
        return False
    prefix_cache = window_summary.get("prefix_cache")
    cache_ratio = 0.0
    if isinstance(prefix_cache, dict):
        cache_ratio = _as_float(prefix_cache.get("repeated_token_ratio_estimate"))
    return _cache_regime(cache_ratio) != "high_prefix_reuse"


def _prefill_prompt_scale(window_summary: dict[str, Any]) -> float:
    prompt_p95 = _as_float(window_summary.get("prompt_tokens_p95"))
    prompt_p99 = _as_float(window_summary.get("prompt_tokens_p99"))
    if prompt_p95 > 0:
        return prompt_p95
    return prompt_p99


def _next_prefill_quantum_step(
    current_mbt: int,
    prompt_scale: float,
    *,
    top_bottleneck: str,
) -> dict[str, Any]:
    if current_mbt <= 0:
        return {
            "direction": "seed",
            "target": _round_up_to_multiple(int(prompt_scale), 1024),
        }
    ratio = current_mbt / prompt_scale if prompt_scale > 0 else 0.0
    if top_bottleneck == "ttft_prefill" and ratio > _PREFILL_QUANTUM_HEAD_OF_LINE_RATIO:
        target = int((current_mbt * prompt_scale) ** 0.5)
        target = _round_up_to_multiple(target, 1024)
        if target < current_mbt:
            return {"direction": "lower", "target": target}
    if ratio < _PREFILL_QUANTUM_FRAGMENTATION_RATIO:
        target = int((current_mbt * prompt_scale) ** 0.5)
        target = _round_up_to_multiple(target, 1024)
        if target > current_mbt:
            return {"direction": "raise", "target": target}
    return {"direction": "hold", "target": None}


def _next_admission_pressure_step(
    study: StudySpec,
    current_mns: int,
    *,
    top_bottleneck: str,
    quantum_direction: str,
) -> dict[str, Any] | None:
    if current_mns <= 0:
        return None
    target_concurrency = max(int(study.trace.max_concurrency), 1)
    current_ratio = current_mns / target_concurrency
    if top_bottleneck == "admission_or_queueing" and current_mns < target_concurrency:
        target = min(target_concurrency, int(current_mns * _ADMISSION_PRESSURE_STEP_UP))
        target = _round_up_to_multiple(target, 8)
        return {
            "direction": "raise",
            "target": target,
            "current_ratio": current_ratio,
            "target_ratio": target / target_concurrency,
        }
    if (
        top_bottleneck == "ttft_prefill"
        and quantum_direction in {"hold", "raise"}
        and current_mns < target_concurrency
    ):
        target = min(target_concurrency, int(current_mns * _ADMISSION_PRESSURE_STEP_UP))
        target = _round_up_to_multiple(target, 8)
        return {
            "direction": "raise",
            "target": target,
            "current_ratio": current_ratio,
            "target_ratio": target / target_concurrency,
        }
    if top_bottleneck == "ttft_prefill" and current_mns > target_concurrency:
        target = min(current_mns - 8, _round_up_to_multiple(target_concurrency, 8))
        if target > 0 and target < current_mns:
            return {
                "direction": "lower",
                "target": target,
                "current_ratio": current_ratio,
                "target_ratio": target / target_concurrency,
            }
    return None


def _prefill_scheduler_action_id(
    quantum_direction: str,
    admission_step: dict[str, Any] | None,
) -> str:
    if quantum_direction == "lower":
        return "lower_prefill_quantum_with_chunked_prefill"
    if quantum_direction == "raise":
        return "raise_prefill_quantum_with_chunked_prefill"
    if quantum_direction == "seed":
        return "seed_chunked_prefill_quantum"
    if admission_step is not None and admission_step["direction"] == "lower":
        return "lower_admission_pressure_with_chunked_prefill"
    if admission_step is not None and admission_step["direction"] == "raise":
        return "raise_admission_pressure_with_chunked_prefill"
    return "enable_chunked_prefill_scheduler_mode"


def _runtime_action(
    *,
    action_id: str,
    knob_family: str,
    score: float,
    score_factors: dict[str, Any] | None = None,
    patch: dict[str, Any],
    hypothesis: str,
    expected_effects: list[str],
) -> dict[str, Any]:
    return {
        "action_id": action_id,
        "knob_family": knob_family,
        "score": round(score, 4),
        "score_factors": score_factors
        or {
            "expected_bottleneck_relief": round(max(score - 0.1, 0.0), 4),
            "information_gain": 0.1,
            "launch_safety": 0.05,
            "regression_risk": 0.05,
        },
        "config_patch": {"env_patch": {}, "flag_patch": patch},
        "hypothesis": hypothesis,
        "expected_effects": expected_effects,
    }


def _annotate_candidate_signatures(
    study: StudySpec,
    candidates: list[dict[str, Any]],
) -> None:
    for candidate in candidates:
        if not isinstance(candidate, dict):
            continue
        patch = candidate.get("config_patch")
        if not isinstance(patch, dict):
            patch = {"env_patch": {}, "flag_patch": {}}
        signature = _effective_config_signature(study, patch)
        fingerprint = _effective_config_fingerprint(signature)
        candidate["status"] = "eligible"
        candidate["candidate_id"] = _candidate_id(candidate.get("action_id"), fingerprint)
        candidate["effective_config_fingerprint"] = fingerprint


def _blocked_candidate(
    *,
    action_id: str,
    knob_family: str,
    config_patch: dict[str, Any],
    blocked_reason: str,
    effective_config_signature: str,
    score: float | None = None,
    score_factors: dict[str, Any] | None = None,
) -> dict[str, Any]:
    fingerprint = _effective_config_fingerprint(effective_config_signature)
    item: dict[str, Any] = {
        "candidate_id": _candidate_id(action_id, fingerprint),
        "action_id": action_id,
        "knob_family": knob_family,
        "status": "blocked",
        "blocked_reason": blocked_reason,
        "blocked_reasons": [{"code": blocked_reason}],
        "config_patch": _normalized_config_patch(config_patch),
        "effective_config_fingerprint": fingerprint,
    }
    if score is not None:
        item["score"] = score
    if score_factors is not None:
        item["score_factors"] = score_factors
    return item


def _candidate_id(action_id: Any, fingerprint: str) -> str:
    action = str(action_id or "candidate").strip() or "candidate"
    return f"{action}:{fingerprint[:12]}"


def _effective_config_fingerprint(effective_config_signature: str) -> str:
    return hashlib.sha256(effective_config_signature.encode("utf-8")).hexdigest()


def _blocked_reason_summary(blocked_candidates: list[dict[str, Any]]) -> dict[str, int]:
    summary: dict[str, int] = {}
    for item in blocked_candidates:
        reason = str(item.get("blocked_reason") or "unknown")
        summary[reason] = summary.get(reason, 0) + 1
    return summary


def _candidate_set_hash(
    eligible_candidates: list[dict[str, Any]],
    blocked_candidates: list[dict[str, Any]],
) -> str:
    records: list[dict[str, Any]] = []
    for status, candidates in (
        ("eligible", eligible_candidates),
        ("blocked", blocked_candidates),
    ):
        for item in candidates:
            records.append(
                {
                    "status": status,
                    "candidate_id": item.get("candidate_id"),
                    "action_id": item.get("action_id"),
                    "knob_family": item.get("knob_family"),
                    "effective_config_fingerprint": item.get("effective_config_fingerprint"),
                    "blocked_reason": item.get("blocked_reason"),
                    "score": item.get("score"),
                }
            )
    payload = json.dumps(
        records,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _anchor_has_topology_patch(anchor: dict[str, Any]) -> bool:
    patch = anchor.get("config_patch")
    if not isinstance(patch, dict):
        return False
    flag_patch = patch.get("flag_patch")
    if not isinstance(flag_patch, dict):
        return False
    return any(key in flag_patch for key in _TOPOLOGY_KEYS)


def _has_unmeasured_higher_tp_candidate(
    study: StudySpec,
    legal: list[dict[str, Any]],
    *,
    current_tp: int,
    current_dp: int,
    tested_signatures: set[str],
) -> bool:
    for point in legal:
        if (
            point["data-parallel-size"] != current_dp
            or point["tensor-parallel-size"] <= current_tp
        ):
            continue
        signature = _effective_config_signature(
            study,
            {"env_patch": {}, "flag_patch": _topology_patch(study, point)},
        )
        if signature not in tested_signatures:
            return True
    return False


def _adjacent_lower_tp_candidate(
    legal: list[dict[str, Any]],
    *,
    current_tp: int,
    current_dp: int,
) -> int | None:
    lower_tps = {
        int(point["tensor-parallel-size"])
        for point in legal
        if point["data-parallel-size"] == current_dp
        and point["tensor-parallel-size"] < current_tp
    }
    return max(lower_tps) if lower_tps else None


def _legal_topology_points(
    study: StudySpec,
    *,
    current_tp: int,
    current_dp: int,
    current_ep: int,
    current_enable_ep: bool,
) -> list[dict[str, Any]]:
    constraints = study.engine.topology_constraints
    tunable = set(study.engine.tunable_flags)
    effective_gpu_count = _effective_gpu_count(study)
    if constraints is not None and constraints.allowed_tensor_parallel_sizes:
        tp_values = sorted(set(constraints.allowed_tensor_parallel_sizes))
    elif "tensor-parallel-size" in tunable:
        tp_values = [value for value in [1, 2, 4, 8] if value <= effective_gpu_count]
    else:
        tp_values = [current_tp]

    if constraints is not None and constraints.allowed_data_parallel_sizes:
        dp_values = sorted(set(constraints.allowed_data_parallel_sizes))
    elif "data-parallel-size" in tunable:
        dp_values = [value for value in [1, 2, 4, 8] if value <= effective_gpu_count]
    else:
        dp_values = [current_dp]

    if (
        study.trace.request_mode == "decode_only"
        and current_enable_ep
        and current_ep > 1
    ):
        ep_values = [current_ep]
    elif not current_enable_ep and current_ep <= 1:
        # Generic topology search should not introduce EP. EP is a separate
        # intervention family and needs direct EP-specific evidence; otherwise
        # TTFT/TPOT topology probes can waste trials on launch-risky EP variants.
        ep_values = [1]
    elif constraints is not None and constraints.allowed_expert_parallel_sizes:
        ep_values = sorted(set(constraints.allowed_expert_parallel_sizes))
    elif "expert-parallel-size" in tunable:
        ep_values = sorted({1, current_ep})
    else:
        ep_values = [current_ep]

    points: list[dict[str, Any]] = []
    for tp in tp_values:
        for dp in dp_values:
            tp_dp_product = tp * dp
            if constraints is not None:
                if (
                    constraints.allowed_tp_dp_products
                    and tp_dp_product not in constraints.allowed_tp_dp_products
                ):
                    continue
                if (
                    constraints.require_tp_dp_product_equals_gpu_count
                    and tp_dp_product != effective_gpu_count
                ):
                    continue
            elif tp_dp_product > effective_gpu_count:
                continue
            if constraints is not None and not constraints.allowed_tp_dp_products:
                if tp_dp_product > effective_gpu_count:
                    continue
            if tp_dp_product > effective_gpu_count:
                continue
            for ep in ep_values:
                enable_ep = current_enable_ep or ep > 1
                if constraints is not None:
                    if constraints.allowed_expert_parallel_sizes and ep not in constraints.allowed_expert_parallel_sizes:
                        continue
                    if constraints.require_ep_size_leq_tp_dp_product and ep > tp_dp_product:
                        continue
                    if constraints.require_ep_size_divides_tp_dp_product and tp_dp_product % ep != 0:
                        continue
                    if (
                        constraints.require_enable_expert_parallel_when_ep_gt_one
                        and ep > 1
                        and not enable_ep
                    ):
                        continue
                points.append(
                    {
                        "tensor-parallel-size": tp,
                        "data-parallel-size": dp,
                        "expert-parallel-size": ep,
                        "enable-expert-parallel": enable_ep,
                    }
                )
    return points


def _topology_patch(study: StudySpec, point: dict[str, Any]) -> dict[str, Any]:
    patch: dict[str, Any] = {}
    tunable = set(study.engine.tunable_flags)
    base = _normalized_topology_flags(study.engine.base_flags)
    for key in (
        "tensor-parallel-size",
        "data-parallel-size",
        "expert-parallel-size",
        "enable-expert-parallel",
    ):
        if key not in tunable:
            continue
        if key in point and point[key] != base.get(key):
            patch[key] = point[key]
    return patch


def _preserve_topology_patch(study: StudySpec, flags: dict[str, Any]) -> dict[str, Any]:
    patch: dict[str, Any] = {}
    tunable = set(study.engine.tunable_flags)
    base = _normalized_topology_flags(study.engine.base_flags)
    normalized = _normalized_topology_flags(flags)
    for key in (
        "tensor-parallel-size",
        "data-parallel-size",
        "expert-parallel-size",
        "enable-expert-parallel",
    ):
        if key not in tunable or key not in normalized:
            continue
        if normalized.get(key) != base.get(key):
            patch[key] = normalized[key]
    return patch


def _explicit_topology_patch(study: StudySpec, flags: dict[str, Any]) -> dict[str, Any]:
    patch: dict[str, Any] = {}
    tunable = set(study.engine.tunable_flags)
    normalized = _normalized_topology_flags(flags)
    for key in (
        "tensor-parallel-size",
        "data-parallel-size",
        "expert-parallel-size",
        "enable-expert-parallel",
    ):
        if key not in tunable:
            continue
        if key not in flags and key not in study.engine.base_flags:
            continue
        patch[key] = normalized[key]
    return patch


def _preserve_runtime_patch(study: StudySpec, flags: dict[str, Any]) -> dict[str, Any]:
    patch: dict[str, Any] = {}
    tunable = set(study.engine.tunable_flags)
    base = study.engine.base_flags
    for key in _RUNTIME_KEYS:
        if key not in tunable or key not in flags:
            continue
        if flags.get(key) != base.get(key):
            patch[key] = flags[key]
    return patch


def _normalized_topology_flags(flags: dict[str, Any]) -> dict[str, Any]:
    return {
        "tensor-parallel-size": _parse_int_like(
            flags.get("tensor-parallel-size"),
            default=1,
        ),
        "data-parallel-size": _parse_int_like(
            flags.get("data-parallel-size"),
            default=1,
        ),
        "expert-parallel-size": _parse_int_like(
            flags.get("expert-parallel-size"),
            default=1,
        ),
        "enable-expert-parallel": bool(flags.get("enable-expert-parallel", False)),
    }


def _topology_key(flags: dict[str, Any]) -> tuple[int, int, int, bool]:
    normalized = _normalized_topology_flags(flags)
    return (
        int(normalized["tensor-parallel-size"]),
        int(normalized["data-parallel-size"]),
        int(normalized["expert-parallel-size"]),
        bool(normalized["enable-expert-parallel"]),
    )


def _topology_parallel_size(flags: dict[str, Any]) -> int:
    normalized = _normalized_topology_flags(flags)
    return max(
        1,
        int(normalized["tensor-parallel-size"])
        * int(normalized["data-parallel-size"]),
    )


def _effective_gpu_count(study: StudySpec) -> int:
    visible = str(study.engine.base_envs.get("CUDA_VISIBLE_DEVICES") or "").strip()
    if not visible:
        return study.hardware.gpu_count
    if visible.lower() in {"none", "void", "-1"}:
        return 0
    devices = [item.strip() for item in visible.split(",") if item.strip()]
    if not devices:
        return study.hardware.gpu_count
    return min(study.hardware.gpu_count, len(devices))


def _parallel_size_can_vary(study: StudySpec) -> bool:
    tunable = set(study.engine.tunable_flags)
    if not ({"tensor-parallel-size", "data-parallel-size"} & tunable):
        return False
    effective_gpu_count = _effective_gpu_count(study)
    if effective_gpu_count <= 1:
        return False
    base = _normalized_topology_flags(study.engine.base_flags)
    legal = _legal_topology_points(
        study,
        current_tp=int(base["tensor-parallel-size"]),
        current_dp=int(base["data-parallel-size"]),
        current_ep=int(base["expert-parallel-size"]),
        current_enable_ep=bool(base["enable-expert-parallel"]),
    )
    signatures: set[str] = set()
    for point in legal:
        patch = _topology_patch(study, point)
        flags = {**study.engine.base_flags, **patch}
        normalized = _normalized_topology_flags(flags)
        if any(normalized.get(key) != point.get(key) for key in point):
            continue
        signatures.add(_effective_config_signature(study, {"env_patch": {}, "flag_patch": patch}))
    return len(signatures) > 1


def _score_topology_candidate(
    top_bottleneck: str,
    bottleneck_hypotheses: list[dict[str, Any]],
    *,
    request_mode: str,
    current_tp: int,
    current_dp: int,
    candidate_tp: int,
    candidate_dp: int,
) -> tuple[float, dict[str, float]]:
    tp_delta = candidate_tp - current_tp
    dp_delta = candidate_dp - current_dp
    confidence = _hypothesis_confidence(bottleneck_hypotheses, top_bottleneck)
    relief = 0.0
    if top_bottleneck == "ttft_prefill":
        relief = 0.42 if tp_delta > 0 else 0.05
    elif top_bottleneck == "decode_tpot" and request_mode == "decode_only":
        if dp_delta > 0 and candidate_tp <= current_tp:
            relief = 0.44
        elif dp_delta > 0:
            relief = 0.24
        elif tp_delta > 0 and candidate_dp < current_dp:
            relief = 0.03
        else:
            relief = 0.08
    elif top_bottleneck == "decode_tpot":
        relief = 0.52 if tp_delta > 0 else 0.02
    elif top_bottleneck == "admission_or_queueing":
        relief = 0.34 if dp_delta > 0 else 0.08
    else:
        relief = 0.04
    info_gain = 0.2 if abs(tp_delta) + abs(dp_delta) > 0 else 0.0
    launch_safety = 0.08 if candidate_tp * candidate_dp <= max(current_tp * current_dp, 1) else 0.04
    distance = abs(_log2_ratio(candidate_tp, current_tp)) + abs(_log2_ratio(candidate_dp, current_dp))
    regression_risk = min(0.28, 0.06 * distance)
    score = relief * max(confidence, 0.35) + info_gain + launch_safety - regression_risk
    return score, {
        "expected_bottleneck_relief": round(relief, 4),
        "bottleneck_confidence": round(confidence, 4),
        "information_gain": round(info_gain, 4),
        "launch_safety": round(launch_safety, 4),
        "regression_risk": round(regression_risk, 4),
    }


def _information_gain(bottleneck_hypotheses: list[dict[str, Any]], family: str) -> float:
    if not bottleneck_hypotheses:
        return 0.08
    top_confidence = _as_float(bottleneck_hypotheses[0].get("confidence"))
    uncertainty = max(0.0, 1.0 - top_confidence)
    return 0.08 + min(0.12, uncertainty * 0.12)


def _hypothesis_confidence(
    bottleneck_hypotheses: list[dict[str, Any]],
    name: str,
) -> float:
    for item in bottleneck_hypotheses:
        if item.get("name") == name:
            return _as_float(item.get("confidence"))
    return 0.0


def _topology_action_id(
    current_tp: int,
    current_dp: int,
    point: dict[str, Any],
) -> str:
    candidate_tp = int(point["tensor-parallel-size"])
    candidate_dp = int(point["data-parallel-size"])
    if candidate_tp > current_tp:
        return "topology_frontier_probe_for_slo_pressure"
    if candidate_dp > current_dp:
        return "increase_data_parallel_probe"
    if candidate_tp < current_tp:
        return "decrease_tensor_parallel_probe"
    return "redistribute_topology_probe"


def _topology_hypothesis(
    top_bottleneck: str,
    *,
    current_tp: int,
    current_dp: int,
    candidate_tp: int,
    candidate_dp: int,
) -> str:
    return (
        f"Ranked bottleneck is {top_bottleneck}. Test topology "
        f"TP={candidate_tp}, DP={candidate_dp} against incumbent TP={current_tp}, "
        f"DP={current_dp}; this distinguishes compute-latency relief from "
        "replica/admission effects under the configured SLO."
    )


def _log2_ratio(new: int, old: int) -> float:
    if new <= 0 or old <= 0:
        return 0.0
    ratio = new / old
    steps = 0.0
    while ratio >= 2.0:
        steps += 1.0
        ratio /= 2.0
    while ratio <= 0.5:
        steps += 1.0
        ratio *= 2.0
    return steps


def _topology_frontier_status(
    study: StudySpec,
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
    window_summary: dict[str, Any] | None = None,
) -> dict[str, Any]:
    default = {
        "frontier_open": False,
        "reason": "topology_frontier_not_needed",
        "current_trial_id": state.best_trial_id,
        "active_bottleneck": None,
        "current_tp": None,
        "current_dp": None,
        "next_tp": None,
        "flag_patch": None,
    }
    if "tensor-parallel-size" not in set(study.engine.tunable_flags):
        return {**default, "reason": "tensor_parallel_size_not_tunable"}
    if not state.best_trial_id:
        return {**default, "reason": "no_incumbent_for_topology_frontier"}

    best = next(
        (item for item in recent_diagnostics if item.get("trial_id") == state.best_trial_id),
        None,
    )
    if best is None:
        return {**default, "reason": "incumbent_not_in_recent_harness_history"}
    if best.get("status") != "completed":
        return {**default, "reason": "incumbent_not_completed"}
    if not isinstance(best.get("best_request_rate_per_gpu"), (int, float)):
        return {**default, "reason": "incumbent_has_no_feasible_rate"}

    active_bottleneck = str(best.get("active_bottleneck") or "")
    if active_bottleneck in {"unknown", "none_obvious", ""} and recent_diagnostics:
        active_bottleneck = str(recent_diagnostics[-1].get("active_bottleneck") or "")
    if active_bottleneck not in {"ttft_prefill", "decode_tpot"}:
        return {
            **default,
            "reason": "active_bottleneck_does_not_require_tp_frontier",
            "active_bottleneck": active_bottleneck,
        }
    if active_bottleneck == "decode_tpot" and study.trace.request_mode == "decode_only":
        return {
            **default,
            "reason": "decode_tpot_uses_topology_redistribution_not_higher_tp_frontier",
            "active_bottleneck": active_bottleneck,
        }

    flags = _effective_flags_for_item(study, best)
    current_tp = _parse_int_like(flags.get("tensor-parallel-size"), default=1)
    current_dp = _parse_int_like(flags.get("data-parallel-size"), default=1)
    flag_patch = _higher_tp_frontier_patch(
        study,
        current_tp=current_tp,
        current_dp=current_dp,
    )
    if flag_patch is None:
        return {
            **default,
            "reason": "no_legal_higher_tp_frontier",
            "active_bottleneck": active_bottleneck,
            "current_tp": current_tp,
            "current_dp": current_dp,
        }
    next_tp = _parse_int_like(flag_patch.get("tensor-parallel-size"), default=current_tp)

    signature = _effective_config_signature(study, {"env_patch": {}, "flag_patch": flag_patch})
    if signature in _state_tested_signatures(study, state):
        return {
            **default,
            "reason": "higher_tp_frontier_already_tested",
            "active_bottleneck": active_bottleneck,
            "current_tp": current_tp,
            "current_dp": current_dp,
            "next_tp": next_tp,
            "flag_patch": flag_patch,
        }
    return {
        "frontier_open": True,
        "reason": "higher_tp_frontier_unmeasured_under_slo_pressure",
        "current_trial_id": state.best_trial_id,
        "active_bottleneck": active_bottleneck,
        "current_tp": current_tp,
        "current_dp": current_dp,
        "next_tp": next_tp,
        "flag_patch": flag_patch,
    }


def _higher_tp_frontier_patch(
    study: StudySpec,
    *,
    current_tp: int,
    current_dp: int,
) -> dict[str, Any] | None:
    next_tp = _next_allowed_tp(study, current_tp=current_tp, current_dp=current_dp)
    if next_tp is None:
        return None
    flag_patch: dict[str, Any] = {"tensor-parallel-size": next_tp}
    base_dp = _parse_int_like(study.engine.base_flags.get("data-parallel-size"), default=1)
    if current_dp != base_dp:
        flag_patch["data-parallel-size"] = current_dp
    return flag_patch


def _effective_flags_for_item(study: StudySpec, item: dict[str, Any]) -> dict[str, Any]:
    flags = dict(study.engine.base_flags)
    patch = item.get("config_patch")
    if isinstance(patch, dict) and isinstance(patch.get("flag_patch"), dict):
        flags.update(patch["flag_patch"])
    return flags


def _state_tested_signatures(study: StudySpec, state: StudyState) -> set[str]:
    return {
        _effective_config_signature(study, trial.config_patch)
        for trial in state.trials
    }


def _runtime_refinement_proposal(
    study: StudySpec,
    window_summary: dict[str, Any],
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
    *,
    tested_signatures: set[str],
) -> dict[str, Any]:
    default = {
        "should_propose": False,
        "reason": "runtime_refinement_not_applicable",
        "diagnosis": "Runtime refinement does not apply yet.",
        "config_patch": {"env_patch": {}, "flag_patch": {}},
        "expected_effects": [],
    }
    if not state.best_trial_id or not recent_diagnostics:
        return default
    best = next(
        (item for item in recent_diagnostics if item.get("trial_id") == state.best_trial_id),
        None,
    )
    if not best or best.get("status") != "completed":
        return default
    if recent_diagnostics[-1].get("trial_id") != state.best_trial_id:
        return default
    best_patch = best.get("config_patch")
    if not isinstance(best_patch, dict):
        return default
    best_flags = best_patch.get("flag_patch")
    if not isinstance(best_flags, dict):
        best_flags = {}
    best_effective_flags = _effective_flags_for_item(study, best)
    best_tp = _parse_int_like(best_effective_flags.get("tensor-parallel-size"), default=1)
    if best_tp <= 1:
        return default
    tunable = set(study.engine.tunable_flags)
    flag_patch = {
        **_preserve_topology_patch(study, best_effective_flags),
        **_preserve_runtime_patch(study, best_effective_flags),
    }
    if "enable-chunked-prefill" in tunable:
        flag_patch["enable-chunked-prefill"] = True
    if "max-num-batched-tokens" not in tunable:
        return default

    current_mbt = _parse_int_like(best_flags.get("max-num-batched-tokens"), default=0)
    if current_mbt <= 0:
        target_mbt = _initial_mbt_from_window(window_summary)
        reason = "same_topology_runtime_seed_after_tp_incumbent"
    else:
        target_mbt = _next_mbt_step(current_mbt)
        reason = "same_topology_mbt_growth_after_feasible_runtime_incumbent"
    if target_mbt is None:
        return {
            **default,
            "reason": "no_larger_mbt_step_available",
        }
    flag_patch["max-num-batched-tokens"] = target_mbt
    signature = _effective_config_signature(study, {"env_patch": {}, "flag_patch": flag_patch})
    if signature in tested_signatures:
        return {
            **default,
            "reason": "same_topology_runtime_probe_already_tested",
        }
    return {
        "should_propose": True,
        "reason": reason,
        "diagnosis": (
            "A TP incumbent improved request_rate_per_gpu; refine batching on the "
            "same topology before trying DP/EP or broad runtime changes."
        ),
        "config_patch": {"env_patch": {}, "flag_patch": flag_patch},
        "expected_effects": [
            "preserve the incumbent topology",
            "increase batching headroom without dropping measured runtime gains",
        ],
        "incumbent_trial_id": best.get("trial_id"),
    }


def _initial_mbt_from_window(window_summary: dict[str, Any]) -> int:
    prompt_p99 = _as_float(window_summary.get("prompt_tokens_p99"))
    target = max(8192, int(prompt_p99 * 2.0))
    return min(32768, _round_up_to_multiple(target, 1024))


def _next_mbt_step(current_mbt: int) -> int | None:
    if current_mbt >= 32768:
        return None
    return min(32768, _round_up_to_multiple(int(current_mbt * 1.5), 1024))


def _round_up_to_multiple(value: int, multiple: int) -> int:
    return ((max(value, 1) + multiple - 1) // multiple) * multiple


def _next_allowed_tp(study: StudySpec, *, current_tp: int, current_dp: int) -> int | None:
    constraints = study.engine.topology_constraints
    effective_gpu_count = _effective_gpu_count(study)
    if constraints is not None and constraints.allowed_tensor_parallel_sizes:
        candidates = sorted({int(item) for item in constraints.allowed_tensor_parallel_sizes})
    else:
        candidates = [1, 2, 4, 8]
    for candidate in candidates:
        if candidate <= current_tp:
            continue
        tp_dp_product = candidate * current_dp
        if tp_dp_product > effective_gpu_count:
            continue
        if constraints is not None:
            if (
                constraints.allowed_tp_dp_products
                and tp_dp_product not in constraints.allowed_tp_dp_products
            ):
                continue
            if (
                constraints.require_tp_dp_product_equals_gpu_count
                and tp_dp_product != effective_gpu_count
            ):
                continue
        return candidate
    return None


def _search_high_saturation_guard(
    study: StudySpec,
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
) -> dict[str, Any]:
    default = {
        "saturated": False,
        "reason": "search_high_not_saturated",
        "summary": "The incumbent has not saturated the configured search high.",
        "incumbent_trial_id": state.best_trial_id,
        "search_high": study.search.high,
        "last_threshold": None,
        "threshold_gap_to_high": None,
        "measurement_evidence": None,
    }
    if not state.best_trial_id:
        return default
    incumbent = next(
        (
            item
            for item in recent_diagnostics
            if item.get("trial_id") == state.best_trial_id
        ),
        None,
    )
    if not incumbent:
        return {
            **default,
            "reason": "incumbent_not_in_recent_harness_history",
        }
    probe_summary = incumbent.get("probe_summary")
    if not isinstance(probe_summary, dict):
        return {
            **default,
            "reason": "incumbent_probe_summary_missing",
        }
    last_probe = probe_summary.get("last_probe")
    if not isinstance(last_probe, dict):
        return {
            **default,
            "reason": "incumbent_last_probe_missing",
        }
    measurement_evidence = (
        incumbent.get("measurement_evidence")
        if isinstance(incumbent.get("measurement_evidence"), dict)
        else None
    )
    search_high = (
        _as_float(measurement_evidence.get("search_high"))
        if isinstance(measurement_evidence, dict)
        and isinstance(measurement_evidence.get("search_high"), (int, float))
        else float(study.search.high)
    )
    last_threshold = _as_float(last_probe.get("threshold"))
    threshold_gap = search_high - last_threshold
    binary_probe_resolution = max(
        float(study.search.tolerance),
        (search_high - float(study.search.low))
        / float(2 ** max(study.search.max_probes, 1)),
    )
    if not last_probe.get("feasible"):
        return {
            **default,
            "reason": "incumbent_last_probe_not_feasible",
            "last_threshold": last_threshold,
            "threshold_gap_to_high": threshold_gap,
        }
    if threshold_gap > binary_probe_resolution + 1e-12:
        return {
            **default,
            "reason": "incumbent_not_close_to_search_high",
            "last_threshold": last_threshold,
            "threshold_gap_to_high": threshold_gap,
            "binary_probe_resolution": binary_probe_resolution,
        }
    reason = "search_high_saturated_by_incumbent"
    summary = (
        "The incumbent's highest measured probe is feasible and is within "
        "the configured binary-search resolution of search.high."
    )
    if (
        isinstance(measurement_evidence, dict)
        and measurement_evidence.get("measurement_ceiling_insufficient")
    ):
        reason = "measurement_ceiling_insufficient"
        summary = (
            "The incumbent saturated the available trace measurement ceiling; "
            "this is insufficient measurement evidence, not stop authorization."
        )
    return {
        "saturated": True,
        "reason": reason,
        "summary": summary,
        "incumbent_trial_id": state.best_trial_id,
        "search_high": search_high,
        "last_threshold": last_threshold,
        "threshold_gap_to_high": threshold_gap,
        "binary_probe_resolution": binary_probe_resolution,
        "measurement_evidence": measurement_evidence,
    }


def _validation_exhausted_guard(
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
) -> dict[str, Any]:
    default = {
        "exhausted": False,
        "reason": "validation_not_exhausted",
        "summary": "Validation probes are not sufficient to stop yet.",
        "incumbent_trial_id": state.best_trial_id,
        "incumbent_gain_vs_baseline": None,
        "validation_trial_ids": [],
        "validation_families": [],
    }
    if not state.best_trial_id or not isinstance(state.best_request_rate_per_gpu, (int, float)):
        return default
    state_completed = _state_completed_trials_with_rates(state)
    if state_completed:
        baseline_rate = float(state_completed[0].best_request_rate_per_gpu)
    else:
        completed = [
            item
            for item in recent_diagnostics
            if item.get("status") == "completed"
            and isinstance(item.get("best_request_rate_per_gpu"), (int, float))
        ]
        if not completed:
            return default
        baseline_rate = _as_float(completed[0].get("best_request_rate_per_gpu"))
    incumbent_rate = _as_float(state.best_request_rate_per_gpu)
    if baseline_rate <= 0 or incumbent_rate <= 0:
        return default
    gain = incumbent_rate / baseline_rate

    best_index = next(
        (
            index
            for index, item in enumerate(recent_diagnostics)
            if item.get("trial_id") == state.best_trial_id
        ),
        None,
    )
    if best_index is None:
        return {
            **default,
            "reason": "incumbent_not_in_recent_harness_history",
            "incumbent_gain_vs_baseline": gain,
        }
    after_best = [
        item
        for item in recent_diagnostics[best_index + 1 :]
        if item.get("status") in {"completed", "failed"}
    ]
    incumbent = next(
        (
            item
            for item in recent_diagnostics
            if item.get("trial_id") == state.best_trial_id
        ),
        {},
    )
    gmu_ceiling_incumbent = _is_gpu_memory_utilization_ceiling_incumbent(incumbent)
    if gain < _STRONG_INCUMBENT_MIN_GAIN and not gmu_ceiling_incumbent:
        return {
            **default,
            "reason": "incumbent_gain_not_large_enough_for_validation_stop",
            "incumbent_gain_vs_baseline": gain,
        }
    if len(after_best) < _MIN_POST_INCUMBENT_VALIDATION_TRIALS:
        return {
            **default,
            "reason": "need_at_least_two_post_incumbent_validation_trials",
            "incumbent_gain_vs_baseline": gain,
            "validation_trial_ids": [str(item.get("trial_id")) for item in after_best],
        }
    improving_trials = [
        item
        for item in after_best
        if isinstance(item.get("best_request_rate_per_gpu"), (int, float))
        and float(item["best_request_rate_per_gpu"]) > incumbent_rate
    ]
    if improving_trials:
        return {
            **default,
            "reason": "post_incumbent_validation_found_improving_candidate",
            "incumbent_gain_vs_baseline": gain,
            "validation_trial_ids": [str(item.get("trial_id")) for item in improving_trials],
        }

    families: set[str] = set()
    for item in after_best:
        families.update(_validation_families(item))
    families.update(_validation_families(incumbent))
    has_topology = "topology" in families
    has_runtime = bool(families & {"runtime", "max-num-seqs", "max-num-batched-tokens"})
    enough_evidence = (
        len(after_best) >= _VALIDATION_TRIALS_WITHOUT_FAMILY_COVERAGE
        or (has_topology and has_runtime)
    )
    if not enough_evidence:
        return {
            **default,
            "reason": "post_incumbent_validation_has_not_covered_enough_families",
            "incumbent_gain_vs_baseline": gain,
            "validation_trial_ids": [str(item.get("trial_id")) for item in after_best],
            "validation_families": sorted(families),
        }

    return {
        "exhausted": True,
        "reason": "post_incumbent_validation_exhausted",
        "summary": (
            "A strong incumbent was followed by validation probes across nearby "
            "topology/runtime families, and none improved request_rate_per_gpu."
        ),
        "incumbent_trial_id": state.best_trial_id,
        "incumbent_gain_vs_baseline": gain,
        "validation_trial_ids": [str(item.get("trial_id")) for item in after_best],
        "validation_families": sorted(families),
    }


def _validation_families(item: dict[str, Any]) -> set[str]:
    config_patch = item.get("config_patch")
    if not isinstance(config_patch, dict):
        return set()
    flag_patch = config_patch.get("flag_patch")
    if not isinstance(flag_patch, dict):
        return set()
    families: set[str] = set()
    if any(key in flag_patch for key in _TOPOLOGY_KEYS):
        families.add("topology")
    for key in _RUNTIME_KEYS:
        if key in flag_patch:
            families.add("runtime")
            families.add(key)
    if not families and flag_patch:
        families.add("other")
    return families


def _is_gpu_memory_utilization_ceiling_incumbent(item: dict[str, Any]) -> bool:
    config_patch = item.get("config_patch")
    if not isinstance(config_patch, dict):
        return False
    flag_patch = config_patch.get("flag_patch")
    if not isinstance(flag_patch, dict):
        return False
    gmu = _parse_float_like(flag_patch.get("gpu-memory-utilization"), default=0.0)
    return gmu >= _GMU_SAFE_CEILING - EPSILON


def _strong_incumbent_guard(
    state: StudyState,
    recent_diagnostics: list[dict[str, Any]],
) -> dict[str, Any]:
    default = {
        "guard_active": False,
        "reason": "no_strong_incumbent_yet",
        "baseline_trial_id": None,
        "baseline_request_rate_per_gpu": None,
        "incumbent_gain_vs_baseline": None,
    }
    if state.best_trial_id is None or state.best_request_rate_per_gpu is None:
        return default
    completed = _state_completed_trials_with_rates(state)
    if len(completed) < 2:
        return default
    baseline = completed[0]
    baseline_rate = float(baseline.best_request_rate_per_gpu)
    incumbent_rate = float(state.best_request_rate_per_gpu)
    if baseline_rate <= 0:
        return default
    gain = incumbent_rate / baseline_rate
    latest = recent_diagnostics[-1] if recent_diagnostics else {}
    if state.best_trial_id == latest.get("trial_id") and gain >= 1.8:
        return {
            "guard_active": True,
            "reason": "incumbent_exceeds_baseline_by_1_8x_and_latest_trial_is_best_enter_validation_phase",
            "baseline_trial_id": baseline.trial_id,
            "baseline_request_rate_per_gpu": baseline_rate,
            "incumbent_gain_vs_baseline": gain,
            "recommended_next_action": (
                "validate_the_incumbent_with_adjacent_topology_or_same_topology_runtime_probes; "
                "do_not_claim_global_optimum_from_the_first_large_gain"
            ),
        }
    return {
        **default,
        "baseline_trial_id": baseline.trial_id,
        "baseline_request_rate_per_gpu": baseline_rate,
        "incumbent_gain_vs_baseline": gain,
        "reason": "need_more_evidence_before_strong_incumbent_stop",
        "recommended_next_action": "continue_harness_guided_search",
    }


def _infeasible_progress_guard(recent_diagnostics: list[dict[str, Any]]) -> dict[str, Any]:
    points = [
        point
        for item in recent_diagnostics
        if (point := _infeasible_progress_point(item)) is not None
    ]
    default = {
        "plateau_detected": False,
        "stop_if_next_probe_repeats_family": False,
        "reason": "insufficient_infeasible_history",
        "blocked_primary_family": None,
        "recommended_next_action": "continue_adjacent_probe_if_a_harness_justifies_it",
        "comparison": None,
    }
    if len(points) < 2:
        return default

    prev, latest = points[-2], points[-1]
    if not _same_threshold(prev["threshold"], latest["threshold"]):
        return {
            **default,
            "reason": "recent_infeasible_trials_used_different_sampling_thresholds",
        }
    family = _changed_primary_family(prev["config_patch"], latest["config_patch"])
    pass_rate_delta = latest["pass_rate"] - prev["pass_rate"]
    p95_delta_ratio = _relative_delta(latest["ttft_p95_ms"], prev["ttft_p95_ms"])
    p99_delta_ratio = _relative_delta(latest["ttft_p99_ms"], prev["ttft_p99_ms"])
    plateau = (
        family is not None
        and abs(pass_rate_delta) <= 0.01
        and p95_delta_ratio is not None
        and p95_delta_ratio >= -0.02
    )
    comparison = {
        "previous": prev,
        "latest": latest,
        "changed_primary_family": family,
        "pass_rate_delta": pass_rate_delta,
        "ttft_p95_delta_ratio": p95_delta_ratio,
        "ttft_p99_delta_ratio": p99_delta_ratio,
    }
    if not plateau:
        return {
            **default,
            "reason": "recent_infeasible_trials_still_show_material_progress",
            "comparison": comparison,
        }
    return {
        "plateau_detected": True,
        "stop_if_next_probe_repeats_family": True,
        "reason": f"{family}_plateau_on_infeasible_trials",
        "blocked_primary_family": family,
        "recommended_next_action": (
            "switch_primary_family_with_a_specific_bottleneck_rationale_or_return_should_stop"
        ),
        "comparison": comparison,
    }


def _infeasible_progress_point(item: dict[str, Any]) -> dict[str, Any] | None:
    if item.get("status") != "completed" or item.get("best_request_rate") is not None:
        return None
    probe_summary = item.get("probe_summary")
    if not isinstance(probe_summary, dict):
        return None
    probe = probe_summary.get("all_infeasible") or probe_summary.get("last_probe")
    compact = _compact_probe(probe if isinstance(probe, dict) else None)
    if not compact or compact.get("feasible"):
        return None
    latency_summary = compact.get("latency_summary")
    if not isinstance(latency_summary, dict):
        latency_summary = {}
    ttft = latency_summary.get("ttft_ms")
    if not isinstance(ttft, dict):
        ttft = {}
    return {
        "trial_id": item.get("trial_id"),
        "threshold": _as_float(compact.get("threshold")),
        "request_rate": _as_float(compact.get("request_rate")),
        "pass_rate": _as_float(compact.get("pass_rate")),
        "ttft_p95_ms": _optional_float(ttft.get("p95")),
        "ttft_p99_ms": _optional_float(ttft.get("p99")),
        "active_bottleneck": item.get("active_bottleneck"),
        "config_patch": item.get("config_patch") if isinstance(item.get("config_patch"), dict) else {},
    }


def _changed_primary_family(
    previous_patch: dict[str, Any],
    latest_patch: dict[str, Any],
) -> str | None:
    previous_flags = previous_patch.get("flag_patch") if isinstance(previous_patch, dict) else {}
    latest_flags = latest_patch.get("flag_patch") if isinstance(latest_patch, dict) else {}
    if not isinstance(previous_flags, dict) or not isinstance(latest_flags, dict):
        return None
    changed: list[str] = []
    for key, family in (
        ("tensor-parallel-size", "tensor-parallel-size"),
        ("data-parallel-size", "data-parallel-size"),
        ("expert-parallel-size", "expert-parallel-size"),
        ("max-num-seqs", "max-num-seqs"),
        ("max-num-batched-tokens", "max-num-batched-tokens"),
        ("gpu-memory-utilization", "gpu-memory-utilization"),
        ("enable-chunked-prefill", "enable-chunked-prefill"),
    ):
        if previous_flags.get(key) != latest_flags.get(key):
            changed.append(family)
    if len(changed) == 1:
        return changed[0]
    if changed:
        return "mixed:" + ",".join(changed)
    return None


def _same_threshold(left: float, right: float) -> bool:
    return abs(left - right) <= max(abs(left), abs(right), 1.0) * 1e-6


def _relative_delta(new: float | None, old: float | None) -> float | None:
    if new is None or old is None or old <= 0:
        return None
    return (new - old) / old


def _proposal_rules() -> list[str]:
    return [
        "First decide the active bottleneck from recent_trial_diagnostics.",
        "For decode_only studies, ignore TTFT unless a TTFT SLO is explicitly configured; prioritize TPOT pass rate and request_rate_per_gpu.",
        "For decode_tpot bottlenecks, prefer legal TP/DP topology redistribution before runtime-only knobs, then tune max-num-seqs or max-num-batched-tokens only from observed SLO headroom/failures.",
        "Pick at most one primary knob family from knob_harnesses unless the history proves a coupled change is needed.",
        "Use adjacent legal values around the incumbent; avoid broad exploratory jumps.",
        "config_patch is applied to the study base config, not as a delta on the incumbent; when validating an incumbent topology, include all topology flags needed to reconstruct that incumbent plus the runtime knob under test.",
        "When strong_incumbent.guard_active is true, treat the run as a validation phase, not as proof of optimum: probe only adjacent topology or same-topology runtime knobs that can falsify the incumbent.",
        "Do not stop solely because a strong incumbent appeared; stop only after nearby validation probes are exhausted, blocked by infeasible evidence, or repeatedly remain near the incumbent.",
        "During validation, prefer probes that answer whether the incumbent is locally optimal over probes that merely chase another large gain.",
        "Do not enable expert parallel for TTFT-prefill bottlenecks unless expert-parallel is the active harness and there is direct positive EP evidence.",
        "If infeasible_progress blocks the last primary knob family, do not continue that family; switch families with direct bottleneck evidence or set should_stop=true.",
        "If a proposed config is likely to reduce request_rate_per_gpu under the active guard, set should_stop=true instead of exploring.",
        "Never repeat an already tested config signature.",
        "Return should_stop=true when the convergence guard fires and no active harness justifies another adjacent probe.",
    ]


def _workload_default_bottleneck(study: StudySpec, window_summary: dict[str, Any]) -> str:
    if study.trace.request_mode == "decode_only":
        if study.slo.tpot_rule is not None:
            return "decode_tpot"
        if study.slo.ttft_rule is not None:
            return "ttft_prefill"
        return "admission_or_queueing"
    tail_ratio = _as_float(window_summary.get("prompt_tail_ratio_p95_p50"))
    prompt_p95 = _as_float(window_summary.get("prompt_tokens_p95"))
    prefix_cache = window_summary.get("prefix_cache")
    cache_ratio = 0.0
    if isinstance(prefix_cache, dict):
        cache_ratio = _as_float(prefix_cache.get("repeated_token_ratio_estimate"))
    if prompt_p95 >= 4096 or tail_ratio >= 2.0:
        return "ttft_prefill"
    if cache_ratio >= 0.25:
        return "admission_or_queueing"
    return "unknown"


def _length_regime(prompt_p95: float, tail_ratio: float) -> str:
    if prompt_p95 >= 8192 or tail_ratio >= 4.0:
        return "heavy_tail_long_prefill"
    if prompt_p95 >= 4096 or tail_ratio >= 2.0:
        return "moderate_tail_prefill_sensitive"
    return "short_or_moderate"


def _cache_regime(repeated_token_ratio: float) -> str:
    if repeated_token_ratio >= 0.5:
        return "high_prefix_reuse"
    if repeated_token_ratio >= 0.2:
        return "moderate_prefix_reuse"
    return "low_prefix_reuse"


def _arrival_regime(burst_ratio: float, interarrival_cv: float) -> str:
    if burst_ratio >= 3.0 or interarrival_cv >= 2.0:
        return "bursty"
    if burst_ratio >= 1.5 or interarrival_cv >= 1.0:
        return "moderately_bursty"
    return "smooth"


def _as_float(value: Any) -> float:
    if isinstance(value, bool):
        return 0.0
    if isinstance(value, (int, float)):
        return float(value)
    return 0.0


def _optional_float(value: Any) -> float | None:
    if isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return float(value)
    return None


def _parse_int_like(value: Any, *, default: int) -> int:
    if value is None:
        return default
    if isinstance(value, bool):
        return default
    if isinstance(value, int):
        return value
    if isinstance(value, float) and value.is_integer():
        return int(value)
    if isinstance(value, str) and value.strip():
        try:
            return int(value.strip())
        except ValueError:
            return default
    return default


def _parse_float_like(value: Any, *, default: float) -> float:
    if value is None or isinstance(value, bool):
        return default
    if isinstance(value, (int, float)):
        return float(value)
    if isinstance(value, str) and value.strip():
        try:
            return float(value.strip())
        except ValueError:
            return default
    return default


def _config_signature(config_patch: Any) -> str:
    return json.dumps(
        _normalized_config_patch(config_patch),
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )


def _effective_config_signature(study: StudySpec, config_patch: Any) -> str:
    return _shared_effective_config_signature(study, config_patch)


def _normalized_config_patch(config_patch: Any) -> dict[str, dict[str, Any]]:
    return _shared_normalized_config_patch(config_patch)
