tests/test_profiling.py

from __future__ import annotations

import json
import os
from collections.abc import Iterator
from contextlib import contextmanager
from decimal import Decimal
from pathlib import Path
from types import SimpleNamespace
from typing import Any, cast

import httpx
import pytest

from harbor_hf.endpoints import bind_endpoint
from harbor_hf.executions import build_execution_lock
from harbor_hf.harbor_adapter.errors import HarborTrialFailure
from harbor_hf.harbor_adapter.models import HarborCompatibilityTrial
from harbor_hf.models import (
    DeploymentProfile,
    EndpointRef,
    ExperimentSpec,
    ServingProfileBinding,
)
from harbor_hf.profile_preflight import preflight_profile_plan
from harbor_hf.profile_submission import build_profile_submit_command, submit_profile
from harbor_hf.profile_worker import (
    ProfileCleanupUnverified,
    ProfileWorkerError,
    _finalize_profile,
    _load_recoverable_points,
    _point_ladder_rate,
    _point_workload,
    _PointResult,
    _prepare_profile_destination,
    _profile_judge_assignments,
    _profile_judge_transport,
    _request,
    _run_ladder,
    _run_point,
    _SmokeObservation,
    _summarize_point,
    _TaskObservation,
    _verify_smoke,
    _write_point,
    run_profile_worker,
)
from harbor_hf.profiling import (
    ProfileObjective,
    ProfilePlan,
    ProfilePoint,
    bind_profile_target,
    build_profile_plan,
    canonical_digest,
    new_unselected_profile,
    select_profile,
)
from harbor_hf.provider_models import (
    ExplicitProviderRoute,
    ProviderLimits,
    ProviderTarget,
)


def test_profile_finalizer_records_special_nodes(tmp_path: Path) -> None:
    evidence = tmp_path / "evidence.json"
    evidence.write_text('{"status":"complete"}\n', encoding="utf-8")
    fifo = tmp_path / "runtime.pipe"
    os.mkfifo(fifo)

    _finalize_profile(tmp_path, "hf_test_secret")

    assert evidence.exists()
    assert not fifo.exists()
    rejection = json.loads(
        (tmp_path / "private-artifact-rejections.json").read_text(encoding="utf-8")
    )
    assert rejection["rejections"] == [
        {"path": "runtime.pipe", "reason": "special_file", "size": None}
    ]
    assert (tmp_path / "checksums.json").exists()


