from __future__ import annotations

import json
import math
import os
import signal
import statistics
import subprocess
import threading
import time
import traceback
from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

from .engine import build_launch_recipe
from .http_client import HttpClientError, stream_chat_completion, wait_for_server
from .lca import find_convergence_prefix, resolve_length_mode
from .search import ThresholdProbe, binary_search_max_feasible
from .slo import RequestOutcome, _rule_threshold_ms, evaluate_request, summarize_evaluations
from .spec import ConfigPatch, SamplingSearchSpec, TrialSpec, load_study_spec, to_jsonable
from .trace import TraceRequest, load_trace_requests, select_requests_for_threshold


@dataclass(frozen=True)
class ProbePayload:
    threshold: float
    request_count: int
    pass_rate: float
    request_rate: float
    feasible: bool
    outcomes: list[dict[str, Any]]
    early_stopped: bool = False
    early_stop_reason: str = ""
    latency_summary: dict[str, Any] | None = None


def _percentile(values: list[float], p: float) -> float | None:
    if not values:
        return None
    ordered = sorted(values)
    idx = min(len(ordered) - 1, max(0, math.ceil((p / 100.0) * len(ordered)) - 1))
    return float(ordered[idx])


def _metric_summary(values: list[float]) -> dict[str, Any]:
    return {
        "count": len(values),
        "mean": float(statistics.fmean(values)) if values else None,
        "p50": _percentile(values, 50.0),
        "p90": _percentile(values, 90.0),
        "p95": _percentile(values, 95.0),
        "p99": _percentile(values, 99.0),
    }


def _reason_counts(evaluations: list[Any]) -> dict[str, int]:
    counts: dict[str, int] = {}
    for evaluation in evaluations:
        for reason in evaluation.reasons:
            counts[reason] = counts.get(reason, 0) + 1
    return counts


def _latency_summary(
    *,
    outcomes: list[RequestOutcome],
    evaluations: list[Any],
    study: Any,
) -> dict[str, Any]:
    ttft_values = [float(item.ttft_ms) for item in outcomes if item.ttft_ms is not None]
    tpot_values = [float(item.tpot_ms) for item in outcomes if item.tpot_ms is not None]
    return {
        "observed_request_count": len(outcomes),
        "request_mode": study.trace.request_mode,
        "ttft_ms": _metric_summary(ttft_values),
        "tpot_ms": _metric_summary(tpot_values),
        "failed_reason_counts": _reason_counts(evaluations),
        "slo": {
            "target_pass_rate": study.slo.target_pass_rate,
            "ttft_rule": study.slo.ttft_rule.__dict__ if study.slo.ttft_rule is not None else None,
            "tpot_rule": study.slo.tpot_rule.__dict__ if study.slo.tpot_rule is not None else None,
        },
    }

def _trial_spec_from_json(path: Path) -> TrialSpec:
    payload = json.loads(path.read_text(encoding="utf-8"))
    return TrialSpec(
        study_id=str(payload["study_id"]),
        trial_id=str(payload["trial_id"]),
        config_patch=ConfigPatch.from_dict(payload["config_patch"]),
        search=SamplingSearchSpec.from_dict(payload["search"]),
        study_spec_path=str(payload["study_spec_path"]),
        artifact_dir=str(payload["artifact_dir"]),
        probe_log_path=str(payload["probe_log_path"]),
        engine_log_path=str(payload["engine_log_path"]),
        result_path=str(payload["result_path"]),
        search_evidence=dict(payload.get("search_evidence") or {}),
    )


