"""Tests for the OpenAI-compatible surface — the endpoint every client uses.

This file exists because it did not. `/v1/chat/completions` is how opencode,
an SDK and plain curl all reach the router, and it had zero tests, which is
how a NameError survived on the pass-through path: `alternatives` read
`decision.runners_up`, but `decision` is only bound inside `if wants_routing`,
so every non-streaming request naming a real model id returned 500 before the
provider was ever called.

Nothing here touches the network. The classifier, the provider and the local
verifier are all stubbed, so the tests run in milliseconds and pin behaviour
rather than reachability.
"""

import io
import json
import sqlite3
from pathlib import Path

import pytest
from starlette.testclient import TestClient

import dispatcher
from dispatcher import Classification, app

ROOT = Path(__file__).resolve().parent.parent
SCHEMA_SQL = (ROOT / "schema.sql").read_text()

CHEAP = "cheap-model"
DEAR = "dear-model"


class FakeResponse:
    """Just enough of requests.Response for both dispatcher paths."""

    def __init__(self, payload=None, *, status_code=200, lines=None):
        self.status_code = status_code
        self._payload = payload or {}
        self._lines = lines or []
        self.text = json.dumps(self._payload)
        self.closed = False

    def json(self):
        return self._payload

    def iter_lines(self, decode_unicode=False):
        yield from self._lines

    def close(self):
        self.closed = True


def completion(model, content="hello there", *, completion_tokens=12):
    """A provider response shaped like NeuralWatt's, telemetry blocks included."""
    return {
        "id": "chatcmpl-test-1",
        "model": model,
        "choices": [
            {"message": {"role": "assistant", "content": content},
             "finish_reason": "stop"}
        ],
        "usage": {"prompt_tokens": 31, "completion_tokens": completion_tokens},
        "energy": {"energy_kwh": 5.0e-05, "avg_power_watts": 400.0,
                   "duration_seconds": 1.4, "attribution_ratio": 0.25,
                   "carbon_g_co2eq": 2.4e-03, "carbon_source": "agent_cache",
                   "grid_id": "FI"},
        "cost": {"request_cost_usd": 4.0e-04},
    }


@pytest.fixture
def router(tmp_path, monkeypatch):
    """The dispatcher pointed at a throwaway catalog, with nothing dialled out."""
    db_path = tmp_path / "test.db"
    conn = sqlite3.connect(db_path)
    conn.executescript(SCHEMA_SQL)
    for model_id, completion_price, vision in (
        (CHEAP, 0.30, 1),
        (DEAR, 9.00, 0),
    ):
        conn.execute(
            """
            INSERT INTO models (
                model_id, provider, base_model_id, tier, context_window,
                effective_context_window, max_output_tokens,
                cost_per_1m_prompt, cost_per_1m_completion,
                supports_vision, supports_json_mode,
                latency_class, reasoning_mode, context_variant,
                access_level, availability, last_updated
            ) VALUES (?, 'neuralwatt', ?, 2, 262128, 192500, 16384, ?, ?,
                      ?, 1, 'standard', 'default', 'full', 'public', 'active',
                      '2026-08-22T00:00:00+00:00')
            """,
            (model_id, model_id, completion_price / 3, completion_price, vision),
        )
    conn.commit()
    conn.close()

    monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
    # No Ollama in the test environment, and the local check would otherwise
    # fire as a background task and try to reach localhost:11434.
    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
    # The local vision fallback is off unless a test opts in; without this it
    # would fire for every image request and try to reach localhost.
    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", False)
    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")

    calls = []

    def fake_post(url, headers=None, json=None, stream=False, timeout=None):
        calls.append({"url": url, "body": json, "stream": stream})
        if stream:
            return FakeResponse(lines=STREAM_LINES)
        return FakeResponse(completion(json["model"]))

    monkeypatch.setattr(dispatcher.requests, "post", fake_post)
    monkeypatch.setattr(
        dispatcher, "classify",
        lambda task, context: Classification(
            task_category="coding_general", task_tier=2,
            required_context_tokens=100, confidence=0.9,
        ),
    )
    yield TestClient(app), calls, db_path


