6krrt/tests/test_admin_config.py

"""Tests for the /admin/api persisted-config endpoints.

``admin.py`` exposes ``GET /admin/api/config`` (the allowlisted values with
provenance, ``{value, source}``) and ``POST /admin/api/config/{key}`` (persist
one allowlisted value to ``config.local.yaml`` with comment-preserving
ruamel.yaml round-trip, a backup copy, and whole-config validation of the
merged base + overlay via ``RouterConfig`` BEFORE anything touches disk).

Each test builds its own isolated router against a temp copy of config.yaml by
passing ``base_dir`` (a temp dir) to ``build_router`` — the real repo
``config.yaml`` is never written. The overlay endpoints create and mutate
``<tmp_dir>/config/config.local.yaml``. It mounts the router on a fresh FastAPI
TestClient at ``prefix="/admin"``.
"""

from __future__ import annotations

import re
import shutil
import sqlite3
import subprocess
import threading
from pathlib import Path

import pytest
import yaml
from pydantic import ValidationError
from fastapi import FastAPI
from starlette.testclient import TestClient

from admin import (
    _CONFIG_ALLOWLIST,
    _persist_config_block,
    _persist_config_value,
    _persist_to,
    build_router,
)
from config import load_config

ROOT = Path(__file__).resolve().parent.parent

_SENTINEL = "# SENTINEL_PRESERVED_12345"


def _insert_sentinel(config_yaml: Path) -> None:
    """Prepend a unique marker comment just above the ``logging:`` map."""
    text = config_yaml.read_text()
    assert "logging:" in text
    config_yaml.write_text(text.replace("logging:", f"{_SENTINEL}\nlogging:", 1))


def _make_db(tmp_path: Path) -> sqlite3.Connection:
    conn = sqlite3.connect(str(tmp_path / "admin.db"))
    conn.row_factory = sqlite3.Row
    return conn


@pytest.fixture
def client(tmp_path, monkeypatch):
    """A TestClient for an isolated admin router over a temp config.yaml copy.

    ``base_dir`` is tmp_path, so the ``/admin/api/config`` endpoints read and
    write ``tmp_path/config/config.yaml`` — never the repo's copy.
    """
    (tmp_path / "config").mkdir(parents=True, exist_ok=True)
    config_yaml = tmp_path / "config" / "config.yaml"
    shutil.copyfile(ROOT / "config" / "config.yaml", config_yaml)

    schema_sql = (ROOT / "config" / "schema.sql").read_text()
    conn = _make_db(tmp_path)
    conn.executescript(schema_sql)
    conn.close()

    def _db_factory() -> sqlite3.Connection:
        return _make_db(tmp_path)

    cfg = load_config(str(config_yaml))

    router = build_router(cfg, _db_factory, base_dir=str(tmp_path))
    app = FastAPI()
    app.include_router(router, prefix="/admin")
    return TestClient(app), config_yaml


def test_config_GET_returns_allowlisted_values(client):
    """GET /admin/api/config returns every allowlisted dotted key."""
    tc, _ = client
    resp = tc.get("/admin/api/config")
    assert resp.status_code == 200
    body = resp.json()
    for key in (
        "logging.level",
        "objective.quality_tolerance",
        "objective.max_energy_per_request",
        "objective.plan_kwh_per_period",
        "circuit_breaker.enabled",
        "session_cache.enabled",
        "session_cache.staleness_seconds",
        "verification.local_llm_enabled",
        "local_vision.enabled",
        "objective.incumbent_cache_pricing",
        "objective.incumbent_challenger_cache_rate",
        "pinch.enabled",
        "pinch.prefix_probe",
        "pinch.relevance.enabled",
        "routing.default_flex_preference",
    ):
        assert key in body
    assert body["logging.level"] == {"value": "info", "source": "base"}
    assert body["objective.quality_tolerance"] == {"value": 0.10, "source": "base"}
    assert body["routing.default_flex_preference"] == {
        "value": "auto",
        "source": "base",
    }