def _run_one_request(
    request: TraceRequest,
    *,
    base_url: str,
    timeout_s: float,
) -> RequestOutcome:
    try:
        metrics = stream_chat_completion(
            base_url=base_url,
            body=request.body,
            timeout_s=timeout_s,
            api_path=request.api_path,
        )
        expected_completion_tokens = request.completion_tokens_hint
        actual_completion_tokens = metrics.completion_tokens
        completion_tokens_source = getattr(metrics, "completion_tokens_source", "")
        if expected_completion_tokens is not None:
            if completion_tokens_source != "usage":
                return RequestOutcome(
                    request_id=request.row_id,
                    success=False,
                    ttft_ms=metrics.ttft_ms,
                    tpot_ms=metrics.tpot_ms,
                    prompt_tokens=request.prompt_tokens_hint,
                    completion_tokens=actual_completion_tokens,
                    error=(
                        "completion_tokens_unverified "
                        f"source={completion_tokens_source or 'unknown'} "
                        f"expected={expected_completion_tokens} "
                        f"actual={actual_completion_tokens}"
                    ),
                    completion_tokens_source=completion_tokens_source,
                    completed_mono_ns=time.monotonic_ns(),
                )
            if actual_completion_tokens != expected_completion_tokens:
                return RequestOutcome(
                    request_id=request.row_id,
                    success=False,
                    ttft_ms=metrics.ttft_ms,
                    tpot_ms=metrics.tpot_ms,
                    prompt_tokens=request.prompt_tokens_hint,
                    completion_tokens=actual_completion_tokens,
                    error=(
                        "completion_tokens_mismatch "
                        f"expected={expected_completion_tokens} "
                        f"actual={actual_completion_tokens}"
                    ),
                    completion_tokens_source=completion_tokens_source,
                    completed_mono_ns=time.monotonic_ns(),
                )
        return RequestOutcome(
            request_id=request.row_id,
            success=True,
            ttft_ms=metrics.ttft_ms,
            tpot_ms=metrics.tpot_ms,
            prompt_tokens=request.prompt_tokens_hint,
            completion_tokens=actual_completion_tokens or request.completion_tokens_hint,
            completion_tokens_source=completion_tokens_source,
            completed_mono_ns=time.monotonic_ns(),
        )
    except HttpClientError as exc:
        return RequestOutcome(
            request_id=request.row_id,
            success=False,
            ttft_ms=None,
            tpot_ms=None,
            prompt_tokens=request.prompt_tokens_hint,
            completion_tokens=request.completion_tokens_hint,
            error=str(exc),
            completed_mono_ns=time.monotonic_ns(),
        )


def _probe_outcome_details(
    *,
    threshold: float,
    selected: list[TraceRequest],
    outcomes: list[RequestOutcome],
    evaluations: list[Any],
    early_stopped: bool,
    early_stop_reason: str,
) -> dict[str, Any]:
    selected_by_id = {request.row_id: request for request in selected}
    return {
        "threshold": threshold,
        "early_stopped": early_stopped,
        "early_stop_reason": early_stop_reason,
        "outcomes": [
            {
                "request_id": outcome.request_id,
                "sampling_u": (
                    selected_by_id[outcome.request_id].sampling_u
                    if outcome.request_id in selected_by_id
                    else None
                ),
                "arrival_s": (
                    selected_by_id[outcome.request_id].arrival_s
                    if outcome.request_id in selected_by_id
                    else None
                ),
                "success": outcome.success,
                "ttft_ms": outcome.ttft_ms,
                "tpot_ms": outcome.tpot_ms,
                "prompt_tokens": outcome.prompt_tokens,
                "expected_completion_tokens": (
                    selected_by_id[outcome.request_id].completion_tokens_hint
                    if outcome.request_id in selected_by_id
                    else None
                ),
                "completion_tokens": outcome.completion_tokens,
                "completion_tokens_source": outcome.completion_tokens_source,
                "error": outcome.error,
                "evaluation": evaluation.passed,
                "reasons": evaluation.reasons,
            }
            for outcome, evaluation in zip(outcomes, evaluations)
        ],
    }


_SIGTERM_NOT_INSTALLED = object()


def _install_sigterm_as_keyboardinterrupt() -> Any:
    """Make SIGTERM raise KeyboardInterrupt so the engine-teardown finally runs.

    When `study tune` is killed, a default SIGTERM skips the finally blocks and
    orphans the vLLM engine (and its EngineCore workers) on the GPUs. Converting
    SIGTERM to KeyboardInterrupt lets _terminate_process_tree run. Only installable
    from the main thread; returns the previous handler (or a sentinel).
    """
    if threading.current_thread() is not threading.main_thread():
        return _SIGTERM_NOT_INSTALLED

    def _handler(signum: int, frame: Any) -> None:
        raise KeyboardInterrupt()

    try:
        return signal.signal(signal.SIGTERM, _handler)
    except (ValueError, OSError):
        return _SIGTERM_NOT_INSTALLED


def _restore_sigterm(previous: Any) -> None:
    if previous is _SIGTERM_NOT_INSTALLED:
        return
    if threading.current_thread() is not threading.main_thread():
        return
    try:
        signal.signal(signal.SIGTERM, previous)
    except (ValueError, OSError):
        pass


def _ignore_sigterm_if_main() -> None:
    """Ignore SIGTERM during teardown so a second signal cannot orphan the engine."""
    if threading.current_thread() is not threading.main_thread():
        return
    try:
        signal.signal(signal.SIGTERM, signal.SIG_IGN)
    except (ValueError, OSError):
        pass