STREAM_LINES = [
    'data: {"id":"chatcmpl-stream-1","choices":[{"delta":{"content":"hel"}}]}',
    "",
    'data: {"id":"chatcmpl-stream-1","choices":[{"delta":{"content":"lo"},'
    '"finish_reason":"stop"}],"usage":{"prompt_tokens":31,'
    '"completion_tokens":9}}',
    "",
    ': energy {"energy_kwh": 5e-05, "carbon_g_co2eq": 0.0024, '
    '"carbon_source": "agent_cache"}',
    ': cost {"request_cost_usd": 0.0004}',
    "data: [DONE]",
    "",
]


def _messages(text="write me a function"):
    return [{"role": "user", "content": text}]


# --- the regression -------------------------------------------------------

def test_a_real_model_id_is_dispatched_rather_than_routed(router):
    """The documented pass-through: 'any real model id — dispatched as asked'.

    This raised NameError on `decision` before the provider was ever called,
    so the whole path 500'd while streaming clients never noticed.
    """
    client, calls, _ = router
    resp = client.post(
        "/v1/chat/completions",
        json={"model": DEAR, "messages": _messages()},
    )

    assert resp.status_code == 200
    assert calls[0]["body"]["model"] == DEAR, "the caller's choice must survive"
    assert resp.headers["X-Router-Model"] == DEAR


def test_a_real_model_id_does_not_pay_for_classification(router, monkeypatch):
    """A named model has nothing to classify; the ~2s round-trip is skipped."""
    client, _, _ = router

    def explode(task, context):
        raise AssertionError("classifier consulted for an explicit model id")

    monkeypatch.setattr(dispatcher, "classify", explode)
    assert client.post(
        "/v1/chat/completions", json={"model": CHEAP, "messages": _messages()}
    ).status_code == 200


# --- routing --------------------------------------------------------------

def test_auto_routes_and_reports_the_model_it_actually_used(router):
    """Quality ties, so cost breaks it — and the client is told what ran."""
    client, calls, _ = router
    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": _messages()},
    )

    assert resp.status_code == 200
    assert calls[0]["body"]["model"] == CHEAP
    assert resp.headers["X-Router-Model"] == CHEAP
    # The body must stay a valid OpenAI response naming the real model, not 'auto'.
    assert resp.json()["model"] == CHEAP


def test_a_provider_prefixed_router_name_still_routes(router):
    """opencode sends `llm-router/auto`; only the virtual names are stripped."""
    client, calls, _ = router
    resp = client.post(
        "/v1/chat/completions",
        json={"model": "llm-router/auto", "messages": _messages()},
    )
    assert resp.status_code == 200
    assert calls[0]["body"]["model"] == CHEAP


def test_no_eligible_model_is_a_422_naming_the_filters(router, monkeypatch):
    """A dead end must say which constraint killed it, not just 'no model'."""
    client, calls, _ = router
    monkeypatch.setattr(
        dispatcher, "classify",
        lambda task, context: Classification(
            task_category="coding_general", task_tier=3,
            required_context_tokens=100, confidence=0.9,
        ),
    )
    resp = client.post(
        "/v1/chat/completions", json={"model": "auto", "messages": _messages()}
    )

    assert resp.status_code == 422
    assert "tier >= 3" in resp.json()["detail"]
    assert not calls, "nothing should be dispatched when nothing qualifies"


# --- streaming ------------------------------------------------------------

def test_a_stream_is_proxied_verbatim_including_the_telemetry_comments(router):
    """NeuralWatt's energy/cost blocks are SSE comments; clients ignore them.

    They must still reach the client untouched — the router reads them on the
    way past rather than buffering the stream to strip them.
    """
    client, calls, _ = router
    resp = client.post(
        "/v1/chat/completions",
        json={"model": DEAR, "messages": _messages(), "stream": True},
    )

    assert resp.status_code == 200
    assert calls[0]["stream"] is True
    body = resp.text
    assert "hel" in body and "lo" in body
    assert ': energy {"energy_kwh": 5e-05' in body
    assert "data: [DONE]" in body