def test_config_POST_preserves_comments_and_changes_value(client):
    """A valid write keeps the file's comments AND updates the value."""
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")
    _insert_sentinel(config_yaml)

    resp = tc.post("/admin/api/config/logging.level", json={"value": "warning"})
    assert resp.status_code == 200
    body = resp.json()
    assert body["key"] == "logging.level"
    assert body["value"] == "warning"
    assert "restart is required" in body["message"]

    # The base file is untouched; the overlay now carries the changed value.
    base_text = config_yaml.read_text()
    assert _SENTINEL in base_text
    assert re.search(r"^\s*level:\s*info\s*$", base_text, re.MULTILINE) is not None
    local_text = local_yaml.read_text()
    assert re.search(
        r"^\s*level:\s*warning\s*$", local_text, re.MULTILINE
    ) is not None


def test_config_pinch_prefix_probe_round_trips_through_the_real_validator(
    client, tmp_path
):
    """Read true, write false, read back false, and the overlay still loads.

    The round-trip through ``load_config`` is the point, not decoration. The
    ``classifier.confidence_threshold`` incident was an admin control writing a
    value (a raw ``80`` meaning 80%) the config loader would later reject or
    silently misread, and it was caught before the restart only by luck. A
    control that writes something the loader will not take back is worse than
    no control, so this asserts the merged base + overlay parses and that the
    value survives as a real ``False``, not the string ``"false"``.
    """
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")

    before = tc.get("/admin/api/config").json()
    assert before["pinch.prefix_probe"] == {"value": True, "source": "base"}

    resp = tc.post("/admin/api/config/pinch.prefix_probe", json={"value": False})
    assert resp.status_code == 200
    assert resp.json()["value"] is False

    after = tc.get("/admin/api/config").json()
    assert after["pinch.prefix_probe"] == {"value": False, "source": "overlay"}

    # The base file never carries an admin write.
    assert yaml.safe_load(config_yaml.read_text())["pinch"]["prefix_probe"] is True
    assert yaml.safe_load(local_yaml.read_text())["pinch"]["prefix_probe"] is False

    # And the merged result parses through RouterConfig, which is what the
    # service does on its next boot.
    loaded = load_config(str(config_yaml), include_overlay=True)
    assert loaded.pinch.prefix_probe is False
    # Nothing else in the pinch block was disturbed by the overlay merge.
    assert loaded.pinch.enabled is True
    assert loaded.pinch.relevance.enabled is True


def test_config_pinch_prefix_probe_is_allowlisted_at_the_right_path(client):
    """The allowlist entry points at ``pinch.prefix_probe``, not a sibling.

    A wrong path here would write a key ``RouterConfig`` forbids (StrictModel
    sets ``extra="forbid"``), so the whole-config validation would 422 every
    save -- but only for this one key, which is exactly the sort of thing a
    green test run hides.
    """
    assert _CONFIG_ALLOWLIST["pinch.prefix_probe"] == ("pinch", "prefix_probe")


def test_config_local_vision_enabled_round_trips_through_the_real_validator(client):
    """local_vision.enabled persists to the temp overlay and reloads as a bool.

    Same reasoning as the ``pinch.prefix_probe`` round-trip: the value that
    reaches disk has to be one the config loader will take back on the next
    boot. ``local_vision.enabled`` is read per request in the dispatcher, so a
    write that lands as the string ``"false"`` would silently keep the fallback
    on. Assert a real ``False`` survives the merged reload.
    """
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")

    before = tc.get("/admin/api/config").json()
    assert before["local_vision.enabled"] == {"value": True, "source": "base"}

    resp = tc.post("/admin/api/config/local_vision.enabled", json={"value": False})
    assert resp.status_code == 200
    assert resp.json()["value"] is False

    after = tc.get("/admin/api/config").json()
    assert after["local_vision.enabled"] == {"value": False, "source": "overlay"}

    # The base file never carries an admin write.
    assert yaml.safe_load(config_yaml.read_text())["local_vision"]["enabled"] is True
    assert yaml.safe_load(local_yaml.read_text())["local_vision"]["enabled"] is False

    # And the merged result parses through RouterConfig, which is what the
    # service does on its next boot.
    loaded = load_config(str(config_yaml), include_overlay=True)
    assert loaded.local_vision.enabled is False
    # The rest of the local_vision block is untouched by the overlay merge.
    base_vision = yaml.safe_load(config_yaml.read_text())["local_vision"]
    assert loaded.local_vision.model == base_vision["model"]
    assert loaded.local_vision.base_url == base_vision["base_url"]