def _probe_drain_deadline(
    reqs: list[TraceRequest], slo: Any, *, ceiling: float | None
) -> float | None:
    """Stop-A-consistent per-probe drain deadline (wall-clock seconds).

    The deadline is the time a *feasible* config needs to drain the admitted set:
    the last admitted arrival plus the worst-case TTFT budget plus the p99 output
    length times the TPOT budget. A config that cannot finish by this deadline is
    genuinely SLO-infeasible, so the clock never pre-empts the LCA-matched offered
    window (Stop-A) -- it only fails the unfit. ``ceiling`` is a hard safety cap.
    """
    if not reqs or slo.tpot_rule is None:
        return ceiling
    last_arrival = max(float(r.arrival_s or 0.0) for r in reqs)
    inputs = sorted(int(r.prompt_tokens_hint or 0) for r in reqs)
    outputs = sorted(int(r.completion_tokens_hint or 0) for r in reqs)

    def _p99(xs: list[int]) -> int:
        return xs[min(len(xs) - 1, int(0.99 * len(xs)))] if xs else 0

    p99_in, p99_out = _p99(inputs), _p99(outputs)
    tpot_ms = _rule_threshold_ms(slo.tpot_rule, p99_in)
    ttft_ms = _rule_threshold_ms(slo.ttft_rule, p99_in) if slo.ttft_rule is not None else 0.0
    margin_s = 30.0
    deadline = last_arrival + (ttft_ms + p99_out * tpot_ms) / 1000.0 + margin_s
    return min(float(ceiling), deadline) if ceiling else deadline


def _adaptive_replay_set(
    selected: list[TraceRequest],
    *,
    study: Any,
    window: Any,
) -> tuple[list[TraceRequest], dict[str, Any] | None]:
    """Stop-A: truncate the replay to the offered-L-C-A convergence prefix.

    Returns the (possibly shortened) request list to replay and a certificate of
    the convergence decision. When Stop-A is disabled, or C never converges, the
    full selected set is replayed (the C-gate: no early stop on a cold cache).
    """
    spec = study.trace.adaptive_stop
    if not getattr(spec, "enabled", False) or not selected:
        return selected, None
    point = find_convergence_prefix(
        selected,
        window,
        gpu_count=study.hardware.gpu_count,
        length_mode=resolve_length_mode(request_mode=study.trace.request_mode),
        tau=spec.tau,
        tau_c=spec.tau_c,
        stable_checks=spec.stable_checks,
        max_checks=spec.max_checks,
        min_fraction=spec.min_fraction,
    )
    replay = selected[: point.stop_index] if point.stop_index > 0 else selected
    certificate = {
        "enabled": True,
        "converged": point.converged,
        "stop_index": point.stop_index,
        "total_selected": len(selected),
        "fraction": point.fraction,
        "stop_time_s": point.stop_time_s,
        "family_similarity": point.family_similarity,
    }
    return replay, certificate


def _should_extend_on_boundary(
    *,
    pass_rate: float,
    target_pass_rate: float,
    certificate: dict[str, Any] | None,
    truncated: bool,
    boundary_delta: float,
) -> bool:
    """SLO-boundary guard: re-measure on the full window when a truncated probe
    lands within +/- boundary_delta of the SLO target.

    Offered-L-C-A convergence cannot see engine-state drift in the window's tail,
    so a near-boundary truncated verdict is untrustworthy. This fires only on
    probes sitting on the feasibility knee, so non-boundary probes keep the Stop-A
    time saving.
    """
    if certificate is None or not certificate.get("converged"):
        return False
    if not truncated or boundary_delta <= 0:
        return False
    return abs(float(pass_rate) - float(target_pass_rate)) <= float(boundary_delta)


def _best_feasible_probe_record(probe_history: list[dict[str, Any]]) -> dict[str, Any] | None:
    feasible = [
        item
        for item in probe_history
        if item.get("feasible") and isinstance(item.get("request_rate"), (int, float))
    ]
    if not feasible:
        return None
    return max(feasible, key=lambda item: float(item["request_rate"]))


def _binary_probe_resolution(search: SamplingSearchSpec) -> float:
    return max(
        float(search.tolerance),
        (float(search.high) - float(search.low)) / float(2 ** max(search.max_probes, 1)),
    )