def test_a_streamed_call_still_logs_its_energy(router):
    """Streaming is how every agent client talks; unlogged, it is most of the traffic."""
    client, _, db_path = router
    client.post(
        "/v1/chat/completions",
        json={"model": DEAR, "messages": _messages(), "stream": True},
    )

    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    row = conn.execute(
        "SELECT model_id, request_id, energy_kwh, cost_usd, completion_tokens "
        "FROM energy_observations ORDER BY id DESC LIMIT 1"
    ).fetchone()
    conn.close()

    assert row["model_id"] == DEAR
    assert row["request_id"] == "chatcmpl-stream-1"
    assert row["energy_kwh"] == pytest.approx(5e-05)
    assert row["cost_usd"] == pytest.approx(4e-04)
    assert row["completion_tokens"] == 9


def test_a_dropped_upstream_connection_ends_the_stream_cleanly(router, monkeypatch):
    """The upstream connection can die mid-stream (NeuralWatt closing early, a
    network blip). Left uncaught, requests.exceptions.ChunkedEncodingError
    propagated straight out of the generator, which Starlette surfaced as an
    unhandled ASGI exception -- a full traceback in the log, and the client's
    connection cut dead with no error payload or [DONE].
    """
    import requests as requests_module

    client, _, _ = router

    def broken_lines():
        yield 'data: {"id":"chatcmpl-stream-1","choices":[{"delta":{"content":"hel"}}]}'
        raise requests_module.exceptions.ChunkedEncodingError("Response ended prematurely")

    def fake_post(url, headers=None, json=None, stream=False, timeout=None):
        return FakeResponse(lines=broken_lines())

    monkeypatch.setattr(dispatcher.requests, "post", fake_post)

    resp = client.post(
        "/v1/chat/completions",
        json={"model": DEAR, "messages": _messages(), "stream": True},
    )

    assert resp.status_code == 200
    body = resp.text
    assert "hel" in body
    assert "upstream stream interrupted" in body
    assert "data: [DONE]" in body


# --- what a request leaves in the journal ---------------------------------
#
# The reason this exists: watching a live session showed only uvicorn's access
# line, which does not even name the model that served it.

@pytest.fixture
def logbuf():
    """Capture what would reach the journal, through the real formatter."""
    import logs

    buf = io.StringIO()
    logs.configure("debug", stream=buf, journald=False)
    yield buf
    for handler in list(logs.log.handlers):
        logs.log.removeHandler(handler)


def _lines(logbuf, event):
    return [ln for ln in logbuf.getvalue().splitlines() if ln.startswith(f"{event} ")]


def _fields(line):
    out = {}
    for token in line.split(" ")[1:]:
        if "=" in token:
            key, _, value = token.partition("=")
            out[key] = value
    return out


def test_a_routed_request_logs_exactly_one_decision_line(router, logbuf):
    client, _, _ = router
    client.post("/v1/chat/completions",
                json={"model": "auto", "messages": _messages()})

    routes = _lines(logbuf, "route")

    assert len(routes) == 1, "routing twice for context must not log twice"
    fields = _fields(routes[0])
    assert fields["pick"] == CHEAP
    assert fields["cat"] == "coding_general"
    assert fields["tier"] == "2"
    assert int(fields["ms"]) >= 0


def test_the_dispatch_line_carries_the_join_key(router, logbuf):
    """rid is what pivots a journal line to its energy_observations row."""
    client, _, _ = router
    client.post("/v1/chat/completions",
                json={"model": "auto", "messages": _messages()})

    fields = _fields(_lines(logbuf, "dispatch")[0])

    assert fields["rid"] == "chatcmpl-test-1"
    assert fields["model"] == CHEAP
    assert fields["c_tok"] == "12"


def test_every_line_of_one_request_shares_a_trace_id(router, logbuf):
    client, _, _ = router
    client.post("/v1/chat/completions",
                json={"model": "auto", "messages": _messages()})

    ids = {_fields(ln)["id"] for ln in logbuf.getvalue().splitlines()}

    assert len(ids) == 1
    assert ids != {"-"}


def test_a_streamed_request_still_has_a_trace_id(router, logbuf):
    """The trap: a StreamingResponse generator runs in a different context.

    Read from the ContextVar inside the generator it comes back empty, and
    every streamed request -- which is all agent traffic -- logs a blank id.
    """
    client, _, _ = router
    client.post("/v1/chat/completions",
                json={"model": DEAR, "messages": _messages(), "stream": True})

    dispatched = _fields(_lines(logbuf, "dispatch")[0])

    assert dispatched["id"] != "-"
    assert dispatched["stream"] == "1"
    assert dispatched["rid"] == "chatcmpl-stream-1"