def test_profile_finalizer_retries_transient_bucket_visibility(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    evidence = tmp_path / "evidence.json"
    evidence.write_text('{"status":"complete"}\n', encoding="utf-8")
    calls = 0

    def scrub(
        _root: Path, _secrets: object, *, allow_symlinks: bool = False
    ) -> list[str]:
        del allow_symlinks
        nonlocal calls
        calls += 1
        if calls == 1:
            raise FileNotFoundError(evidence)
        return []

    monkeypatch.setattr("harbor_hf.profile_worker.scrub_secret", scrub)
    monkeypatch.setattr("harbor_hf.profile_worker.time.sleep", lambda _delay: None)

    _finalize_profile(tmp_path, "hf_test_secret")

    assert calls == 2
    assert (tmp_path / "checksums.json").exists()


def profiled_spec(spec: ExperimentSpec) -> ExperimentSpec:
    return spec.model_copy(
        update={
            "execution": spec.execution.model_copy(
                update={
                    "server_context_tokens": 65_536,
                    "max_output_tokens": 8192,
                    "reasoning_required": True,
                }
            )
        }
    )


def profiled_provider_spec(spec: ExperimentSpec) -> ExperimentSpec:
    profiled = profiled_spec(spec)
    task_digests = {
        f"provider-task-{index:02d}": "sha256:" + f"{index:064x}" for index in range(32)
    }
    agent = profiled.matrix.agents[0].model_copy(
        update={
            "import_path": "harbor_hf_agents.openclaw.agent:OpenClawAgent",
            "parameters": {"openclaw_config": {}},
        }
    )
    return profiled.model_copy(
        update={
            "benchmark": profiled.benchmark.model_copy(
                update={
                    "task_names": sorted(task_digests),
                    "task_digests": task_digests,
                }
            ),
            "matrix": profiled.matrix.model_copy(update={"agents": [agent]}),
        }
    )


def plan(spec: ExperimentSpec) -> ProfilePlan:
    return build_profile_plan(
        profiled_spec(spec),
        profile_id="profile-one",
        candidate_concurrency=[1, 2, 4, 8],
        max_spend_usd="10.00",
        profile_timeout_seconds=3600,
    )


def point(concurrency: int, throughput: float) -> ProfilePoint:
    payload = {
        "concurrency": concurrency,
        "repetition": 1,
        "status": "completed",
        "planned_count": 8,
        "completed_count": 8,
        "failed_count": 0,
        "error_rate": 0.0,
        "goodput_rate": 1.0,
        "aggregate_output_tokens_per_second": float(throughput),
        "tasks_per_hour": float(throughput),
        "artifact_prefix": f"points/{concurrency}/1.json",
    }
    return ProfilePoint.model_validate(
        {"point_sha256": canonical_digest(payload), **payload}
    )


def failed_repetition(concurrency: int, repetition: int) -> ProfilePoint:
    payload = {
        "concurrency": concurrency,
        "repetition": repetition,
        "status": "failed",
        "planned_count": 8,
        "completed_count": 0,
        "failed_count": 8,
        "error_rate": 1.0,
        "goodput_rate": 0.0,
        "artifact_prefix": f"points/{concurrency}/{repetition}/evidence.json",
        "failure_reason": "failed",
    }
    return ProfilePoint.model_validate(
        {"point_sha256": canonical_digest(payload), **payload}
    )


def run_expected_profile(
    plan_path: Path,
    output_root: Path,
    *,
    smoke_fails: bool,
) -> None:
    if smoke_fails:
        with pytest.raises(ProfileWorkerError, match="smoke failed"):
            run_profile_worker(plan_path, output_root)
        destination = output_root / "serving-profiles/profile-one"
        assert (destination / "_FAILED").is_file()
        checksums = json.loads((destination / "checksums.json").read_text())
        assert set(checksums) == {"failure.json", "lifecycle.json", "plan.json"}
        return
    destination = run_profile_worker(plan_path, output_root)
    assert (destination / "_SELECTED").is_file()
    checksums = json.loads((destination / "checksums.json").read_text())
    assert set(checksums) == {"lifecycle.json", "plan.json", "profile.json"}


def test_profile_worker_recovers_selected_profile_after_restart(
    remote_spec: ExperimentSpec,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    resolved = plan(remote_spec)
    plan_path = tmp_path / "plan.json"
    plan_path.write_text(resolved.model_dump_json(), encoding="utf-8")
    output_root = tmp_path / "output"
    destination = output_root / resolved.artifacts.prefix
    recovery = _prepare_profile_destination(resolved, destination)
    assert recovery is not None
    selected = select_profile(
        new_unselected_profile(resolved).model_copy(update={"points": [point(1, 10)]})
    )
    (destination / "profile.json").write_text(
        selected.model_dump_json(), encoding="utf-8"
    )

    class Manager:
        def __init__(self, *_args: object, **_kwargs: object) -> None:
            pass

        def describe(self) -> dict[str, object]:
            return {}

    monkeypatch.setenv("HF_TOKEN", "hf_test")
    monkeypatch.setattr("harbor_hf.profile_worker.EndpointManager", Manager)
    monkeypatch.setattr(
        "harbor_hf.profile_worker.validate_endpoint_model", lambda *_args: None
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker.require_paused_endpoint", lambda *_args: None
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker.require_executable", lambda *_args: None
    )

    recovered = run_profile_worker(plan_path, output_root)

    assert recovered == destination
    assert (destination / "_SELECTED").read_text(encoding="utf-8").strip() == (
        canonical_digest(selected)
    )


def test_profile_ladder_reuses_validated_points_after_restart(
    remote_spec: ExperimentSpec,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    resolved = plan(remote_spec).model_copy(update={"candidate_concurrency": [1, 2]})
    existing = point(1, 10)
    calls: list[tuple[int, int]] = []

    def run_point(*_args: object, repetition: int, **_kwargs: object) -> _PointResult:
        calls.append((2, repetition))
        observations = [
            _TaskObservation(True, 1000, 10, 20, f"task-{index}") for index in range(8)
        ]
        return _PointResult(observations, 8000)

    monkeypatch.setattr("harbor_hf.profile_worker._run_point", run_point)
    monkeypatch.setattr("harbor_hf.profile_worker._write_point", lambda *_args: None)
    monkeypatch.setattr(
        "harbor_hf.profile_worker._run_boundary_repetitions",
        lambda *_args, **_kwargs: [],
    )

    points = _run_ladder(
        resolved,
        cast(Any, None),
        cast(Any, None),
        tmp_path,
        "token",
        tmp_path,
        10**12,
        existing_points=[existing],
    )

    assert points[0] == existing
    assert calls == [(2, 1)]


def test_profile_recovery_recomputes_raw_point_evidence(
    remote_spec: ExperimentSpec, tmp_path: Path
) -> None:
    resolved = plan(remote_spec)
    observations = [
        _TaskObservation(True, 1000, 10, 20, f"task-{index}") for index in range(8)
    ]
    measured = _summarize_point(1, observations, elapsed_ms=8000, repetition=1)
    _write_point(tmp_path, measured, observations, 8000)

    recovered = _load_recoverable_points(resolved, tmp_path)

    assert recovered == [measured]


def test_profile_plan_is_deterministic(remote_spec: ExperimentSpec) -> None:
    first = plan(remote_spec)
    second = plan(remote_spec)

    assert first == second
    assert first.identity.server_context_tokens == 65_536
    assert first.workload.sample_task_count == 1
    assert first.plan_sha256.startswith("sha256:")


def test_profile_selection_prefers_lower_concurrency_on_tie(
    remote_spec: ExperimentSpec,
) -> None:
    profile = new_unselected_profile(plan(remote_spec)).model_copy(
        update={"points": [point(1, 20), point(2, 20), point(4, 10)]}
    )

    selected = select_profile(profile)

    assert selected.selection is not None
    assert selected.selection.concurrency == 1


def test_profile_selection_rejects_tampered_point(remote_spec: ExperimentSpec) -> None:
    profile = new_unselected_profile(plan(remote_spec)).model_copy(
        update={
            "points": [
                point(1, 20).model_copy(
                    update={"aggregate_output_tokens_per_second": 200}
                )
            ]
        }
    )

    with pytest.raises(ValueError, match="point digest"):
        select_profile(profile)


def test_serving_profile_rejects_point_outside_candidate_ladder(
    remote_spec: ExperimentSpec,
) -> None:
    profile = new_unselected_profile(plan(remote_spec))

    with pytest.raises(ValueError, match="points must be in the candidate ladder"):
        type(profile).model_validate(
            profile.model_copy(update={"points": [point(16, 20)]}).model_dump()
        )


def test_maximum_goodput_does_not_discount_failed_tasks_twice(
    remote_spec: ExperimentSpec,
) -> None:
    resolved = new_unselected_profile(plan(remote_spec))

    def measured(
        concurrency: int, tasks_per_hour: float, goodput: float
    ) -> ProfilePoint:
        payload = point(concurrency, tasks_per_hour).model_dump(
            mode="json", exclude={"point_sha256"}, exclude_none=True
        )
        payload.update(
            error_rate=1 - goodput,
            goodput_rate=goodput,
            tasks_per_hour=float(tasks_per_hour),
        )
        return ProfilePoint.model_validate(
            {"point_sha256": canonical_digest(payload), **payload}
        )

    profile = resolved.model_copy(
        update={
            "objective": resolved.objective.model_copy(
                update={"maximum_error_rate": 0.5}
            ),
            "points": [measured(1, 80, 1.0), measured(2, 100, 0.5)],
        }
    )

    selected = select_profile(profile)

    assert selected.selection is not None
    assert selected.selection.concurrency == 2


def test_profile_selection_disqualifies_failed_boundary_repetition(
    remote_spec: ExperimentSpec,
) -> None:
    first = point(1, 10)
    second = point(2, 20)
    repeated = second.model_copy(
        update={
            "repetition": 2,
            "point_sha256": canonical_digest(
                second.model_dump(
                    mode="json",
                    exclude={"point_sha256"},
                    exclude_none=True,
                )
                | {"repetition": 2}
            ),
        }
    )
    profile = new_unselected_profile(plan(remote_spec)).model_copy(
        update={
            "points": [first, second, repeated, failed_repetition(2, 3)],
        }
    )

    selected = select_profile(profile)

    assert selected.selection is not None
    assert selected.selection.concurrency == 1


def test_point_throughput_uses_complete_wall_time() -> None:
    observations = [
        _TaskObservation(True, 1000, 10, 20, f"task-{index}") for index in range(8)
    ]

    result = _summarize_point(1, observations, elapsed_ms=8000, repetition=1)

    assert result.tasks_per_hour == 3600
    assert result.aggregate_output_tokens_per_second == 20
    assert result.ttft_ms_p95 is None
    assert result.tpot_ms_p95 is None


def compatibility_trial(
    task_name: str, *, exception_type: str | None = None
) -> HarborCompatibilityTrial:
    digest = "sha256:" + "1" * 64
    return HarborCompatibilityTrial.model_validate(
        {
            "path": f"job/{task_name}",
            "trial_id": task_name,
            "trial_name": task_name,
            "lock_digest": digest,
            "result_digest": digest,
            "task_name": task_name,
            "task_digest": digest,
            "agent_name": "openclaw",
            "agent_version": "1",
            "exception_type": exception_type,
            "step_exceptions": [],
            "rewards": {"reward": 0},
            "timing": {
                "trial": {
                    "started_at": "2026-01-01T00:00:00+00:00",
                    "finished_at": "2026-01-01T00:00:01+00:00",
                }
            },
            "usage": {
                "input_tokens": 10,
                "cache_tokens": 0,
                "output_tokens": 5,
                "cost_usd": 0,
            },
            "artifacts": [],
        }
    )


def test_profile_judge_assignments_preserve_all_calls(tmp_path: Path) -> None:
    native = compatibility_trial("judged")
    verifier = tmp_path / "job" / "judged" / "verifier"
    verifier.mkdir(parents=True)
    (verifier / "judge-selection.json").write_text(
        json.dumps(
            {
                "schema_version": "harbor-hf/judge-selection/v1",
                "exchange_id": "judge-0002",
            }
        ),
        encoding="utf-8",
    )
    (verifier / "judge-calls.json").write_text(
        json.dumps(
            {
                "schema_version": "harbor-hf/judge-calls/v1",
                "exchange_ids": ["judge-0001", "judge-0002"],
            }
        ),
        encoding="utf-8",
    )

    assigned = _profile_judge_assignments([native], tmp_path)

    assert assigned == {"judge-0001": native, "judge-0002": native}


def test_profile_point_preserves_individual_harbor_trial_failures(
    remote_spec: ExperimentSpec,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    resolved = plan(remote_spec)
    request_path = tmp_path / "execution" / "harbor-request.json"
    request_path.parent.mkdir()
    prepared = SimpleNamespace(request_path=request_path, request=object())

    class Adapter:
        def prepare(self, *_args: object, **_kwargs: object) -> object:
            return prepared

        def execute(self, *_args: object, **_kwargs: object) -> object:
            raise HarborTrialFailure("one failed", "SandboxError")

    class Transport:
        @contextmanager
        def scope(self, _scope: str) -> Iterator[tuple[str, str, None]]:
            yield "https://endpoint.test", "model", None

    @contextmanager
    def process_environment(
        *_args: object, **_kwargs: object
    ) -> Iterator[dict[str, str]]:
        yield {}

    @contextmanager
    def judge_scope(*_args: object, **_kwargs: object) -> Iterator[tuple[None, None]]:
        yield None, None

    monkeypatch.setattr(
        "harbor_hf.profile_worker.FilesystemHarborExecutionAdapter", Adapter
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker.harbor_process_environment", process_environment
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker._sample_tasks",
        lambda _plan: {"success": "digest", "failure": "digest"},
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker.load_compatibility_bundle",
        lambda *_args: SimpleNamespace(
            trials=[
                compatibility_trial("success"),
                compatibility_trial("failure", exception_type="SandboxError"),
            ]
        ),
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker._assemble_profile_point_evidence",
        lambda *_args: None,
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker._profile_point_judge_scope",
        judge_scope,
    )

    result = _run_point(
        resolved,
        cast(Any, object()),
        cast(Any, Transport()),
        tmp_path / "harbor",
        "hf_test",
        1,
        repetition=1,
        destination=tmp_path / "profile",
        deadline=10**12,
    )

    assert [observation.success for observation in result.observations] == [True, False]
    assert result.observations[1].error == "SandboxError"
    assert (tmp_path / "profile" / "points" / "1" / "1" / "harbor-jobs").is_dir()
    assert (tmp_path / "profile" / "points" / "1" / "1" / "harbor-execution").is_dir()


def test_profile_ladder_skips_repetition_used_by_health_retry(
    remote_spec: ExperimentSpec,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    resolved = plan(remote_spec).model_copy(update={"candidate_concurrency": [1]})
    repetitions: list[int] = []

    def run_point(*_args: object, repetition: int, **_kwargs: object) -> _PointResult:
        repetitions.append(repetition)
        success = repetition != 1
        observations = [
            _TaskObservation(success, 1000, 10, 20, f"task-{index}")
            for index in range(8)
        ]
        return _PointResult(observations, 8000)

    monkeypatch.setattr("harbor_hf.profile_worker._run_point", run_point)
    monkeypatch.setattr("harbor_hf.profile_worker._verify_smoke", lambda *_args: None)
    monkeypatch.setattr("harbor_hf.profile_worker._write_point", lambda *_args: None)

    points = _run_ladder(
        resolved,
        cast(Any, None),
        cast(Any, None),
        tmp_path,
        "token",
        tmp_path,
        float("inf"),
    )

    assert repetitions == [1, 2, 3, 4]
    profile = new_unselected_profile(resolved).model_copy(update={"points": points})
    assert select_profile(profile).selection is not None


def test_profile_ladder_continues_after_successful_health_retry(
    remote_spec: ExperimentSpec,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    resolved = plan(remote_spec).model_copy(update={"candidate_concurrency": [1, 2]})
    calls: list[tuple[int, int]] = []

    def run_point(
        *args: object,
        repetition: int,
        **_kwargs: object,
    ) -> _PointResult:
        concurrency = cast(int, args[5])
        calls.append((concurrency, repetition))
        success = (concurrency, repetition) != (1, 1)
        observations = [
            _TaskObservation(success, 1000, 10, 20, f"task-{index}")
            for index in range(8)
        ]
        return _PointResult(observations, 8000)

    monkeypatch.setattr("harbor_hf.profile_worker._run_point", run_point)
    monkeypatch.setattr("harbor_hf.profile_worker._verify_smoke", lambda *_args: None)
    monkeypatch.setattr("harbor_hf.profile_worker._write_point", lambda *_args: None)
    monkeypatch.setattr(
        "harbor_hf.profile_worker._run_boundary_repetitions",
        lambda *_args, **_kwargs: [],
    )

    _run_ladder(
        resolved,
        cast(Any, None),
        cast(Any, None),
        tmp_path,
        "token",
        tmp_path,
        10**12,
    )

    assert calls == [(1, 1), (1, 2), (2, 1)]


def test_profile_ladder_uses_selected_objective_metric(
    remote_spec: ExperimentSpec,
) -> None:
    resolved = plan(remote_spec)
    throughput = resolved.model_copy(
        update={
            "objective": resolved.objective.model_copy(
                update={"kind": "maximum_throughput"}
            )
        }
    )
    measured = point(1, 100).model_copy(
        update={"aggregate_output_tokens_per_second": 200.0}
    )

    assert _point_ladder_rate(throughput, measured) == 200.0
    stable = resolved.model_copy(
        update={
            "objective": resolved.objective.model_copy(
                update={"kind": "maximum_stable_concurrency"}
            )
        }
    )
    assert _point_ladder_rate(stable, measured) is None


def test_profile_plan_rejects_unmeasured_latency_objectives(
    remote_spec: ExperimentSpec,
) -> None:
    with pytest.raises(ValueError, match="streaming measurements"):
        build_profile_plan(
            profiled_spec(remote_spec),
            profile_id="latency-profile",
            candidate_concurrency=[1],
            max_spend_usd="10",
            profile_timeout_seconds=3600,
            objective=ProfileObjective(maximum_ttft_ms_p95=1000),
        )


def test_endpoint_smoke_does_not_forward_endpoint_settings(
    remote_spec: ExperimentSpec, monkeypatch: pytest.MonkeyPatch
) -> None:
    resolved = plan(remote_spec)
    captured: dict[str, object] = {}

    def post(*_args: object, **kwargs: object) -> httpx.Response:
        payload = kwargs["json"]
        assert isinstance(payload, dict)
        for key, value in payload.items():
            assert isinstance(key, str)
            captured[key] = value
        return httpx.Response(
            200,
            json={
                "choices": [{"message": {"content": "OK"}, "finish_reason": "stop"}],
                "usage": {"prompt_tokens": 4, "completion_tokens": 1},
            },
            request=httpx.Request("POST", "https://endpoint.test"),
        )

    monkeypatch.setattr("harbor_hf.profile_worker.httpx.post", post)

    observation = _request(
        resolved,
        "https://endpoint.test",
        "model",
        "token",
        "OK",
        tools=False,
        timeout=10,
    )

    assert observation.success
    assert "min_replicas" not in captured
    assert "health_route" not in captured


def test_smoke_verifies_declared_context_and_output_limits(
    remote_spec: ExperimentSpec,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    resolved = plan(remote_spec)
    calls: list[tuple[int | None, int]] = []

    def request(
        _plan: ProfilePlan,
        _base_url: str,
        _model_name: str,
        _token: str,
        prompt: str,
        *,
        tools: bool,
        timeout: int,
        max_tokens: int | None = None,
        allow_length: bool = False,
        require_visible_output: bool = True,
    ) -> _SmokeObservation:
        del timeout, allow_length, require_visible_output
        repeats = prompt.count("x ")
        calls.append((max_tokens, repeats))
        if tools:
            return _SmokeObservation(True, 20, 2, False, True, True)
        if repeats:
            return _SmokeObservation(True, repeats + 20, 1, True, False, False)
        return _SmokeObservation(True, 20, 2, True, True, False)

    class Transport:
        @contextmanager
        def scope(self, _scope: str) -> Iterator[tuple[str, str, None]]:
            yield "https://endpoint.test", "model", None

    monkeypatch.setattr("harbor_hf.profile_worker._request", request)

    _verify_smoke(resolved, cast(Any, Transport()), "token", 10**12)

    assert calls[-1][0] == 8192
    assert calls[-1][1] + 20 + 8192 >= 65_536 - 512


def test_provider_smoke_retries_transient_request_failure(
    remote_spec: ExperimentSpec,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    model = remote_spec.matrix.models[0]
    provider = ProviderTarget(
        id="provider",
        model=model.repo,
        routing=ExplicitProviderRoute(provider="fireworks-ai"),
        limits=ProviderLimits(max_concurrent_requests=2, max_attempts=3),
    )
    spec = profiled_provider_spec(
        remote_spec.model_copy(
            update={
                "matrix": remote_spec.matrix.model_copy(
                    update={"deployments": [provider]}
                )
            }
        )
    )
    resolved = build_profile_plan(
        spec,
        profile_id="provider-profile",
        candidate_concurrency=[1, 2],
        max_spend_usd="10",
        profile_timeout_seconds=3600,
    )
    calls = 0

    def request(
        _plan: ProfilePlan,
        _base_url: str,
        _model_name: str,
        _token: str,
        prompt: str,
        *,
        tools: bool,
        timeout: int,
        max_tokens: int | None = None,
        allow_length: bool = False,
        require_visible_output: bool = True,
    ) -> _SmokeObservation:
        nonlocal calls
        del timeout, max_tokens, allow_length, require_visible_output
        calls += 1
        if calls == 1:
            return _SmokeObservation(
                False, 0, 0, False, False, False, "HTTPStatusError"
            )
        repeats = prompt.count("x ")
        if tools:
            return _SmokeObservation(True, 20, 2, False, True, True)
        if repeats:
            return _SmokeObservation(True, repeats + 20, 1, True, False, False)
        return _SmokeObservation(True, 20, 2, True, True, False)

    class Transport:
        @contextmanager
        def scope(self, _scope: str) -> Iterator[tuple[str, str, None]]:
            yield "https://provider.test", "model", None

    monkeypatch.setattr("harbor_hf.profile_worker._request", request)
    monkeypatch.setattr("harbor_hf.profile_worker.time.sleep", lambda _delay: None)

    _verify_smoke(resolved, cast(Any, Transport()), "token", 10**12)

    assert calls == 6


def test_context_calibration_accepts_usage_before_visible_output(
    remote_spec: ExperimentSpec,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    resolved = plan(remote_spec)

    def post(*_args: object, **_kwargs: object) -> httpx.Response:
        return httpx.Response(
            200,
            json={
                "choices": [
                    {
                        "message": {"content": ""},
                        "finish_reason": "length",
                    }
                ],
                "usage": {"prompt_tokens": 282, "completion_tokens": 1},
            },
            request=httpx.Request("POST", "https://endpoint.test"),
        )

    monkeypatch.setattr("harbor_hf.profile_worker.httpx.post", post)

    observation = _request(
        resolved,
        "https://endpoint.test",
        "model",
        "token",
        "x " * 256,
        tools=False,
        timeout=10,
        max_tokens=1,
        allow_length=True,
        require_visible_output=False,
    )

    assert observation.success
    assert observation.input_tokens == 282
    assert observation.output_tokens == 1


def test_serving_profile_binding_fails_closed_on_concurrency(
    remote_spec: ExperimentSpec,
) -> None:
    resolved = plan(remote_spec)
    binding = ServingProfileBinding(
        profile_id=resolved.profile_id,
        profile_sha256="sha256:" + "9" * 64,
        artifact_uri="hf://buckets/example-org/benchmark-executions/serving-profiles/profile-one/profile.json",
        concurrency=2,
        **resolved.identity.model_dump(mode="python"),
    )
    spec = profiled_spec(remote_spec)
    execution = spec.execution.model_copy(update={"serving_profile": binding})

    with pytest.raises(ValueError, match="concurrent_trials"):
        ExperimentSpec.model_validate(
            spec.model_copy(update={"execution": execution}).model_dump(mode="python")
        )


def test_serving_profile_binding_fails_closed_on_workload_identity(
    remote_spec: ExperimentSpec,
) -> None:
    spec = profiled_spec(remote_spec)
    resolved = plan(remote_spec)
    binding = ServingProfileBinding(
        profile_id=resolved.profile_id,
        profile_sha256="sha256:" + "9" * 64,
        artifact_uri="hf://buckets/example-org/benchmark-executions/serving-profiles/profile-one/profile.json",
        concurrency=spec.execution.concurrent_trials,
        **resolved.identity.model_dump(mode="python"),
    )

    reasoning_execution = spec.execution.model_copy(
        update={
            "reasoning_required": False,
            "serving_profile": binding,
        }
    )
    with pytest.raises(ValueError, match="reasoning mode"):
        ExperimentSpec.model_validate(
            spec.model_copy(update={"execution": reasoning_execution}).model_dump(
                mode="python"
            )
        )

    assert spec.remote is not None
    changed_remote = spec.remote.model_copy(
        update={
            "harbor": spec.remote.harbor.model_copy(
                update={"sandbox_flavor": "cpu-performance"}
            )
        }
    )
    runtime_execution = spec.execution.model_copy(update={"serving_profile": binding})
    with pytest.raises(ValueError, match="harbor_runtime_sha256"):
        ExperimentSpec.model_validate(
            spec.model_copy(
                update={"execution": runtime_execution, "remote": changed_remote}
            ).model_dump(mode="python")
        )

    changed_binding = binding.model_copy(
        update={"sample_tasks_sha256": "sha256:" + "8" * 64}
    )
    workload_execution = spec.execution.model_copy(
        update={"serving_profile": changed_binding}
    )
    with pytest.raises(ValueError, match="sampled workload"):
        ExperimentSpec.model_validate(
            spec.model_copy(update={"execution": workload_execution}).model_dump(
                mode="python"
            )
        )


def test_managed_endpoint_binding_preserves_profile_identity(
    remote_spec: ExperimentSpec,
) -> None:
    deployment = remote_spec.matrix.deployments[0].model_copy(update={"endpoint": None})
    spec = profiled_spec(
        remote_spec.model_copy(
            update={
                "matrix": remote_spec.matrix.model_copy(
                    update={"deployments": [deployment]}
                )
            }
        )
    )
    resolved = build_profile_plan(
        spec,
        profile_id="profile-one",
        candidate_concurrency=[1],
        max_spend_usd="5.00",
        profile_timeout_seconds=3600,
    )
    binding = ServingProfileBinding(
        profile_id=resolved.profile_id,
        profile_sha256="sha256:" + "9" * 64,
        artifact_uri=(
            "hf://buckets/example-org/benchmark-executions/serving-profiles/"
            "profile-one/profile.json"
        ),
        concurrency=1,
        **resolved.identity.model_dump(mode="python"),
    )
    profiled = ExperimentSpec.model_validate(
        spec.model_copy(
            update={
                "execution": spec.execution.model_copy(
                    update={"serving_profile": binding}
                )
            }
        ).model_dump(mode="python")
    )

    bound = bind_endpoint(
        profiled,
        deployment_id=deployment.id,
        endpoint=EndpointRef(
            namespace="example-org",
            name="managed-profile-endpoint",
            served_model_name="/repository",
        ),
    )

    assert ExperimentSpec.model_validate(bound.model_dump(mode="python")) == bound


def test_profile_submit_command_is_remote_only(remote_spec: ExperimentSpec) -> None:
    command = build_profile_submit_command(
        plan(remote_spec), input_dir="hf://buckets/input", bucket="example-org/results"
    )

    assert command[:3] == ["hf", "jobs", "run"]
    assert "profile-worker" in command
    assert "/input/plan.json" in command
    assert not any("llama-server" in argument for argument in command)


def test_provider_profile_submit_command_exposes_recorder(
    remote_spec: ExperimentSpec,
) -> None:
    model = remote_spec.matrix.models[0]
    provider = ProviderTarget(
        id="provider",
        model=model.repo,
        routing=ExplicitProviderRoute(provider="fireworks-ai"),
        limits=ProviderLimits(max_concurrent_requests=2),
    )
    spec = profiled_provider_spec(
        remote_spec.model_copy(
            update={
                "matrix": remote_spec.matrix.model_copy(
                    update={"deployments": [provider]}
                )
            }
        )
    )
    command = build_profile_submit_command(
        build_profile_plan(
            spec,
            profile_id="provider-profile",
            candidate_concurrency=[1, 2],
            max_spend_usd="10.00",
            profile_timeout_seconds=3600,
        ),
        input_dir="hf://buckets/input",
        bucket="example-org/results",
    )

    expose = command.index("--expose")
    assert command[expose : expose + 2] == ["--expose", "8000"]


def test_judged_profile_submit_command_exposes_judge_recorder(
    remote_spec: ExperimentSpec,
) -> None:
    raw = remote_spec.model_dump(mode="python")
    raw["benchmark"]["judge"] = {
        "api_url": "https://router.huggingface.co/v1/chat/completions",
        "model": "deepseek-ai/DeepSeek-V3.2",
    }
    spec = ExperimentSpec.model_validate(raw)

    command = build_profile_submit_command(
        plan(spec), input_dir="hf://buckets/input", bucket="example-org/results"
    )

    exposed = [
        command[index + 1]
        for index, argument in enumerate(command)
        if argument == "--expose"
    ]
    assert exposed == ["8001"]


def test_profile_judge_transport_uses_locked_direct_judge(
    remote_spec: ExperimentSpec,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    raw = remote_spec.model_dump(mode="python")
    raw["benchmark"]["judge"] = {
        "api_url": "https://api.openai.com/v1/chat/completions",
        "api_key_secret_name": "OPENAI_API_KEY",
        "model": "gpt-5.6-luna",
        "reasoning_effort": "xhigh",
        "strip_temperature": True,
    }
    lock = build_execution_lock(ExperimentSpec.model_validate(raw))
    monkeypatch.setenv("OPENAI_API_KEY", "openai-test-secret")
    captured: dict[str, object] = {}

    class Recorder:
        def __init__(self, **kwargs: object) -> None:
            captured.update(kwargs)

        def start(self, *, port: int) -> None:
            captured["port"] = port

        def close(self) -> None:
            captured["closed"] = True

    class Transport:
        def attach_judge_recorder(self, recorder: object, base_url: str) -> None:
            captured["attached"] = (recorder, base_url)

        def detach_judge_recorder(self) -> None:
            captured["detached"] = True

    monkeypatch.setattr("harbor_hf.profile_worker.JudgeEvidenceRecorder", Recorder)
    monkeypatch.setattr(
        "harbor_hf.profile_worker.job_ingress_base_url",
        lambda _port: "https://profile-job--8001.hf.jobs",
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker.wait_ready",
        lambda base_url, token, deadline: captured.update(
            readiness=(base_url, token, deadline)
        ),
    )
    transport = Transport()

    with _profile_judge_transport(
        lock, cast(Any, transport), "hf-ingress-secret", 123.0
    ) as selected:
        assert selected is transport

    assert captured["token"] == "openai-test-secret"
    assert captured["upstream_url"] == "https://api.openai.com/v1/chat/completions"
    assert captured["reasoning_effort"] == "xhigh"
    assert captured["strip_temperature"] is True
    assert captured["deadline"] == 123.0
    assert captured["readiness"] == (
        "https://profile-job--8001.hf.jobs",
        "hf-ingress-secret",
        123.0,
    )
    assert captured["detached"] is True
    assert captured["closed"] is True


def test_provider_profile_uses_distinct_tasks_at_maximum_concurrency(
    remote_spec: ExperimentSpec,
) -> None:
    model = remote_spec.matrix.models[0]
    provider = ProviderTarget(
        id="provider",
        model=model.repo,
        routing=ExplicitProviderRoute(provider="fireworks-ai"),
        limits=ProviderLimits(max_concurrent_requests=16),
    )
    spec = profiled_provider_spec(
        remote_spec.model_copy(
            update={
                "matrix": remote_spec.matrix.model_copy(
                    update={"deployments": [provider]}
                )
            }
        )
    )

    resolved = build_profile_plan(
        spec,
        profile_id="provider-profile",
        candidate_concurrency=[1, 2, 4, 8, 16],
        max_spend_usd="25",
        profile_timeout_seconds=5400,
    )

    assert resolved.workload.sample_task_count == 32
    low_tasks, low_attempts = _point_workload(resolved, 1)
    high_tasks, high_attempts = _point_workload(resolved, 16)
    assert low_tasks == high_tasks
    assert len(low_tasks) == 32
    assert low_attempts == high_attempts == 1


def test_provider_profile_uses_explicit_compatible_task_cohort(
    remote_spec: ExperimentSpec,
) -> None:
    model = remote_spec.matrix.models[0]
    provider = ProviderTarget(
        id="provider",
        model=model.repo,
        routing=ExplicitProviderRoute(provider="fireworks-ai"),
        limits=ProviderLimits(max_concurrent_requests=2),
    )
    spec = profiled_provider_spec(
        remote_spec.model_copy(
            update={
                "matrix": remote_spec.matrix.model_copy(
                    update={"deployments": [provider]}
                ),
            }
        )
    )
    task_digests = spec.benchmark.task_digests
    compatible = [
        "provider-task-00",
        "provider-task-02",
        "provider-task-04",
        "provider-task-05",
        "provider-task-06",
        "provider-task-07",
        "provider-task-08",
        "provider-task-09",
    ]

    resolved = build_profile_plan(
        spec,
        profile_id="compatible-provider-profile",
        candidate_concurrency=[1, 2],
        max_spend_usd="25",
        profile_timeout_seconds=5400,
        sample_task_names=compatible,
    )

    assert resolved.workload.sample_task_names == compatible
    assert resolved.identity.sample_task_names == compatible
    assert _point_workload(resolved, 2) == (
        {task: task_digests[task] for task in compatible},
        1,
    )


def test_profile_without_endpoint_gets_deterministic_managed_binding(
    remote_spec: ExperimentSpec,
) -> None:
    deployment = remote_spec.matrix.deployments[0].model_copy(update={"endpoint": None})
    spec = remote_spec.model_copy(
        update={
            "matrix": remote_spec.matrix.model_copy(
                update={"deployments": [deployment]}
            )
        }
    )
    resolved = plan(spec)

    first, desired = bind_profile_target(resolved)
    second, repeated = bind_profile_target(resolved)

    assert desired is not None
    assert desired == repeated
    assert first == second
    bound = first.matrix.deployments[0]
    assert isinstance(bound, DeploymentProfile)
    assert bound.endpoint is not None
    assert bound.endpoint.name == desired.identity.name
    command = build_profile_submit_command(
        resolved, input_dir="hf://buckets/input", bucket="example-org/results"
    )
    assert "harbor-hf-endpoint=" in " ".join(command)


def test_profile_submission_initializes_coordination_storage(
    remote_spec: ExperimentSpec,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls: list[str] = []

    class Runner:
        def run_text(self, command: list[str]) -> str:
            del command
            return "6a5a937cbee6ee1cf4ecded4"

    monkeypatch.setenv("GITHUB_TOKEN", "github-test")
    monkeypatch.setattr(
        "harbor_hf.profile_submission.ensure_private_coordination_repository",
        lambda *_args, **_kwargs: calls.append("coordination") or "coordination",
    )
    monkeypatch.setattr(
        "harbor_hf.profile_submission.ensure_private_job_input_bucket",
        lambda *_args, **_kwargs: calls.append("input") or "example-org/jobs-artifacts",
    )
    monkeypatch.setattr(
        "harbor_hf.profile_submission.require_private_bucket",
        lambda *_args, **_kwargs: calls.append("output"),
    )
    monkeypatch.setattr(
        "harbor_hf.profile_submission.stage_job_input",
        lambda *_args, **_kwargs: "hf://buckets/example-org/jobs-artifacts/input",
    )

    submission = submit_profile(
        plan(remote_spec),
        runner=Runner(),
        bucket_api=cast(Any, object()),
    )

    assert submission.job_id == "6a5a937cbee6ee1cf4ecded4"
    assert calls == ["coordination", "input", "output"]


class FakeApi:
    def model_info(self, repo: str, **kwargs: object) -> object:
        del kwargs
        return SimpleNamespace(sha="a" * 40, inference_provider_mapping={})

    def bucket_info(self, bucket_id: str) -> object:
        del bucket_id
        return SimpleNamespace(private=True)

    def list_repo_files(self, *_args: object, **_kwargs: object) -> list[str]:
        return []


class FakeProviderApi(FakeApi):
    def model_info(self, repo: str, **kwargs: object) -> object:
        del repo, kwargs
        return SimpleNamespace(
            sha="a" * 40,
            inference_provider_mapping=[
                SimpleNamespace(provider="fireworks-ai", status="live")
            ],
        )


def test_endpoint_preflight_reports_quota_and_cost(remote_spec: ExperimentSpec) -> None:
    spec = profiled_spec(remote_spec)
    model = spec.matrix.models[0].model_copy(update={"revision": "a" * 40})
    spec = spec.model_copy(
        update={
            "matrix": spec.matrix.model_copy(update={"models": [model]}),
        }
    )
    resolved = build_profile_plan(
        spec,
        profile_id="profile-one",
        candidate_concurrency=[1, 2],
        max_spend_usd="6.00",
        profile_timeout_seconds=3600,
    )
    deployment = resolved.deployment
    assert isinstance(deployment, DeploymentProfile)
    response = {
        "vendors": [
            {
                "name": "aws",
                "regions": [
                    {
                        "name": deployment.region.removeprefix("aws-"),
                        "computes": [
                            {
                                "instanceType": deployment.hardware,
                                "numAccelerators": deployment.accelerator_count,
                                "status": "available",
                                "pricePerHour": 5.0,
                                "quota": {
                                    "maxAccelerators": 2,
                                    "usedAccelerators": 0,
                                },
                            }
                        ],
                    }
                ],
            }
        ]
    }
    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda _request: httpx.Response(200, json=response)
        )
    )

    report = preflight_profile_plan(
        resolved, api=FakeApi(), client=client, token="hf_test"
    )

    assert report.available_accelerators == 2
    assert report.estimated_cost_usd == Decimal(65) / Decimal(12)

    rejected = build_profile_plan(
        spec,
        profile_id="profile-one",
        candidate_concurrency=[1, 2],
        max_spend_usd="5.00",
        profile_timeout_seconds=3600,
    )
    with pytest.raises(ValueError, match="exceeds spend cap"):
        preflight_profile_plan(rejected, api=FakeApi(), client=client, token="hf_test")


def test_endpoint_preflight_accounts_for_maximum_replicas(
    remote_spec: ExperimentSpec,
) -> None:
    spec = profiled_spec(remote_spec)
    model = spec.matrix.models[0].model_copy(update={"revision": "a" * 40})
    deployment = spec.matrix.deployments[0]
    assert isinstance(deployment, DeploymentProfile)
    deployment = deployment.model_copy(
        update={"parameters": deployment.parameters | {"max_replicas": 2}}
    )
    spec = spec.model_copy(
        update={
            "matrix": spec.matrix.model_copy(
                update={"models": [model], "deployments": [deployment]}
            )
        }
    )
    resolved = build_profile_plan(
        spec,
        profile_id="profile-one",
        candidate_concurrency=[1, 2],
        max_spend_usd="11.00",
        profile_timeout_seconds=3600,
    )
    response = {
        "vendors": [
            {
                "name": "aws",
                "regions": [
                    {
                        "name": deployment.region.removeprefix("aws-"),
                        "computes": [
                            {
                                "instanceType": deployment.hardware,
                                "numAccelerators": deployment.accelerator_count,
                                "status": "available",
                                "pricePerHour": 5.0,
                                "quota": {
                                    "maxAccelerators": 2,
                                    "usedAccelerators": 0,
                                },
                            }
                        ],
                    }
                ],
            }
        ]
    }
    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda _request: httpx.Response(200, json=response)
        )
    )

    report = preflight_profile_plan(
        resolved, api=FakeApi(), client=client, token="hf_test"
    )

    assert report.required_accelerators == 2
    assert report.estimated_cost_usd == Decimal(65) / Decimal(6)


def test_managed_endpoint_preflight_uses_remote_namespace(
    remote_spec: ExperimentSpec,
) -> None:
    deployment = remote_spec.matrix.deployments[0].model_copy(update={"endpoint": None})
    assert isinstance(deployment, DeploymentProfile)
    model = remote_spec.matrix.models[0].model_copy(update={"revision": "a" * 40})
    spec = profiled_spec(
        remote_spec.model_copy(
            update={
                "matrix": remote_spec.matrix.model_copy(
                    update={"models": [model], "deployments": [deployment]}
                )
            }
        )
    )
    resolved = build_profile_plan(
        spec,
        profile_id="profile-one",
        candidate_concurrency=[1, 2],
        max_spend_usd="6.00",
        profile_timeout_seconds=3600,
    )
    response = {
        "vendors": [
            {
                "name": "aws",
                "regions": [
                    {
                        "name": deployment.region.removeprefix("aws-"),
                        "computes": [
                            {
                                "instanceType": deployment.hardware,
                                "numAccelerators": deployment.accelerator_count,
                                "status": "available",
                                "pricePerHour": 5.0,
                                "quota": {
                                    "maxAccelerators": 2,
                                    "usedAccelerators": 0,
                                },
                            }
                        ],
                    }
                ],
            }
        ]
    }
    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda _request: httpx.Response(200, json=response)
        )
    )

    report = preflight_profile_plan(
        resolved, api=FakeApi(), client=client, token="hf_test"
    )

    assert report.target_kind == "inference-endpoint"
    assert report.available_accelerators == 2


def test_endpoint_preflight_rejects_missing_repository_artifact(
    remote_spec: ExperimentSpec,
) -> None:
    deployment = remote_spec.matrix.deployments[0]
    assert isinstance(deployment, DeploymentProfile)
    deployment = deployment.model_copy(
        update={
            "engine": deployment.engine.model_copy(
                update={"arguments": ["-m", "/repository/missing.gguf"]}
            )
        }
    )
    model = remote_spec.matrix.models[0].model_copy(update={"revision": "a" * 40})
    spec = profiled_spec(
        remote_spec.model_copy(
            update={
                "matrix": remote_spec.matrix.model_copy(
                    update={"models": [model], "deployments": [deployment]}
                )
            }
        )
    )
    resolved = build_profile_plan(
        spec,
        profile_id="profile-one",
        candidate_concurrency=[1, 2],
        max_spend_usd="5.00",
        profile_timeout_seconds=3600,
    )

    with pytest.raises(ValueError, match="missing model artifacts: missing.gguf"):
        preflight_profile_plan(resolved, api=FakeApi(), token="hf_test")


def test_provider_preflight_requires_bounded_full_profile_estimate(
    remote_spec: ExperimentSpec,
) -> None:
    spec = profiled_provider_spec(remote_spec)
    model = spec.matrix.models[0].model_copy(update={"revision": "a" * 40})
    provider = ProviderTarget(
        id="provider",
        model=model.repo,
        routing=ExplicitProviderRoute(provider="fireworks-ai"),
        limits=ProviderLimits(max_concurrent_requests=2),
    )
    spec = spec.model_copy(
        update={
            "matrix": spec.matrix.model_copy(
                update={"models": [model], "deployments": [provider]}
            )
        }
    )
    resolved = build_profile_plan(
        spec,
        profile_id="profile-one",
        candidate_concurrency=[1, 2],
        max_spend_usd="5.00",
        profile_timeout_seconds=3600,
    )

    with pytest.raises(ValueError, match="bounded full-profile cost estimate"):
        preflight_profile_plan(resolved, api=FakeProviderApi(), token="hf_test")


def test_provider_preflight_enforces_profile_spend_cap(
    remote_spec: ExperimentSpec,
) -> None:
    spec = profiled_provider_spec(remote_spec)
    model = spec.matrix.models[0].model_copy(update={"revision": "a" * 40})
    provider = ProviderTarget(
        id="provider",
        model=model.repo,
        routing=ExplicitProviderRoute(provider="fireworks-ai"),
        limits=ProviderLimits(
            max_concurrent_requests=2,
            max_spend_usd=Decimal("10"),
            estimated_wave_cost_usd=Decimal("6"),
        ),
    )
    spec = spec.model_copy(
        update={
            "matrix": spec.matrix.model_copy(
                update={"models": [model], "deployments": [provider]}
            )
        }
    )
    resolved = build_profile_plan(
        spec,
        profile_id="profile-one",
        candidate_concurrency=[1, 2],
        max_spend_usd="5.00",
        profile_timeout_seconds=3600,
        estimated_profile_cost_usd="6",
    )

    with pytest.raises(ValueError, match="exceeds spend cap"):
        preflight_profile_plan(resolved, api=FakeProviderApi(), token="hf_test")


def test_provider_preflight_uses_full_profile_not_wave_estimate(
    remote_spec: ExperimentSpec,
) -> None:
    spec = profiled_provider_spec(remote_spec)
    model = spec.matrix.models[0].model_copy(update={"revision": "a" * 40})
    provider = ProviderTarget(
        id="provider",
        model=model.repo,
        routing=ExplicitProviderRoute(provider="fireworks-ai"),
        limits=ProviderLimits(
            max_concurrent_requests=2,
            max_spend_usd=Decimal("10"),
            estimated_wave_cost_usd=Decimal("1"),
        ),
    )
    spec = spec.model_copy(
        update={
            "matrix": spec.matrix.model_copy(
                update={"models": [model], "deployments": [provider]}
            )
        }
    )
    resolved = build_profile_plan(
        spec,
        profile_id="profile-one",
        candidate_concurrency=[1, 2],
        max_spend_usd="10",
        profile_timeout_seconds=3600,
        estimated_profile_cost_usd="6",
    )

    report = preflight_profile_plan(resolved, api=FakeProviderApi(), token="hf_test")

    assert report.estimated_cost_usd == Decimal("6")


def test_profile_worker_rebuilds_provider_cost_estimate(
    remote_spec: ExperimentSpec,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    model = remote_spec.matrix.models[0]
    provider = ProviderTarget(
        id="provider",
        model=model.repo,
        routing=ExplicitProviderRoute(provider="fireworks-ai"),
        limits=ProviderLimits(
            max_concurrent_requests=2,
            max_spend_usd=Decimal("10"),
            estimated_wave_cost_usd=Decimal("1"),
        ),
    )
    spec = profiled_provider_spec(
        remote_spec.model_copy(
            update={
                "matrix": remote_spec.matrix.model_copy(
                    update={"deployments": [provider]}
                )
            }
        )
    )
    resolved = build_profile_plan(
        spec,
        profile_id="profile-one",
        candidate_concurrency=[1, 2],
        max_spend_usd="10",
        profile_timeout_seconds=3600,
        estimated_profile_cost_usd="6",
    )
    plan_path = tmp_path / "plan.json"
    plan_path.write_text(resolved.model_dump_json(), encoding="utf-8")

    def rebuild(*_args: object, **kwargs: object) -> ProfilePlan:
        assert kwargs["estimated_profile_cost_usd"] == "6"
        raise RuntimeError("rebuild observed")

    monkeypatch.setattr("harbor_hf.profile_worker.build_profile_plan", rebuild)

    with pytest.raises(RuntimeError, match="rebuild observed"):
        run_profile_worker(plan_path, tmp_path / "output")


def _profile_test_deployment(
    remote_spec: ExperimentSpec, managed_endpoint: bool
) -> DeploymentProfile:
    deployment = remote_spec.matrix.deployments[0]
    assert isinstance(deployment, DeploymentProfile)
    return (
        deployment.model_copy(update={"endpoint": None})
        if managed_endpoint
        else deployment
    )


def _assert_profile_endpoint_order(calls: list[str], managed_endpoint: bool) -> None:
    assert calls.index("watchdog") < calls.index("resume")
    if managed_endpoint:
        assert calls.index("watchdog") < calls.index("provision")
        assert calls.index("provision") < calls.index("resume")
    else:
        assert "provision" not in calls
        assert calls.index("describe") < calls.index("watchdog")
        assert calls.index("validated") < calls.index("watchdog")
        assert calls.index("baseline-paused") < calls.index("watchdog")
    assert calls[-1] == "pause"


def _profile_smoke(calls: list[str], smoke_fails: bool) -> object:
    def smoke(*_args: object, **_kwargs: object) -> None:
        calls.append("smoke")
        if smoke_fails:
            raise ProfileWorkerError("smoke failed")

    return smoke


def _run_profile_with_cleanup_expectation(
    plan_path: Path,
    output_root: Path,
    *,
    smoke_fails: bool,
    cleanup_fails: bool,
) -> None:
    if not cleanup_fails:
        run_expected_profile(plan_path, output_root, smoke_fails=smoke_fails)
        return
    with pytest.raises(ProfileCleanupUnverified, match="remains nonterminal"):
        run_profile_worker(plan_path, output_root)
    destination = output_root / "serving-profiles/profile-one"
    assert (destination / "failure.json").is_file()
    assert (destination / "checksums.json").is_file()
    assert not (destination / "_FAILED").exists()
    assert not (destination / "_SELECTED").exists()


@pytest.mark.parametrize("cleanup_fails", [False, True])
@pytest.mark.parametrize("smoke_fails", [False, True])
@pytest.mark.parametrize("managed_endpoint", [False, True])
def test_profile_worker_always_pauses_owned_endpoint(
    remote_spec: ExperimentSpec,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    smoke_fails: bool,
    managed_endpoint: bool,
    cleanup_fails: bool,
) -> None:
    deployment = _profile_test_deployment(remote_spec, managed_endpoint)
    spec = remote_spec.model_copy(
        update={
            "matrix": remote_spec.matrix.model_copy(
                update={"deployments": [deployment]}
            )
        }
    )
    resolved = plan(spec)
    plan_path = tmp_path / "plan.json"
    plan_path.write_text(resolved.model_dump_json(), encoding="utf-8")
    calls: list[str] = []

    class Manager:
        def __init__(self, *_args: object, **_kwargs: object) -> None:
            calls.append("manager")

        def describe(self) -> dict[str, object]:
            calls.append("describe")
            return {}

        def pause_and_verify(self) -> dict[str, object]:
            calls.append("pause")
            if cleanup_fails:
                raise RuntimeError("pause failed")
            return {}

    monkeypatch.setenv("HF_TOKEN", "hf_test")
    monkeypatch.setattr(
        "harbor_hf.profile_worker.prepare_locked_source",
        lambda *_args, **_kwargs: calls.append("source"),
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker.HuggingFaceEndpointAdapter",
        lambda **_kwargs: object(),
    )

    class Provisioner:
        def __init__(self, _adapter: object) -> None:
            calls.append("provisioner")
            self.created = False

        def inspect(self, *_args: object, **_kwargs: object) -> object | None:
            calls.append("provision-inspect")
            return {} if self.created else None

        def create_or_adopt(self, *_args: object, **_kwargs: object) -> None:
            calls.append("provision")
            self.created = True

    monkeypatch.setattr("harbor_hf.profile_worker.EndpointProvisioner", Provisioner)
    monkeypatch.setattr("harbor_hf.profile_worker.EndpointManager", Manager)
    monkeypatch.setattr(
        "harbor_hf.profile_worker.validate_endpoint_model",
        lambda *_args: calls.append("validated"),
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker.require_paused_endpoint",
        lambda *_args: calls.append("baseline-paused"),
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker.launch_cleanup_watchdog",
        lambda *_args: calls.append("watchdog"),
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker.resume_and_probe_endpoint",
        lambda *_args, **_kwargs: calls.append("resume") or "https://endpoint.test",
    )

    monkeypatch.setattr(
        "harbor_hf.profile_worker._verify_smoke",
        _profile_smoke(calls, smoke_fails),
    )
    monkeypatch.setattr(
        "harbor_hf.profile_worker._run_ladder",
        lambda *_args, **_kwargs: [point(1, 10.0)],
    )

    _run_profile_with_cleanup_expectation(
        plan_path,
        tmp_path / "output",
        smoke_fails=smoke_fails,
        cleanup_fails=cleanup_fails,
    )

    _assert_profile_endpoint_order(calls, managed_endpoint)