def _measurement_ceiling_evidence(
    *,
    search: SamplingSearchSpec,
    requests: list[TraceRequest],
    best_threshold: float | None,
    best_payload: ProbePayload | None,
) -> dict[str, Any]:
    trace_max_sampling_u = max((float(request.sampling_u) for request in requests), default=None)
    policy = search.auto_high
    evidence: dict[str, Any] = {
        "auto_high": {
            "enabled": policy.enabled,
            "max_sampling_u": policy.max_sampling_u,
            "require_human_confirmation_beyond_trace": (
                policy.require_human_confirmation_beyond_trace
            ),
        },
        "search_high": search.high,
        "trace_max_sampling_u": trace_max_sampling_u,
        "measurement_ceiling_insufficient": False,
        "reason": "measurement_ceiling_not_reached",
    }
    if trace_max_sampling_u is None:
        evidence["reason"] = "trace_has_no_requests"
        return evidence
    if best_threshold is None or best_payload is None:
        evidence["reason"] = "no_feasible_probe"
        return evidence
    resolution = _binary_probe_resolution(search)
    threshold_gap_to_high = float(search.high) - float(best_threshold)
    evidence["best_sampling_u"] = best_threshold
    evidence["best_request_count"] = best_payload.request_count
    evidence["threshold_gap_to_high"] = threshold_gap_to_high
    evidence["binary_probe_resolution"] = resolution
    full_trace_selected = best_payload.request_count >= len(requests)
    high_reaches_trace = float(search.high) + 1e-12 >= float(trace_max_sampling_u)
    if (
        full_trace_selected
        and high_reaches_trace
        and threshold_gap_to_high <= resolution + 1e-12
    ):
        evidence["measurement_ceiling_insufficient"] = True
        evidence["reason"] = "measurement_ceiling_insufficient"
    return evidence


def _replay_requests(
    requests: list[TraceRequest],
    *,
    base_url: str,
    timeout_s: float,
    max_concurrency: int,
    target_pass_rate: float,
    max_lag_s: float | None,
    max_elapsed_s: float | None,
    evaluate_outcome: Callable[[RequestOutcome], Any],
    drain_inflight_on_early_stop: bool = True,
) -> tuple[list[RequestOutcome], bool, str]:
    outcomes_by_id: dict[str, RequestOutcome] = {}
    lock = threading.Lock()
    start = time.monotonic()
    allowed_failures = max(0, len(requests) - math.ceil(len(requests) * target_pass_rate))
    failed_evaluations = 0
    early_stopped = False
    early_stop_reason = ""
    next_index = 0
    futures_by_request: dict[Future[RequestOutcome], TraceRequest] = {}
    submitted_ids: set[str] = set()
    pool = ThreadPoolExecutor(max_workers=max_concurrency)
    try:
        while next_index < len(requests) or futures_by_request:
            now = time.monotonic()
            elapsed = now - start
            if max_elapsed_s is not None and elapsed > max_elapsed_s:
                early_stopped = True
                early_stop_reason = f"probe_elapsed_s>{max_elapsed_s}"
                break
            while next_index < len(requests):
                request = requests[next_index]
                lag_s = elapsed - request.arrival_s
                if max_lag_s is not None and lag_s > max_lag_s:
                    early_stopped = True
                    early_stop_reason = f"arrival_lag_s>{max_lag_s}"
                    break
                if request.arrival_s > elapsed:
                    break
                future = pool.submit(
                    _run_one_request,
                    request,
                    base_url=base_url,
                    timeout_s=timeout_s,
                )
                futures_by_request[future] = request
                submitted_ids.add(request.row_id)
                next_index += 1
                if len(futures_by_request) >= max_concurrency:
                    break
            if early_stopped:
                break
            if futures_by_request:
                timeout = 0.5
                if next_index < len(requests):
                    timeout = min(timeout, max(0.0, requests[next_index].arrival_s - elapsed))
                if max_elapsed_s is not None:
                    remaining_elapsed = max(0.0, max_elapsed_s - elapsed)
                    timeout = min(timeout, remaining_elapsed)
                done, _ = wait(
                    list(futures_by_request),
                    timeout=timeout,
                    return_when=FIRST_COMPLETED,
                )
                if not done:
                    continue
                for future in done:
                    request = futures_by_request.pop(future)
                    outcome = future.result()
                    with lock:
                        outcomes_by_id[outcome.request_id] = outcome
                    if not evaluate_outcome(outcome).passed:
                        failed_evaluations += 1
                        if failed_evaluations > allowed_failures:
                            early_stopped = True
                            early_stop_reason = "slo_pass_rate_unrecoverable"
                            break
                if early_stopped:
                    break
            elif next_index < len(requests):
                sleep_for = max(0.0, requests[next_index].arrival_s - elapsed)
                if sleep_for > 0:
                    time.sleep(min(sleep_for, 0.1))
        if early_stopped:
            if drain_inflight_on_early_stop and futures_by_request:
                done, not_done = wait(list(futures_by_request), timeout=timeout_s)
                for future in done:
                    request = futures_by_request[future]
                    try:
                        outcomes_by_id[request.row_id] = future.result(timeout=0)
                    except Exception:  # noqa: BLE001
                        outcomes_by_id[request.row_id] = RequestOutcome(
                            request_id=request.row_id,
                            success=False,
                            ttft_ms=None,
                            tpot_ms=None,
                            prompt_tokens=request.prompt_tokens_hint,
                            completion_tokens=request.completion_tokens_hint,
                            error=early_stop_reason or "probe_early_stopped",
                        )
                for future in not_done:
                    future.cancel()
                    request = futures_by_request[future]
                    outcomes_by_id[request.row_id] = RequestOutcome(
                        request_id=request.row_id,
                        success=False,
                        ttft_ms=None,
                        tpot_ms=None,
                        prompt_tokens=request.prompt_tokens_hint,
                        completion_tokens=request.completion_tokens_hint,
                        error=early_stop_reason or "probe_early_stopped",
                    )
            else:
                for future in list(futures_by_request):
                    future.cancel()
                for request in futures_by_request.values():
                    outcomes_by_id[request.row_id] = RequestOutcome(
                        request_id=request.row_id,
                        success=False,
                        ttft_ms=None,
                        tpot_ms=None,
                        prompt_tokens=request.prompt_tokens_hint,
                        completion_tokens=request.completion_tokens_hint,
                        error=early_stop_reason or "probe_early_stopped",
                    )
            for request in requests:
                if request.row_id in submitted_ids:
                    continue
                outcomes_by_id[request.row_id] = RequestOutcome(
                    request_id=request.row_id,
                    success=False,
                    ttft_ms=None,
                    tpot_ms=None,
                    prompt_tokens=request.prompt_tokens_hint,
                    completion_tokens=request.completion_tokens_hint,
                    error=early_stop_reason or "probe_early_stopped",
                )
        else:
            for future, request in list(futures_by_request.items()):
                try:
                    outcome = future.result(timeout=timeout_s)
                except Exception:  # noqa: BLE001
                    outcome = RequestOutcome(
                        request_id=request.row_id,
                        success=False,
                        ttft_ms=None,
                        tpot_ms=None,
                        prompt_tokens=request.prompt_tokens_hint,
                        completion_tokens=request.completion_tokens_hint,
                        error="probe_early_stopped",
                    )
                outcomes_by_id[outcome.request_id] = outcome
    finally:
        pool.shutdown(wait=False, cancel_futures=True)
    ordered = [outcomes_by_id[item.row_id] for item in requests if item.row_id in outcomes_by_id]
    return ordered, early_stopped, early_stop_reason