def test_config_incumbent_knobs_round_trip_through_the_real_validator(client):
    """Both Wave 2 knobs survive a write as genuine Python types.

    Same reasoning as the ``pinch.prefix_probe`` round-trip: the value that
    reaches disk has to be one the config loader will take back on the next
    boot. The extra assertion here is the TYPE -- a real ``bool`` and a real
    ``float``, not the strings ``"true"`` and ``"0.0"``, because the dial is
    read straight into an arithmetic expression in ``routing.estimated_cost``
    and a string would not fail until a request was already in flight.
    """
    tc, config_yaml = client

    before = tc.get("/admin/api/config").json()
    assert before["objective.incumbent_cache_pricing"] == {
        "value": False,
        "source": "base",
    }
    # Shipped blank in config.yaml, which YAML reads as null.
    assert before["objective.incumbent_challenger_cache_rate"] == {
        "value": None,
        "source": "base",
    }

    assert (
        tc.post(
            "/admin/api/config/objective.incumbent_cache_pricing",
            json={"value": True},
        ).status_code
        == 200
    )
    assert (
        tc.post(
            "/admin/api/config/objective.incumbent_challenger_cache_rate",
            json={"value": 0.0},
        ).status_code
        == 200
    )

    after = tc.get("/admin/api/config").json()
    assert after["objective.incumbent_cache_pricing"] == {
        "value": True,
        "source": "overlay",
    }
    assert after["objective.incumbent_challenger_cache_rate"] == {
        "value": 0.0,
        "source": "overlay",
    }

    # The base file never carries an admin write.
    base = yaml.safe_load(config_yaml.read_text())["objective"]
    assert base["incumbent_cache_pricing"] is False
    assert base["incumbent_challenger_cache_rate"] is None

    loaded = load_config(str(config_yaml), include_overlay=True)
    assert loaded.objective.incumbent_cache_pricing is True
    rate = loaded.objective.incumbent_challenger_cache_rate
    assert isinstance(rate, float) and not isinstance(rate, bool)
    assert rate == 0.0
    # Nothing else in the objective block moved on the overlay merge.
    assert loaded.objective.quality_tolerance == 0.10
    assert loaded.objective.incumbent_rate_min_observations == 25


def test_config_blank_challenger_dial_persists_as_null_not_zero(client):
    """Clearing the field writes ``null``, which the loader reads as NEUTRAL.

    ``null`` and ``0.0`` are opposite ends of this dial: null follows
    ``assumed_cache_rate`` and costs a challenger nothing, 0.0 prices every
    challenger as a fully cold prompt. A UI that sent 0.0 for an empty input
    would hand an operator the maximum penalty at the exact moment they were
    trying to switch the feature off, so the null has to survive all the way
    to the file AND be resolved by the loader rather than by the UI.
    """
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")

    # Start at full penalty so a neutral write has something to undo.
    tc.post(
        "/admin/api/config/objective.incumbent_challenger_cache_rate",
        json={"value": 0.0},
    )
    assert (
        yaml.safe_load(local_yaml.read_text())["objective"][
            "incumbent_challenger_cache_rate"
        ]
        == 0.0
    )

    resp = tc.post(
        "/admin/api/config/objective.incumbent_challenger_cache_rate",
        json={"value": None},
    )
    assert resp.status_code == 200

    overlay = yaml.safe_load(local_yaml.read_text())["objective"]
    assert "incumbent_challenger_cache_rate" in overlay
    assert overlay["incumbent_challenger_cache_rate"] is None

    body = tc.get("/admin/api/config").json()
    assert body["objective.incumbent_challenger_cache_rate"] == {
        "value": None,
        "source": "overlay",
    }

    loaded = load_config(str(config_yaml), include_overlay=True)
    assert loaded.objective.incumbent_challenger_cache_rate == (
        loaded.objective.assumed_cache_rate
    )
    assert loaded.objective.incumbent_challenger_cache_rate != 0.0