def test_debug_says_which_filter_dropped_a_model(router, logbuf):
    """'No model satisfies the hard filters' is otherwise a dead end."""
    client, _, db_path = router
    conn = sqlite3.connect(db_path)
    conn.execute(
        """
        INSERT INTO models (
            model_id, provider, tier, context_window, effective_context_window,
            latency_class, access_level, availability, last_updated
        ) VALUES ('held-model', 'neuralwatt', 2, 262128, 192500, 'flex',
                  'public', 'active', '2026-08-22T00:00:00+00:00')
        """
    )
    conn.commit()
    conn.close()

    client.post("/v1/chat/completions",
                json={"model": "auto", "messages": _messages()})

    dropped = {_fields(ln)["model"]: _fields(ln)["reason"]
               for ln in _lines(logbuf, "filter")}

    assert dropped["held-model"] == "latency_class(flex)"


def test_a_pass_through_is_labelled_as_one(router, logbuf):
    client, _, _ = router
    client.post("/v1/chat/completions",
                json={"model": DEAR, "messages": _messages()})

    assert not _lines(logbuf, "route"), "nothing was routed"
    assert _fields(_lines(logbuf, "passthrough")[0])["model"] == DEAR


def test_no_conversation_text_ever_reaches_the_log(router, logbuf):
    """Prompts here run 60k-150k tokens and the journal is on disk.

    Asserted at DEBUG, the most verbose level, on both the request and the
    answer.
    """
    secret = "SQUAMOUS-EPHEMERAL-9137"
    client, _, _ = router

    client.post("/v1/chat/completions",
                json={"model": "auto", "messages": _messages(f"refactor {secret}")})

    assert secret not in logbuf.getvalue()
    # The stubbed provider answers "hello there"; that must not appear either.
    assert "hello there" not in logbuf.getvalue()


def test_the_decision_line_reports_the_real_provenance(router, logbuf):
    """The re-route for measured context must not read as a client override.

    chat_completions routes a second time when the conversation measures larger
    than the classifier guessed, passing category and tier back in — which
    makes the resulting Classification say source='override'. In a log line
    that means "the client chose this", and the client chose nothing: the
    classifier ran and only the context figure was replaced.
    """
    client, _, _ = router
    # ~1200 chars / 3 = 400 tokens measured, above the stubbed estimate of 100.
    client.post("/v1/chat/completions",
                json={"model": "auto", "messages": _messages("refactor " + "x " * 600)})

    fields = _fields(_lines(logbuf, "route")[0])

    assert fields["src"] == "classifier", "the classifier did decide this"
    assert fields["ctx_src"] == "measured", "but the context came from measurement"
    assert int(fields["ctx"]) > 100


def test_a_caller_supplied_context_says_so(router, logbuf):
    client, _, _ = router
    client.post("/route", json={"task": "refactor this",
                                "task_category": "coding_general",
                                "task_tier": 2,
                                "required_context_tokens": 5000})

    fields = _fields(_lines(logbuf, "route")[0])

    assert fields["src"] == "override"
    assert fields["ctx_src"] == "caller"


# --- capability gates: vision and JSON mode -------------------------------

def _image_messages(text="what is in this image?"):
    """A user turn carrying an inline image_url part, OpenAI multimodal shape."""
    return [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": text},
                {
                    "type": "image_url",
                    "image_url": {"url": "data:image/png;base64,AAAA"},
                },
            ],
        }
    ]


def _drop_cheap_by_tier(client_type, db_path):
    """Make CHEAP ineligible so DEAR is the only remaining candidate."""
    conn = sqlite3.connect(db_path)
    conn.execute("UPDATE models SET tier = 1 WHERE model_id = ?", (CHEAP,))
    conn.commit()
    conn.close()