def _wait_for_server_or_exit(
    process: subprocess.Popen[str],
    *,
    base_url: str,
    healthcheck_path: str,
    ready_timeout_s: float,
) -> None:
    deadline = time.monotonic() + ready_timeout_s
    last_error = "server_not_ready"
    while time.monotonic() < deadline:
        exit_code = process.poll()
        if exit_code is not None:
            raise RuntimeError(f"engine_process_exited_before_ready exit_code={exit_code}")
        try:
            wait_for_server(base_url, healthcheck_path, timeout_s=1.0)
            return
        except HttpClientError as exc:
            last_error = str(exc)
        time.sleep(1.0)
    raise HttpClientError(f"Timed out waiting for {base_url}{healthcheck_path}: {last_error}")


def _process_group_exists(pgid: int) -> bool:
    try:
        os.killpg(pgid, 0)
        return True
    except ProcessLookupError:
        return False


def _pids_matching_env(marker_env: dict[str, str] | None) -> list[int]:
    if not marker_env:
        return []
    expected = {
        f"{key}={value}".encode()
        for key, value in marker_env.items()
    }
    pids: list[int] = []
    proc_root = Path("/proc")
    for entry in proc_root.iterdir():
        if not entry.name.isdigit():
            continue
        pid = int(entry.name)
        if pid == os.getpid():
            continue
        try:
            environ = (entry / "environ").read_bytes()
        except (FileNotFoundError, PermissionError, ProcessLookupError):
            continue
        if expected.issubset(set(environ.split(b"\0"))):
            pids.append(pid)
    return sorted(pids)


def _signal_pids(pids: list[int], sig: signal.Signals) -> None:
    for pid in pids:
        try:
            os.kill(pid, sig)
        except (ProcessLookupError, PermissionError):
            continue