@pytest.mark.parametrize("bad", [-0.1, 1.5, 80])
def test_config_challenger_dial_refuses_values_outside_zero_to_one(client, bad):
    """The field is a rate in [0, 1], never a percentage.

    ``80`` is the shape of the ``classifier.confidence_threshold`` incident:
    the admin UI saved a raw 80 meaning 80% for a field the loader wanted as
    0.0-1.0, and every reading would have been below threshold on the next
    restart. Here RouterConfig validates the MERGED config before a byte
    reaches disk, so the write is refused instead.
    """
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")

    resp = tc.post(
        "/admin/api/config/objective.incumbent_challenger_cache_rate",
        json={"value": bad},
    )
    assert resp.status_code == 422
    assert "must be in [0, 1]" in resp.json()["detail"]
    assert not local_yaml.exists()


def test_config_incumbent_keys_are_allowlisted_at_the_right_paths():
    """A wrong path would write a key StrictModel forbids, 422ing every save."""
    assert _CONFIG_ALLOWLIST["objective.incumbent_cache_pricing"] == (
        "objective",
        "incumbent_cache_pricing",
    )
    assert _CONFIG_ALLOWLIST["objective.incumbent_challenger_cache_rate"] == (
        "objective",
        "incumbent_challenger_cache_rate",
    )


def test_config_staleness_window_round_trips_as_a_real_int(client):
    """The window survives a write as an ``int`` the loader takes straight back.

    The type assertion is the point, and it is the ``classifier.confidence_threshold``
    lesson in its other form: that incident wrote a raw ``80`` meaning 80% for a
    field the loader wanted as 0.0-1.0, and nothing caught it until someone
    looked. Here the units are seconds and the field is declared ``int``, so
    the UI must not scale it and the write must not float it — a ``1200.0`` on
    disk is a value ``load_config`` refuses, and a quietly-truncated ``20.5``
    is a window the operator never chose.
    """
    tc, config_yaml = client

    before = tc.get("/admin/api/config").json()
    assert before["session_cache.staleness_seconds"] == {"value": 1200, "source": "base"}

    assert (
        tc.post(
            "/admin/api/config/session_cache.staleness_seconds",
            json={"value": 5},
        ).status_code
        == 200
    )

    after = tc.get("/admin/api/config").json()
    assert after["session_cache.staleness_seconds"] == {"value": 5, "source": "overlay"}

    # The shipped default never moves; the admin write lands in the overlay.
    base = yaml.safe_load(config_yaml.read_text())["session_cache"]
    assert base["staleness_seconds"] == 1200

    loaded = load_config(str(config_yaml), include_overlay=True)
    seconds = loaded.session_cache.staleness_seconds
    assert isinstance(seconds, int) and not isinstance(seconds, bool)
    assert seconds == 5
    # The switch beside it did not move on the overlay merge.
    assert loaded.session_cache.enabled is True


@pytest.mark.parametrize("bad", [0, -1, 4, 7201, 10000, None])
def test_config_staleness_window_refuses_out_of_range_and_blank(client, bad):
    """Whole-config validation refuses before a byte reaches the overlay.

    ``0`` is in here on purpose. It looks like "stop reusing classifications"
    and is not: ``session_cache.put`` still writes and the classifier-failure
    cascade's ``stale_read`` ignores staleness entirely, so a 0-second window
    still replays a session's label whenever the classifier is down. The knob
    that actually stops reuse is ``session_cache.enabled``, which has its own
    control, so 0 is refused rather than quietly honoured — and ``None``
    (a cleared field) with it, since blank is not a setting either.
    """
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")

    resp = tc.post(
        "/admin/api/config/session_cache.staleness_seconds", json={"value": bad}
    )
    assert resp.status_code == 422
    assert not local_yaml.exists()
    assert (
        tc.get("/admin/api/config").json()["session_cache.staleness_seconds"]["value"]
        == 1200
    )