def _local_vision_fake(monkeypatch, router_calls, content="local caption", status=200):
    """Point the local vision fallback at a fake that returns `content`."""
    def fake_post(url, headers=None, json=None, stream=False, timeout=None):
        if "/chat/completions" in url:
            router_calls.append({"url": url, "body": json, "stream": stream, "local": True})
            return FakeResponse(
                {
                    "choices": [
                        {"message": {"role": "assistant", "content": content},
                         "finish_reason": "stop"}
                    ]
                },
                status_code=status,
            )
        router_calls.append({"url": url, "body": json, "stream": stream})
        return FakeResponse(completion(json["model"]))

    monkeypatch.setattr(dispatcher.requests, "post", fake_post)


def test_an_image_request_is_routed_to_a_vision_model(router):
    """Image parts hard-restrict to vision-capable rows; CHEAP has vision."""
    client, calls, _ = router
    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": _image_messages()},
    )

    assert resp.status_code == 200
    assert calls[0]["body"]["model"] == CHEAP
    assert resp.headers["X-Router-Model"] == CHEAP


def test_an_image_request_excludes_a_non_vision_candidate(router):
    """DEAR is eligible-tiered but lacks vision, so no candidate meets the ask."""
    client, calls, db_path = router
    _drop_cheap_by_tier(dispatcher, db_path)
    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": _image_messages()},
    )

    assert resp.status_code == 422
    assert "vision" in resp.json()["detail"]
    assert not calls, "a non-vision model must never be dispatched for an image"


def test_no_vision_model_uses_local_fallback_when_enabled(router, monkeypatch):
    """No cloud vision candidate, local vision on: the local model answers."""
    client, calls, db_path = router
    _drop_cheap_by_tier(dispatcher, db_path)
    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)
    _local_vision_fake(monkeypatch, calls, content="local caption")

    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": _image_messages()},
    )

    assert resp.status_code == 200
    assert resp.json()["choices"][0]["message"]["content"] == "local caption"
    local = [c for c in calls if c["url"].endswith("/chat/completions")]
    assert local, "the local fallback should POST to its own /chat/completions"
    parts = local[0]["body"]["messages"][0]["content"]
    assert any(
        isinstance(p, dict) and p.get("type") == "image_url" for p in parts
    ), "the image_url part must survive into the local call"


def test_local_fallback_skips_when_json_mode_is_requested(router, monkeypatch):
    """Local vision answers in prose; a json_object request must not use it.

    JSON mode -- not vision -- is what excluded every cloud candidate here, so
    the fallback must not fire or it would silently break the json_object
    contract with free-form text instead of the 422 naming the missing
    capability.
    """
    client, calls, db_path = router
    _drop_cheap_by_tier(dispatcher, db_path)
    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)
    _local_vision_fake(monkeypatch, calls, content="should not be used")

    resp = client.post(
        "/v1/chat/completions",
        json={
            "model": "auto",
            "messages": _image_messages(),
            "response_format": {"type": "json_object"},
        },
    )

    assert resp.status_code == 422
    assert "json" in resp.json()["detail"]
    local = [c for c in calls if c.get("local")]
    assert not local, "the local vision fallback must not run for a json_object request"


def test_local_fallback_respects_streaming(router, monkeypatch):
    client, calls, db_path = router
    _drop_cheap_by_tier(dispatcher, db_path)
    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)
    _local_vision_fake(monkeypatch, calls, content="streamed caption")

    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": _image_messages(), "stream": True},
    )

    assert resp.status_code == 200
    assert "streamed caption" in resp.text
    assert "data: [DONE]" in resp.text


def test_local_fallback_failure_still_422s(router, monkeypatch):
    """A local vision failure must be visible, not swallowed into a 200."""
    client, calls, db_path = router
    _drop_cheap_by_tier(dispatcher, db_path)
    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)
    _local_vision_fake(monkeypatch, calls, status=500)

    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": _image_messages()},
    )

    assert resp.status_code == 422
    assert "vision" in resp.json()["detail"]


def test_a_json_object_response_format_routes_to_a_json_capable_model(router):
    """response_format json_object hard-restricts to JSON-mode-capable rows."""
    client, calls, _ = router
    resp = client.post(
        "/v1/chat/completions",
        json={
            "model": "auto",
            "messages": _messages(),
            "response_format": {"type": "json_object"},
        },
    )

    assert resp.status_code == 200
    assert calls[0]["body"]["model"] == CHEAP