def _terminate_process_tree(
    process: subprocess.Popen[str],
    *,
    timeout_s: float = 30.0,
    marker_env: dict[str, str] | None = None,
) -> None:
    try:
        pgid = os.getpgid(process.pid)
    except ProcessLookupError:
        # Children can keep the session/process group alive after the vLLM API
        # server exits.  In that case the group id is still the original pid
        # because the process was launched with start_new_session=True.
        pgid = process.pid
    try:
        os.killpg(pgid, signal.SIGTERM)
    except ProcessLookupError:
        pass
    _signal_pids(_pids_matching_env(marker_env), signal.SIGTERM)
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        if not _process_group_exists(pgid) and not _pids_matching_env(marker_env):
            return
        time.sleep(0.1)
    try:
        os.killpg(pgid, signal.SIGKILL)
    except ProcessLookupError:
        pass
    _signal_pids(_pids_matching_env(marker_env), signal.SIGKILL)
    if process.poll() is None:
        process.wait(timeout=timeout_s)


def run_trial(trial_spec_path: Path) -> dict[str, Any]:
    from .store import StudyStore

    trial = _trial_spec_from_json(trial_spec_path)
    study_spec_path = Path(Path(trial.study_spec_path).read_text(encoding="utf-8").strip())
    study = load_study_spec(study_spec_path)
    window, requests = load_trace_requests(study, study_spec_path=study_spec_path)
    recipe = build_launch_recipe(study.engine, trial.config_patch)
    artifact_dir = Path(trial.artifact_dir)
    artifact_dir.mkdir(parents=True, exist_ok=True)
    engine_log_path = Path(trial.engine_log_path)
    probe_details_path = artifact_dir / "probe_details.jsonl"
    if probe_details_path.exists():
        probe_details_path.unlink()
    trial_marker_env = {
        "AITUNER_STUDY_ID": trial.study_id,
        "AITUNER_TRIAL_ID": trial.trial_id,
    }
    with engine_log_path.open("w", encoding="utf-8") as engine_log:
        def launch_process() -> subprocess.Popen[str]:
            launch_env = {**recipe.env, **trial_marker_env}
            return subprocess.Popen(  # noqa: S603
                recipe.argv,
                cwd=recipe.cwd,
                env=launch_env,
                stdout=engine_log,
                stderr=subprocess.STDOUT,
                text=True,
                start_new_session=True,
            )

        process = launch_process()
        previous_sigterm = _install_sigterm_as_keyboardinterrupt()
        probe_history: list[dict[str, Any]] = []
        failure_stage = "engine_launch"
        try:
            _wait_for_server_or_exit(
                process,
                base_url=recipe.base_url,
                healthcheck_path=recipe.healthcheck_path,
                ready_timeout_s=recipe.ready_timeout_s,
            )
            failure_stage = "probe_search"

            def evaluator(threshold: float) -> ThresholdProbe[ProbePayload]:
                nonlocal process
                selected = select_requests_for_threshold(requests, threshold=threshold)
                replay_set, adaptive_stop_certificate = _adaptive_replay_set(
                    selected, study=study, window=window
                )
                restart_after_early_stop = study.trace.restart_engine_after_early_stop

                def _run(reqs: list[TraceRequest]) -> tuple[list[RequestOutcome], bool, str]:
                    return _replay_requests(
                        reqs,
                        base_url=recipe.base_url,
                        timeout_s=recipe.request_timeout_s,
                        max_concurrency=study.trace.max_concurrency,
                        target_pass_rate=study.slo.target_pass_rate,
                        max_lag_s=study.trace.early_stop_max_lag_s,
                        max_elapsed_s=_probe_drain_deadline(
                            reqs, study.slo, ceiling=study.trace.early_stop_max_elapsed_s
                        ),
                        evaluate_outcome=lambda outcome: evaluate_request(outcome, study.slo),
                        drain_inflight_on_early_stop=not restart_after_early_stop,
                    )

                outcomes, early_stopped, early_stop_reason = _run(replay_set)
                evaluations, summary = summarize_evaluations(outcomes, study.slo)
                if _should_extend_on_boundary(
                    pass_rate=summary["slo_pass_rate"],
                    target_pass_rate=study.slo.target_pass_rate,
                    certificate=adaptive_stop_certificate,
                    truncated=len(replay_set) < len(selected),
                    boundary_delta=study.trace.adaptive_stop.boundary_delta,
                ):
                    # On the feasibility knee the truncated verdict is untrustworthy;
                    # re-measure the full window and use that result.
                    replay_set = selected
                    outcomes, early_stopped, early_stop_reason = _run(selected)
                    evaluations, summary = summarize_evaluations(outcomes, study.slo)
                    if adaptive_stop_certificate is not None:
                        adaptive_stop_certificate["boundary_extended"] = True
                probe_details = _probe_outcome_details(
                    threshold=threshold,
                    selected=replay_set,
                    outcomes=outcomes,
                    evaluations=evaluations,
                    early_stopped=early_stopped,
                    early_stop_reason=early_stop_reason,
                )
                probe_details["adaptive_stop"] = adaptive_stop_certificate
                with probe_details_path.open("a", encoding="utf-8") as details_handle:
                    details_handle.write(
                        json.dumps(probe_details, ensure_ascii=False) + "\n"
                    )
                request_rate = (
                    len(selected) / max(window.window_end - window.window_start, 1e-9)
                    if selected
                    else 0.0
                )
                payload = ProbePayload(
                    threshold=threshold,
                    request_count=len(selected),
                    pass_rate=float(summary["slo_pass_rate"]),
                    request_rate=request_rate,
                    feasible=bool(summary["feasible"]),
                    early_stopped=early_stopped,
                    early_stop_reason=early_stop_reason,
                    latency_summary=_latency_summary(
                        outcomes=outcomes,
                        evaluations=evaluations,
                        study=study,
                    ),
                    outcomes=[
                        {
                            "request_id": outcome.request_id,
                            "success": outcome.success,
                            "ttft_ms": outcome.ttft_ms,
                            "tpot_ms": outcome.tpot_ms,
                            "prompt_tokens": outcome.prompt_tokens,
                            "completion_tokens": outcome.completion_tokens,
                            "completion_tokens_source": outcome.completion_tokens_source,
                            "evaluation": evaluation.passed,
                            "reasons": evaluation.reasons,
                        }
                        for outcome, evaluation in zip(outcomes, evaluations)
                    ],
                )
                probe_record = {
                    "threshold": threshold,
                    "request_count": payload.request_count,
                    "replayed_request_count": len(replay_set),
                    "pass_rate": payload.pass_rate,
                    "request_rate": payload.request_rate,
                    "feasible": payload.feasible,
                    "early_stopped": payload.early_stopped,
                    "early_stop_reason": payload.early_stop_reason,
                    "latency_summary": payload.latency_summary,
                    "adaptive_stop": adaptive_stop_certificate,
                }
                probe_history.append(probe_record)
                StudyStore.write_json(Path(trial.probe_log_path), probe_history)
                if early_stopped and restart_after_early_stop:
                    _terminate_process_tree(
                        process,
                        timeout_s=30.0,
                        marker_env=trial_marker_env,
                    )
                    process = launch_process()
                    _wait_for_server_or_exit(
                        process,
                        base_url=recipe.base_url,
                        healthcheck_path=recipe.healthcheck_path,
                        ready_timeout_s=recipe.ready_timeout_s,
                    )
                return ThresholdProbe(
                    threshold=threshold,
                    feasible=payload.feasible,
                    payload=payload,
                )

            primary_search = binary_search_max_feasible(
                low=trial.search.low,
                high=trial.search.high,
                tolerance=trial.search.tolerance,
                max_probes=trial.search.max_probes,
                evaluator=evaluator,
            )
            search_for_best = primary_search
            best = primary_search.best_feasible_payload
            best_source = "primary_search"
            fallback_search = None
            skipped_lower_range_fallback = False
            lower_range_fallback_skip_reason = ""
            original_search_low = float(study.search.low)
            inherited_search_floor = float(trial.search.low)
            if best is None and inherited_search_floor > original_search_low:
                if trial.search.inherit_incumbent_floor:
                    skipped_lower_range_fallback = True
                    lower_range_fallback_skip_reason = (
                        "primary_search_above_incumbent_floor_all_infeasible"
                    )
                else:
                    fallback_search = binary_search_max_feasible(
                        low=original_search_low,
                        high=inherited_search_floor,
                        tolerance=trial.search.tolerance,
                        max_probes=trial.search.max_probes,
                        evaluator=evaluator,
                    )
                    if fallback_search.best_feasible_payload is not None:
                        search_for_best = fallback_search
                        best = fallback_search.best_feasible_payload
                        best_source = "lower_range_fallback"

            def serialize_probe(probe: ThresholdProbe[ProbePayload]) -> dict[str, Any]:
                return {
                    "threshold": probe.threshold,
                    "feasible": probe.feasible,
                    "payload": {
                        "request_count": probe.payload.request_count,
                        "pass_rate": probe.payload.pass_rate,
                        "request_rate": probe.payload.request_rate,
                        "early_stopped": probe.payload.early_stopped,
                        "early_stop_reason": probe.payload.early_stop_reason,
                        "latency_summary": probe.payload.latency_summary,
                    },
                }

            all_probes = [
                *primary_search.probes,
                *((fallback_search.probes if fallback_search is not None else [])),
            ]
            measurement = _measurement_ceiling_evidence(
                search=trial.search,
                requests=requests,
                best_threshold=search_for_best.best_threshold if best is not None else None,
                best_payload=best,
            )
            measurement["auto_high_resolution"] = trial.search_evidence
            result = {
                "study_id": trial.study_id,
                "trial_id": trial.trial_id,
                "status": "completed",
                "config_patch": to_jsonable(trial.config_patch),
                "measurement": measurement,
                "best_source": best_source,
                "best_sampling_u": search_for_best.best_threshold if best is not None else None,
                "best_request_rate": best.request_rate if best is not None else None,
                "best_pass_rate": best.pass_rate if best is not None else None,
                "best_request_count": best.request_count if best is not None else None,
                "probes": [serialize_probe(probe) for probe in all_probes],
            }
            if fallback_search is not None or skipped_lower_range_fallback:
                result["primary_search"] = {
                    "low": inherited_search_floor,
                    "high": trial.search.high,
                    "best_sampling_u": primary_search.best_threshold
                    if primary_search.best_feasible_payload is not None
                    else None,
                    "best_request_rate": primary_search.best_feasible_payload.request_rate
                    if primary_search.best_feasible_payload is not None
                    else None,
                    "probes": [serialize_probe(probe) for probe in primary_search.probes],
                }
            if skipped_lower_range_fallback:
                result["lower_range_fallback"] = {
                    "triggered": False,
                    "skipped": True,
                    "reason": lower_range_fallback_skip_reason,
                    "low": original_search_low,
                    "high": inherited_search_floor,
                    "probes": [],
                }
            if fallback_search is not None:
                result["lower_range_fallback"] = {
                    "triggered": True,
                    "low": original_search_low,
                    "high": inherited_search_floor,
                    "best_sampling_u": fallback_search.best_threshold
                    if fallback_search.best_feasible_payload is not None
                    else None,
                    "best_request_rate": fallback_search.best_feasible_payload.request_rate
                    if fallback_search.best_feasible_payload is not None
                    else None,
                    "probes": [serialize_probe(probe) for probe in fallback_search.probes],
                }
            if best is None and all_probes:
                last_probe = all_probes[-1]
                result["all_infeasible_diagnostics"] = {
                    "threshold": last_probe.threshold,
                    "request_count": last_probe.payload.request_count,
                    "request_rate": last_probe.payload.request_rate,
                    "pass_rate": last_probe.payload.pass_rate,
                    "early_stopped": last_probe.payload.early_stopped,
                    "early_stop_reason": last_probe.payload.early_stop_reason,
                    "latency_summary": last_probe.payload.latency_summary,
                }
            StudyStore.write_json(Path(trial.result_path), result)
            return result
        except Exception as exc:  # noqa: BLE001
            partial_best = _best_feasible_probe_record(probe_history)
            if partial_best is not None:
                result = {
                    "study_id": trial.study_id,
                    "trial_id": trial.trial_id,
                    "status": "completed",
                    "config_patch": to_jsonable(trial.config_patch),
                    "best_source": "partial_probe_before_failure",
                    "best_sampling_u": partial_best.get("threshold"),
                    "best_request_rate": partial_best.get("request_rate"),
                    "best_pass_rate": partial_best.get("pass_rate"),
                    "best_request_count": partial_best.get("request_count"),
                    "completed_with_probe_failure": True,
                    "failure_stage": failure_stage,
                    "failure_reason": str(exc),
                    "failure_traceback": traceback.format_exc(),
                    "probes": probe_history,
                }
                StudyStore.write_json(Path(trial.result_path), result)
                return result
            result = {
                "study_id": trial.study_id,
                "trial_id": trial.trial_id,
                "status": "failed",
                "config_patch": to_jsonable(trial.config_patch),
                "best_sampling_u": None,
                "best_request_rate": None,
                "best_pass_rate": None,
                "best_request_count": None,
                "failure_stage": failure_stage,
                "failure_reason": str(exc),
                "failure_traceback": traceback.format_exc(),
                "probes": probe_history,
            }
            StudyStore.write_json(Path(trial.result_path), result)
            return result
        finally:
            _ignore_sigterm_if_main()
            _terminate_process_tree(process, timeout_s=30.0, marker_env=trial_marker_env)
            _restore_sigterm(previous_sigterm)