def test_config_staleness_window_is_allowlisted_at_the_right_path():
    """A wrong path would write a key StrictModel forbids, 422ing every save."""
    assert _CONFIG_ALLOWLIST["session_cache.staleness_seconds"] == (
        "session_cache",
        "staleness_seconds",
    )


def test_config_POST_rejects_non_allowlisted_key(client):
    """classifier.base_url (and any off-allowlist key) is refused with 403."""
    tc, _ = client
    resp = tc.post("/admin/api/config/classifier.base_url", json={"value": "http://x"})
    assert resp.status_code == 403


def test_config_POST_invalid_value_leaves_file_unchanged(client):
    """quality_tolerance=1.5 (>1) fails RouterConfig validation -> 422, no write."""
    tc, config_yaml = client
    before_base = config_yaml.read_text()
    local_yaml = config_yaml.with_name("config.local.yaml")
    before_local = local_yaml.read_text() if local_yaml.exists() else ""

    resp = tc.post(
        "/admin/api/config/objective.quality_tolerance", json={"value": 1.5}
    )
    assert resp.status_code == 422

    assert config_yaml.read_text() == before_base
    assert (
        local_yaml.read_text() if local_yaml.exists() else ""
    ) == before_local


def test_config_GET_includes_routing_default_profile(client):
    """GET /admin/api/config exposes routing.default_profile with the repo default."""
    tc, _ = client
    resp = tc.get("/admin/api/config")
    assert resp.status_code == 200
    body = resp.json()
    assert "routing.default_profile" in body
    assert body["routing.default_profile"] == {
        "value": "default",
        "source": "base",
    }