def test_a_json_object_request_without_a_json_capable_model_422s(router):
    client, calls, db_path = router
    conn = sqlite3.connect(db_path)
    conn.execute("UPDATE models SET supports_json_mode = 0")
    conn.commit()
    conn.close()

    resp = client.post(
        "/v1/chat/completions",
        json={
            "model": "auto",
            "messages": _messages(),
            "response_format": {"type": "json_object"},
        },
    )

    assert resp.status_code == 422
    assert "json" in resp.json()["detail"]
    assert not calls


def test_a_plain_text_request_is_unaffected_by_the_gates(router):
    """No images, no response_format: routing is exactly as before."""
    client, calls, _ = router
    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": _messages()},
    )

    assert resp.status_code == 200
    assert calls[0]["body"]["model"] == CHEAP
    assert resp.headers["X-Router-Model"] == CHEAP


def test_a_pinned_non_vision_model_with_an_image_is_a_clear_422(router):
    """A pin that cannot satisfy the request fails early with a named reason."""
    client, calls, _ = router
    resp = client.post(
        "/v1/chat/completions",
        json={"model": DEAR, "messages": _image_messages()},
    )

    assert resp.status_code == 422
    assert "vision" in resp.json()["detail"]
    assert not calls, "no provider call should happen for an impossible pin"


def test_a_pinned_non_vision_model_dispatches_when_vision_gate_is_off(router, monkeypatch):
    """The pin check honors `routing.require_vision`, like routed traffic.

    `require_vision: false` accepts the occasional provider 400 in exchange
    for not fail-closing on an unconfirmed flag; a pinned model must face the
    same config switch as `auto` routing, not an unconditional 422.
    """
    client, calls, _ = router
    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)

    resp = client.post(
        "/v1/chat/completions",
        json={"model": DEAR, "messages": _image_messages()},
    )

    assert resp.status_code == 200
    assert calls[0]["body"]["model"] == DEAR, "the pin must dispatch as asked"


def test_a_provider_prefixed_pin_is_resolved_for_the_capability_check(router):
    """opencode sends a pin as `provider/model` too, not just for `auto`.

    Before the fix, `requested` kept its `llm-router/` prefix past this
    point: the capability pre-check looked up a model_id the catalog has
    never heard of, found no row, and fail-closed a genuinely capable pin
    into a false 422. It must resolve to the bare id and go through, and the
    provider must see the bare id too -- NeuralWatt has never heard of
    `llm-router/...` either.
    """
    client, calls, _ = router
    resp = client.post(
        "/v1/chat/completions",
        json={"model": f"llm-router/{CHEAP}", "messages": _image_messages()},
    )

    assert resp.status_code == 200
    assert calls[0]["body"]["model"] == CHEAP
    assert resp.headers["X-Router-Model"] == CHEAP


def test_a_provider_prefixed_pin_still_fails_closed_when_incapable(router):
    """The prefix-resolution fix must not turn into a bypass."""
    client, calls, _ = router
    resp = client.post(
        "/v1/chat/completions",
        json={"model": f"llm-router/{DEAR}", "messages": _image_messages()},
    )

    assert resp.status_code == 422
    assert "vision" in resp.json()["detail"]
    assert not calls, "no provider call should happen for an impossible pin"


def test_the_two_pass_reroute_passes_image_capability(router):
    """The measured-context reroute must carry the image gate, not lose it."""
    client, calls, _ = router
    # ~1200 chars / 3 = 400 measured tokens > the stubbed estimate of 100, so
    # chat_completions reroutes — and the reroute must still require vision.
    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": _image_messages("refactor " + "x " * 600)},
    )

    assert resp.status_code == 200
    assert resp.headers["X-Router-Model"] == CHEAP
    assert calls[0]["body"]["model"] == CHEAP


def test_local_fallback_refuses_a_remote_image_url(router, monkeypatch):
    """A remote http(s) URL is an SSRF vector by indirection and must not be
    forwarded to the local model, even when the fallback is enabled."""
    client, calls, db_path = router
    _drop_cheap_by_tier(dispatcher, db_path)
    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)
    _local_vision_fake(monkeypatch, calls, content="should not be used")

    messages = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "describe this"},
                {
                    "type": "image_url",
                    "image_url": {"url": "http://internal/private.png"},
                },
            ],
        }
    ]
    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": messages},
    )

    assert resp.status_code == 422
    assert "vision" in resp.json()["detail"]
    local = [c for c in calls if c.get("local")]
    assert not local, "a remote image URL must never reach the local vision endpoint"