def test_config_POST_default_profile_persists_valid_builtin(client):
    """POST routing.default_profile=batch persists and keeps comments."""
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")
    _insert_sentinel(config_yaml)

    resp = tc.post(
        "/admin/api/config/routing.default_profile", json={"value": "batch"}
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["key"] == "routing.default_profile"
    assert body["value"] == "batch"
    # This key is the one allowlisted setting that does NOT need a bounce: the
    # write path also applies it to the running cfg, because the router reads
    # cfg.routing.default_profile per request. Every other key here still
    # answers "restart is required", so the message is deliberately narrow --
    # see test_config_POST_other_keys_still_say_restart below.
    assert "no restart required" in body["message"]
    assert "restart is required" not in body["message"]

    # Base comments preserved; value lives in overlay.
    base_text = config_yaml.read_text()
    assert _SENTINEL in base_text
    local_text = local_yaml.read_text()
    assert re.search(
        r'^\s*default_profile:\s*["\']?batch["\']?\s*$', local_text, re.MULTILINE
    ) is not None


def test_config_POST_other_keys_still_say_restart(client):
    """The no-restart answer is scoped to routing.default_profile alone.

    Every other allowlisted key really does bind at import -- a provider's
    base_url, the log level -- so a blanket "no restart required" would be a
    lie that costs someone an afternoon. This is the guard against the
    narrow exception widening by accident.
    """
    tc, _config_yaml = client
    resp = tc.post("/admin/api/config/logging.level", json={"value": "DEBUG"})
    assert resp.status_code == 200
    assert "restart is required" in resp.json()["message"]


def test_config_POST_default_profile_rejects_unknown_and_leaves_overlay_unchanged(
    client,
):
    """Unknown default_profile returns 422 and does not touch config.local.yaml."""
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")
    before_base = config_yaml.read_text()
    before_local = local_yaml.read_text() if local_yaml.exists() else ""

    resp = tc.post(
        "/admin/api/config/routing.default_profile",
        json={"value": "nosuchprofile"},
    )
    assert resp.status_code == 422
    detail = resp.json()["detail"]
    assert "nosuchprofile" in detail
    assert "default" in detail or "batch" in detail

    assert config_yaml.read_text() == before_base
    assert (
        local_yaml.read_text() if local_yaml.exists() else ""
    ) == before_local


def test_config_POST_creates_overlay_backup_before_write(client, tmp_path):
    """A successful write produces a ``config.local.yaml.bak.<ts>`` backup copy."""
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")
    _insert_sentinel(config_yaml)

    resp = tc.post("/admin/api/config/circuit_breaker.enabled", json={"value": True})
    assert resp.status_code == 200

    backups = sorted(tmp_path.glob("config/config.local.yaml.bak.*"))
    assert len(backups) == 1
    # The backup captured the pre-write empty overlay (just the header).
    backup_text = backups[0].read_text()
    assert backup_text.strip() != ""
    # The current overlay carries the new value.
    local_text = local_yaml.read_text()
    assert yaml.safe_load(local_text)["circuit_breaker"]["enabled"] is True


def test_config_concurrent_writes_are_atomic_no_zero_byte_backups(tmp_path):
    """Concurrent writes never truncate config.yaml or leave 0-byte backups."""
    config_yaml = tmp_path / "config.yaml"
    shutil.copyfile(ROOT / "config" / "config.yaml", config_yaml)
    path = _CONFIG_ALLOWLIST["logging.level"]

    stop_reader = threading.Event()

    def reader() -> None:
        while not stop_reader.is_set():
            try:
                text = config_yaml.read_text()
            except FileNotFoundError:
                continue
            assert text.strip() != "", "config.yaml observed empty"
            assert "logging:" in text

    reader_thread = threading.Thread(target=reader)
    reader_thread.start()

    def writer(level: str) -> None:
        _persist_config_value(config_yaml, path, level)

    threads = [
        threading.Thread(target=writer, args=("warning",)),
        threading.Thread(target=writer, args=("error",)),
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    stop_reader.set()
    reader_thread.join()

    final = config_yaml.read_text()
    assert re.search(
        r"^\s*level:\s*(warning|error)\s*$", final, re.MULTILINE
    ) is not None

    backups = list(tmp_path.glob("config.yaml.bak.*"))
    assert backups, "expected at least one backup"
    for b in backups:
        assert b.stat().st_size > 0, f"zero-byte backup: {b}"
        assert b.read_text().strip() != ""


def test_profile_create_writes_overlay_and_leaves_base_unchanged(
    client, tmp_path
):
    """A profile CRUD write creates config.local.yaml and leaves config.yaml bytes unchanged."""
    tc, config_yaml = client
    _insert_sentinel(config_yaml)
    base_before = config_yaml.read_bytes()
    local_yaml = config_yaml.with_name("config.local.yaml")

    resp = tc.post(
        "/admin/api/profiles/",
        json={"name": "minit", "max_tier": 2},
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["profile"]["name"] == "minit"
    assert body["profile"]["definition"]["max_tier"] == 2
    assert body["profile"]["source"] == "overlay"

    assert config_yaml.read_bytes() == base_before
    assert local_yaml.exists()
    local_text = local_yaml.read_text()
    assert "profiles:" in local_text
    assert "minit:" in local_text
    assert "max_tier: 2" in local_text


def test_profile_create_creates_overlay_backup_before_write(client, tmp_path):
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")
    _insert_sentinel(config_yaml)

    resp = tc.post(
        "/admin/api/profiles/",
        json={"name": "minit", "max_tier": 2},
    )
    assert resp.status_code == 200

    backups = sorted(tmp_path.glob("config/config.local.yaml.bak.*"))
    assert len(backups) == 1
    backup_text = backups[0].read_text()
    assert "profiles:" not in backup_text
    assert local_yaml.exists()
    local_text = local_yaml.read_text()
    assert "profiles:" in local_text
    assert "minit:" in local_text


def test_profile_create_blocked_nested_path_returns_403(client):
    tc, _ = client
    resp = tc.post("/admin/api/config/profiles.foo", json={"value": {}})
    assert resp.status_code == 403


def test_profile_create_builtin_name_returns_403(client):
    tc, _ = client
    resp = tc.post("/admin/api/profiles/", json={"name": "batch", "max_tier": 2})
    assert resp.status_code == 403
    assert "built-in" in resp.json()["detail"].lower()


def test_persist_to_merged_validation_refuses_invalid_block(tmp_path):
    """Merged-config guard is enforced at the helper level, not the endpoint.

    ``admin_profile_create`` validates ``RoutingProfile(**profile_dict)`` first,
    so an endpoint-level collision with a built-in name is impossible.  This test
    directly exercises ``_persist_to`` to prove that when the merged base +
    overlay trigger an invalid RouterConfig, the overlay validation raises
    ``ValidationError`` and neither file is modified.
    """
    (tmp_path / "config").mkdir(parents=True, exist_ok=True)
    base_yaml = tmp_path / "config" / "config.yaml"
    shutil.copyfile(ROOT / "config" / "config.yaml", base_yaml)
    overlay_yaml = tmp_path / "config" / "config.local.yaml"

    base_before = base_yaml.read_bytes()
    overlay_before = overlay_yaml.read_bytes() if overlay_yaml.exists() else b""

    # ``default`` is a built-in profile name; merging an overlay profile named
    # ``default`` into a base with no such profile conflicts with RouterConfig's
    # builtins-vs-config validation.
    profile_dict = {"max_tier": 2}
    with pytest.raises(ValidationError):
        _persist_to(
            base_yaml,
            overlay_yaml,
            ("profiles", "default"),
            profile_dict,
        )

    assert base_yaml.read_bytes() == base_before
    assert (
        overlay_yaml.read_bytes() if overlay_yaml.exists() else b""
    ) == overlay_before




def test_config_GET_corrupt_yaml_returns_clean_503(tmp_path):
    """A duplicate-key config.yaml refuses with 503, not a raw 500."""
    (tmp_path / "config").mkdir(parents=True, exist_ok=True)
    config_yaml = tmp_path / "config" / "config.yaml"
    config_yaml.write_text(
        (ROOT / "config" / "config.yaml").read_text()
        + "profiles:\n  a:\n    min_tier: 1\nprofiles:\n  b:\n    min_tier: 2\n"
    )

    cfg = load_config(ROOT / "config" / "config.yaml")

    def _db_factory() -> sqlite3.Connection:
        return _make_db(tmp_path)

    router = build_router(cfg, _db_factory, base_dir=str(tmp_path))
    app = FastAPI()
    app.include_router(router, prefix="/admin")
    tc = TestClient(app, raise_server_exceptions=False)

    resp = tc.get("/admin/api/config")
    assert resp.status_code == 503
    assert "config.yaml" in resp.json()["detail"]


def test_config_concurrent_block_writes_are_atomic_no_zero_byte_backups(tmp_path):
    """Concurrent _persist_config_block writes never truncate config or leave 0-byte backups."""
    config_yaml = tmp_path / "config.yaml"
    shutil.copyfile(ROOT / "config" / "config.yaml", config_yaml)

    stop_reader = threading.Event()

    def reader() -> None:
        while not stop_reader.is_set():
            try:
                text = config_yaml.read_text()
            except FileNotFoundError:
                continue
            assert text.strip() != "", "config.yaml observed empty"
            assert "logging:" in text

    reader_thread = threading.Thread(target=reader)
    reader_thread.start()

    def writer(name: str, max_tier: int) -> None:
        _persist_config_block(config_yaml, ("profiles", name), {"max_tier": max_tier})

    threads = [
        threading.Thread(target=writer, args=("alpha", 1)),
        threading.Thread(target=writer, args=("beta", 2)),
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    stop_reader.set()
    reader_thread.join()

    final = yaml.safe_load(config_yaml.read_text())
    assert final["profiles"]["alpha"]["max_tier"] == 1
    assert final["profiles"]["beta"]["max_tier"] == 2

    backups = list(tmp_path.glob("config.yaml.bak.*"))
    assert backups, "expected at least one backup"
    for b in backups:
        assert b.stat().st_size > 0, f"zero-byte backup: {b}"
        assert b.read_text().strip() != ""


# ---------------------------------------------------------------------------
# Overlay survivability  / provenance edge-case tests
# ---------------------------------------------------------------------------

_OVERLAY_HEADER_PREFIX = "# Machine-local overlay"


def test_overlay_survives_git_restore_of_base(tmp_path):
    """A machine-local overlay (config.local.yaml) survives a git checkout of
    the base file — it is gitignored and therefore unaffected by ``git
    checkout -- config/config.yaml``."""
    repo = tmp_path / "repo"
    repo.mkdir()
    (repo / "config").mkdir()
    base_yaml = repo / "config" / "config.yaml"
    shutil.copyfile(ROOT / "config" / "config.yaml", base_yaml)

    local_yaml = repo / "config" / "config.local.yaml"
    local_yaml.write_text(
        f"{_OVERLAY_HEADER_PREFIX} — gitignored, never committed.\n"
        "logging:\n  level: warning\n"
    )

    subprocess.run(
        ["git", "init", "-q", str(repo)],
        check=True,
        capture_output=True,
    )
    subprocess.run(
        ["git", "-C", str(repo), "config", "user.email", "test@test.com"],
        check=True,
        capture_output=True,
    )
    subprocess.run(
        ["git", "-C", str(repo), "config", "user.name", "test"],
        check=True,
        capture_output=True,
    )
    (repo / ".gitignore").write_text("config.local.yaml\n")
    subprocess.run(
        ["git", "-C", str(repo), "add", "config/config.yaml", ".gitignore"],
        check=True,
        capture_output=True,
    )
    subprocess.run(
        ["git", "-C", str(repo), "commit", "-m", "init", "-q"],
        check=True,
        capture_output=True,
    )

    text = base_yaml.read_text()
    base_yaml.write_text(text.replace('level: "info"', 'level: "debug"'))

    subprocess.run(
        ["git", "-C", str(repo), "checkout", "--", "config/config.yaml"],
        check=True,
        capture_output=True,
    )

    assert local_yaml.exists()
    overlay_text = local_yaml.read_text()
    assert "level: warning" in overlay_text

    restored = base_yaml.read_text()
    assert 'level: "info"' in restored or "level: info" in restored


def test_config_GET_shows_overlay_source_when_overlay_exists(client):
    """When ``config.local.yaml`` overrides one allowlisted key, GET
    /admin/api/config reports ``source: "overlay"`` for that key and
    ``source: "base"`` for the rest."""
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")
    local_yaml.write_text(
        f"{_OVERLAY_HEADER_PREFIX} — gitignored, never committed.\n"
        "logging:\n  level: warning\n"
    )

    resp = tc.get("/admin/api/config")
    assert resp.status_code == 200
    body = resp.json()

    assert body["logging.level"]["source"] == "overlay"
    assert body["logging.level"]["value"] == "warning"
    assert body["objective.quality_tolerance"]["source"] == "base"


def test_first_write_creates_overlay_with_header(client):
    """POST to an allowlisted key when no overlay exists creates
    ``config.local.yaml`` starting with the expected comment header."""
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")
    assert not local_yaml.exists(), "overlay must not exist before the write"

    resp = tc.post(
        "/admin/api/config/circuit_breaker.enabled", json={"value": True}
    )
    assert resp.status_code == 200

    assert local_yaml.exists()
    text = local_yaml.read_text()
    assert text.startswith("# Machine-local overlay")
    assert "# Written by the admin portal" in text
    local = yaml.safe_load(text)
    assert local["circuit_breaker"]["enabled"] is True


def test_second_write_keeps_header(client):
    """A second admin write preserves the machine-local header comment."""
    tc, config_yaml = client
    local_yaml = config_yaml.with_name("config.local.yaml")

    resp = tc.post("/admin/api/config/logging.level", json={"value": "warning"})
    assert resp.status_code == 200

    resp = tc.post("/admin/api/config/circuit_breaker.enabled", json={"value": True})
    assert resp.status_code == 200

    text = local_yaml.read_text()
    assert text.startswith("# Machine-local overlay")
    # The header must appear exactly once.
    assert text.count("# Machine-local overlay") == 1
    local = yaml.safe_load(text)
    assert local["logging"]["level"] == "warning"
    assert local["circuit_breaker"]["enabled"] is True