def test_local_fallback_refuses_a_degenerate_image_url_part(router, monkeypatch):
    """A part that declares itself an image but carries no readable value at
    all must still fail the "every part is a verified data: URI" check --
    silently skipping it would let it slip past the SSRF guard the same way
    a remote URL is refused above."""
    client, calls, db_path = router
    _drop_cheap_by_tier(dispatcher, db_path)
    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)
    _local_vision_fake(monkeypatch, calls, content="should not be used")

    messages = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "describe this"},
                {"type": "image_url"},
            ],
        }
    ]
    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": messages},
    )

    assert resp.status_code == 422
    assert "vision" in resp.json()["detail"]
    local = [c for c in calls if c.get("local")]
    assert not local, "an unverifiable image part must never reach the local vision endpoint"


def test_local_fallback_wont_masquerade_an_empty_answer_as_200(router, monkeypatch):
    """A 200 with empty content is a failed answer, and must fall through to 422
    rather than return an empty 200 — a hidden failure must not look like a win."""
    client, calls, db_path = router
    _drop_cheap_by_tier(dispatcher, db_path)
    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)
    _local_vision_fake(monkeypatch, calls, content=" ")

    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": _image_messages()},
    )

    assert resp.status_code == 422
    assert "vision" in resp.json()["detail"]


def test_local_fallback_refuses_too_many_images(router, monkeypatch):
    """An image count above the budget refuses the local call before spending it."""
    client, calls, db_path = router
    _drop_cheap_by_tier(dispatcher, db_path)
    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)
    _local_vision_fake(monkeypatch, calls, content="caption")

    many = _image_messages()
    for _ in range(dispatcher.cfg.local_vision.max_images + 1):
        many[0]["content"].append({"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}})

    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": many},
    )

    assert resp.status_code == 422
    local = [c for c in calls if c.get("local")]
    assert not local, "over-budget image count must skip the local call"


def test_local_fallback_refuses_when_api_key_env_is_missing(router, monkeypatch):
    """A configured local vision api_key_env with no env var declines the call."""
    client, calls, db_path = router
    _drop_cheap_by_tier(dispatcher, db_path)
    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)
    monkeypatch.setattr(dispatcher.cfg.local_vision, "api_key_env", "LOCAL_VISION_KEY")
    monkeypatch.delenv("LOCAL_VISION_KEY", raising=False)
    _local_vision_fake(monkeypatch, calls, content="caption")

    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": _image_messages()},
    )

    assert resp.status_code == 422
    local = [c for c in calls if c.get("local")]
    assert not local, "a missing vision key must skip the local call"


def test_local_fallback_refuses_an_unparseable_response(router, monkeypatch):
    """An unparseable local response body must not yield a 200."""
    client, calls, db_path = router
    _drop_cheap_by_tier(dispatcher, db_path)
    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)

    def fake_post(url, headers=None, json=None, stream=False, timeout=None):
        calls.append({"url": url, "body": json, "stream": stream, "local": True})
        return FakeResponse({}, status_code=200)

    monkeypatch.setattr(dispatcher.requests, "post", fake_post)

    resp = client.post(
        "/v1/chat/completions",
        json={"model": "auto", "messages": _image_messages()},
    )

    assert resp.status_code == 422
    assert "vision" in resp.json()["detail"]


def test_a_pinned_non_json_model_with_json_response_format_422s(router):
    """The JSON-mode arm of the pass-through check mirrors the vision one: a pin
    whose model lacks JSON mode gets a clear 422, not a provider 400."""
    client, calls, db_path = router
    conn = sqlite3.connect(db_path)
    conn.execute("UPDATE models SET supports_json_mode = 0 WHERE model_id = ?", (DEAR,))
    conn.commit()
    conn.close()

    resp = client.post(
        "/v1/chat/completions",
        json={
            "model": DEAR,
            "messages": _messages(),
            "response_format": {"type": "json_object"},
        },
    )

    assert resp.status_code == 422
    assert "json" in resp.json()["detail"]
    assert not calls, "no provider call should happen for an impossible json pin"