portfolio-optimiser/tests/test_hosting_loadbearing.py

"""Fase 4d hosted entrypoint (``hosting.py`` + root ``main.py``): the Foundry hosted-agent
runtime contract implemented DIRECTLY around ``run_project`` — the wrapper form BOTH 13.08
measurements prescribe (hosting-pakka: ``InvocationsHostServer`` finnes kun i bygg som krever
core>=1.13.0; gjenbrukt workflow: kall-serie [2, 0, 0] — single-use på 1.9.0).

Load-bearing surface pinned here:

* ``GET /readiness`` → 200. We use NO protocol library (the measurement above), so the endpoint
  the platform health-checks is OURS to serve — nothing serves it for us.
* ``POST /invocations`` wires the payload's whitelisted fields into ``run_project`` and returns
  the outbox-shaped outcome payload. ``profile`` defaults to ``"azure"`` on THIS surface (a
  hosted container has no local endpoint; run_project's own default stays LOCAL).
* Validation, NEVER repair: an unknown field is a 400 naming the field — the permissive-schema
  trap (valg-doc §0: "en fil som ser konfigurert ut og ikke er det") applied to our own surface.
* Honest error mapping: ``ValueError`` (incl. pydantic contract violations) → 400; run failures
  → 500 ``{error_type, error}`` (mirrors ``RunFailure``); a ``Rejection`` is a SUCCESSFUL run →
  200 — the negative outcome belongs to the payload, never to the transport.
* The server is asyncio on the ONE loop (NG1: ``test_no_thread_or_process_path_exists_under_src``
  ratchets src/ thread-free) — these tests run client and server as coroutines on the SAME loop,
  which only works because nothing in the server blocks it.
* Root ``main.py`` is the ONE process entry (Dockerfile CMD + azure.yaml point at it): the
  subprocess test is the ONLY test that catches a detached shim or a detached SIGTERM handler
  (P4-presedensen: entry-point-mutasjoner fanges aldri av in-process-tester).

Fase 4e closes two gaps the above leaves open, and both are about things a stand-in cannot see:

* **The whitelist composes with the REAL ``run_project``.** Every test above hands ``invoke`` a
  stand-in that swallows ``**kwargs``, so the whitelist could name a field ``run_project`` does not
  take — or hand it the same argument twice — and every one of them would stay green while a live
  container answered 500. The end-to-end test drives a whole run (bundle navigation → debate →
  deterministic validator → checker gate → verdict) against a SCRIPTED backend. The seam is
  ``run._default_factory``, not the payload: ``client_factory`` is refused by the whitelist ON
  PURPOSE (a caller must never choose the server's model client), so patching the factory the run
  falls back to is the only injection point this surface leaves — the same argument
  ``test_run_cli_loadbearing`` makes for ``main()``.
* **The deployment artifacts are raw-text-gated.** ``Dockerfile`` and ``azure.yaml`` are the two
  files that decide whether the image the platform pulls can run at all, and NO test executes
  them here (``docker build``/``azd deploy`` are operator-gated). A raw-text gate is therefore the
  only mechanism available: it pins ``--platform linux/amd64`` (measured required — spike §1.4;
  an arm64 image built on this Intel-free-of-charge assumption would fail only in the cloud) and
  the ONE-copy rule for the start command (the image's ``CMD``; ``azure.yaml`` carries no
  ``startupCommand`` to drift from it). Guard-tester leser kildefiler som RÅ TEKST — reformulate
  the prose around them, never the strings they pin.
"""

from __future__ import annotations

import asyncio
import json
import os
import signal
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any

import pytest

from portfolio_optimiser import hosting
from portfolio_optimiser import run as run_module
from portfolio_optimiser.budget import BudgetExceeded
from portfolio_optimiser.ir import AffectedItem, SavingsProposal
from portfolio_optimiser.provenance import Citation, ProvenanceStamp
from portfolio_optimiser.retrieval import TextSpan
from portfolio_optimiser.run import RunResult
from portfolio_optimiser.simulation import scripted_factory
from portfolio_optimiser.validator import Rejection, ValidatedProposal
from portfolio_optimiser.verdicts import ProposalFeatures, Verdict, VerdictStore

_REPO_ROOT = Path(__file__).resolve().parents[1]
_BUNDLE_DIR = _REPO_ROOT / "shared" / "examples" / "bygg-energi-mikro"
_BUNDLE_PID = "BYGG-KONTOR-NORD"

# The proposer's scripted reply for the end-to-end run. Its cost line IS the bundle's own code, and
# the claim (30 000) sits under the degenerate Monte Carlo P90 of 90 000 that the DETERMINISTIC
# validator computes from it — so the validated outcome is the validator's arithmetic, not a
# scripted string. bygg-energi-mikro ships no ``cost-baseline.json``, so the S4.0 stage-0 anchoring
# is legitimately absent here (a commons-owned golden predates the amendment).
_BUNDLE_PROPOSER_REPLY = (
    '{"measure":"LED-retrofit av kontorbelysning","affected_items":'
    '[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],"claimed_saving_nok":30000}'
)
_BUNDLE_P90 = 90000.0

_PROPOSAL = SavingsProposal(
    project_id="P1",
    measure="LED-retrofit av kontorbelysning",
    affected_items=[AffectedItem(code="ENERGI-TOTAL-EL", quantity=1000.0, unit_cost=1.0)],
    claimed_saving_nok=200.0,
    assumptions={"ENERGI-TOTAL-EL": (0.8, 1.2)},
)
_PROVENANCE = ProvenanceStamp(
    citations=[
        Citation(file="f.md", locator=TextSpan(start_index=0, end_index=5), snippet="hello")
    ],
    model="synthetic",
    role="proposer",
    validator_decision="validated",
    token_usage=8,
)
_VALIDATED = ValidatedProposal(
    proposal=_PROPOSAL, p10=100.0, p50=150.0, p90=200.0, nominal_feasible=180.0
)
_REJECTION = Rejection(proposal=_PROPOSAL, reason="stage 0: unknown cost code")
_VERDICT = Verdict(
    id="vid-hosted",
    proposal_features=ProposalFeatures(
        affected_codes=frozenset({"ENERGI-TOTAL-EL"}),
        measure_type="LED-retrofit av kontorbelysning",
        claimed_saving_nok=200.0,
    ),
    decision="approved",
    rationale="expert reviewed",
)

_PAYLOAD = {
    "project_id": "P1",
    "docs_dir": "docs",
    "verdict_input": {"decision": "approved", "rationale": "expert"},
}


def _result(outcome: ValidatedProposal | Rejection) -> RunResult:
    return RunResult(
        outcome=outcome,
        provenance=_PROVENANCE,
        verdict=_VERDICT,
        retrieved=[],
        store=VerdictStore(verdicts=[]),
        debate_output="debate",
        checker_verdict="approve",
    )


class _Recorder:
    """An async stand-in for ``run_project``: records every (args, kwargs), then returns the
    configured RunResult or raises the configured error."""

    def __init__(
        self,
        result: RunResult | None = None,
        error: Exception | None = None,
    ) -> None:
        self.calls: list[tuple[tuple[Any, ...], dict[str, Any]]] = []
        self._run_result = result
        self._error = error

    async def __call__(self, *args: Any, **kwargs: Any) -> RunResult:
        self.calls.append((args, kwargs))
        if self._error is not None:
            raise self._error
        assert self._run_result is not None
        return self._run_result


@pytest.fixture()
async def served() -> Any:
    """The contract server on THIS test's loop (port 0 → ephemeral). Client and server share
    the loop, so a server that blocked it would hang these tests — the fixture is itself a
    check that nothing in the request path blocks."""
    server = await hosting.start_server("127.0.0.1", 0)
    port = server.sockets[0].getsockname()[1]
    yield f"127.0.0.1:{port}"
    server.close()
    await server.wait_closed()


async def _request(
    base: str, method: str, path: str, body: bytes | None = None
) -> tuple[int, bytes]:
    """A minimal HTTP/1.1 client on the same loop (urllib would block the shared loop)."""
    host, port_text = base.split(":")
    reader, writer = await asyncio.open_connection(host, int(port_text))
    head = f"{method} {path} HTTP/1.1\r\nHost: {base}\r\nConnection: close\r\n"
    if body is not None:
        head += f"Content-Type: application/json\r\nContent-Length: {len(body)}\r\n"
    writer.write(head.encode("latin-1") + b"\r\n" + (body or b""))
    await writer.drain()
    raw = await reader.read()
    writer.close()
    await writer.wait_closed()
    status = int(raw.split(b" ", 2)[1])
    return status, raw.split(b"\r\n\r\n", 1)[1]


async def _get(base: str, path: str) -> tuple[int, bytes]:
    return await _request(base, "GET", path)


async def _post(base: str, path: str, payload: Any) -> tuple[int, dict[str, Any]]:
    body = payload if isinstance(payload, bytes) else json.dumps(payload).encode("utf-8")
    status, raw = await _request(base, "POST", path, body)
    return status, json.loads(raw.decode("utf-8"))


async def test_readiness_returns_200(served: str) -> None:
    """The platform health-checks GET /readiness; with no protocol library, serving it is ours."""
    status, _body = await _get(served, "/readiness")
    assert status == 200


async def test_unknown_paths_are_404(served: str) -> None:
    status, _ = await _get(served, "/other")
    assert status == 404
    status, _ = await _post(served, "/other", _PAYLOAD)
    assert status == 404


async def test_invocations_wires_payload_into_run_project(
    served: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The wiring seam: the payload's fields reach run_project as its own arguments, the hosted
    default profile is 'azure', and the response carries the outbox-shaped outcome + proposal +
    provenance. RED if the handler detaches the runner or drops the field mapping."""
    recorder = _Recorder(result=_result(_VALIDATED))
    monkeypatch.setattr(hosting, "run_project", recorder)

    status, body = await _post(served, "/invocations", {**_PAYLOAD, "max_rounds": 2})

    assert status == 200
    assert recorder.calls, "run_project was never called"
    args, kwargs = recorder.calls[0]
    assert args == ("P1",)
    assert kwargs["docs_dir"] == "docs"
    assert kwargs["verdict_input"] == {"decision": "approved", "rationale": "expert"}
    assert kwargs["max_rounds"] == 2
    # Hosted default: a container has no local OpenAI-compatible endpoint (run_project's own
    # default stays LOCAL — the two defaults are different on purpose, and this pins OURS).
    assert kwargs["profile"] == "azure"
    assert body["outcome_type"] == "validated"
    assert body["p90"] == 200.0
    assert body["checker_verdict"] == "approve"
    assert body["verdict_id"] == "vid-hosted"
    assert body["proposal"]["measure"] == "LED-retrofit av kontorbelysning"
    assert body["provenance"]["model"] == "synthetic"


async def test_profile_in_payload_overrides_hosted_default(
    served: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    recorder = _Recorder(result=_result(_VALIDATED))
    monkeypatch.setattr(hosting, "run_project", recorder)

    status, _body = await _post(served, "/invocations", {**_PAYLOAD, "profile": "local"})

    assert status == 200
    _args, kwargs = recorder.calls[0]
    assert kwargs["profile"] == "local"


async def test_unknown_field_is_refused_never_repaired(
    served: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Validation, never repair: an unknown field is a 400 naming the field and the runner is
    never reached — dropping it silently would be the permissive-schema failure mode (valg §0).
    The control arm (same payload minus the key) proves the refusal is FOR the key."""
    recorder = _Recorder(result=_result(_VALIDATED))
    monkeypatch.setattr(hosting, "run_project", recorder)

    status, body = await _post(served, "/invocations", {**_PAYLOAD, "outbox_dir": "/x"})
    assert status == 400
    assert "outbox_dir" in body["error"]
    assert recorder.calls == []  # refused BEFORE the runner — never repaired-and-run

    control_status, _ = await _post(served, "/invocations", _PAYLOAD)
    assert control_status == 200
    assert len(recorder.calls) == 1


@pytest.mark.parametrize("missing", ["project_id", "docs_dir", "verdict_input"])
async def test_missing_required_field_is_400(
    served: str, monkeypatch: pytest.MonkeyPatch, missing: str
) -> None:
    recorder = _Recorder(result=_result(_VALIDATED))
    monkeypatch.setattr(hosting, "run_project", recorder)

    payload = {k: v for k, v in _PAYLOAD.items() if k != missing}
    status, body = await _post(served, "/invocations", payload)

    assert status == 400
    assert missing in body["error"]
    assert recorder.calls == []


async def test_non_object_or_invalid_json_body_is_400(
    served: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    recorder = _Recorder(result=_result(_VALIDATED))
    monkeypatch.setattr(hosting, "run_project", recorder)

    status, _ = await _post(served, "/invocations", b"[1, 2]")
    assert status == 400
    status, _ = await _post(served, "/invocations", b"not json")
    assert status == 400
    assert recorder.calls == []


async def test_rejected_outcome_is_200_with_reason(
    served: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A Rejection is a successful run with a negative outcome — 200, outcome_type=rejected, the
    reason, and NO percentile keys (key-absence on the parsed dict, not a substring)."""
    recorder = _Recorder(result=_result(_REJECTION))
    monkeypatch.setattr(hosting, "run_project", recorder)

    status, body = await _post(served, "/invocations", _PAYLOAD)

    assert status == 200
    assert body["outcome_type"] == "rejected"
    assert body["reason"] == "stage 0: unknown cost code"
    assert "p90" not in body


async def test_contract_violation_is_400_and_run_failure_is_500(
    served: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    """ValueError (pydantic contract violations subclass it) is the CALLER's error → 400; any
    other failure is an honest 500 carrying {error_type, error} (mirrors RunFailure's shape).
    BudgetExceeded is RuntimeError, so it lands in the 500 arm — with observed != limit so the
    two can never be conflated by an echo (kø-(y))."""
    monkeypatch.setattr(
        hosting, "run_project", _Recorder(error=ValueError("docs_dir does not exist"))
    )
    status, body = await _post(served, "/invocations", _PAYLOAD)
    assert status == 400
    assert "docs_dir does not exist" in body["error"]

    monkeypatch.setattr(hosting, "run_project", _Recorder(error=BudgetExceeded("tokens", 100, 173)))
    status, body = await _post(served, "/invocations", _PAYLOAD)
    assert status == 500
    assert body["error_type"] == "BudgetExceeded"
    assert "limit=100" in body["error"]
    assert "173" in body["error"]


async def test_readiness_answers_while_an_invocation_is_in_flight(
    served: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Liveness on ONE loop: while an invocation awaits (model I/O), /readiness must still
    answer — the platform health-checks during long runs, and a server that serialized the
    whole process on one in-flight request would be killed as unready."""
    release = asyncio.Event()

    class _Blocking(_Recorder):
        async def __call__(self, *args: Any, **kwargs: Any) -> RunResult:
            await release.wait()
            return await super().__call__(*args, **kwargs)

    monkeypatch.setattr(hosting, "run_project", _Blocking(result=_result(_VALIDATED)))
    in_flight = asyncio.ensure_future(_post(served, "/invocations", _PAYLOAD))
    try:
        status, _ = await asyncio.wait_for(_get(served, "/readiness"), timeout=10)
        assert status == 200
    finally:
        release.set()
    status, _body = await asyncio.wait_for(in_flight, timeout=10)
    assert status == 200


def test_port_resolution_is_truthiness_not_presence(monkeypatch: pytest.MonkeyPatch) -> None:
    """PORT on truthiness (the 4b rule): an exported-empty PORT is a shell accident, not a bind
    instruction. Default 8088 is the hosted-agent contract's port."""
    monkeypatch.delenv("PORT", raising=False)
    assert hosting.resolve_port() == 8088
    monkeypatch.setenv("PORT", "9001")
    assert hosting.resolve_port() == 9001
    monkeypatch.setenv("PORT", "")
    assert hosting.resolve_port() == 8088


def _blocking_get(url: str) -> tuple[int, str]:
    try:
        with urllib.request.urlopen(url, timeout=10) as resp:
            return resp.status, resp.read().decode("utf-8")
    except urllib.error.HTTPError as err:
        return err.code, err.read().decode("utf-8")


def test_main_entrypoint_serves_and_stops_on_sigterm() -> None:
    """Root main.py is the ONE process entry (Dockerfile CMD + azure.yaml point at it): started
    as a subprocess it must serve /readiness and exit 0 on SIGTERM. This is the only test that
    catches a shim that stops calling hosting.main() or a detached SIGTERM handler."""
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        port = probe.getsockname()[1]
    repo_root = Path(__file__).resolve().parents[1]
    proc = subprocess.Popen(
        [sys.executable, str(repo_root / "main.py")],
        env={**os.environ, "PORT": str(port)},
        cwd=repo_root,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    try:
        deadline = time.monotonic() + 60
        up = False
        while time.monotonic() < deadline:
            try:
                status, _ = _blocking_get(f"http://127.0.0.1:{port}/readiness")
                if status == 200:
                    up = True
                    break
            except (urllib.error.URLError, OSError):
                time.sleep(0.2)
        assert up, "main.py never served /readiness"
        proc.send_signal(signal.SIGTERM)
        assert proc.wait(timeout=15) == 0
    finally:
        if proc.poll() is None:
            proc.kill()
            proc.wait()


# --- Fase 4e: the whitelist against the REAL run_project, and the deployment artifacts ----------


@pytest.fixture()
def _scripted_backend(monkeypatch: pytest.MonkeyPatch) -> list[str]:
    """Inject a network-free scripted backend into the seam a hosted invocation ACTUALLY resolves
    through.

    ``client_factory`` is refused by the invocations whitelist on purpose — the caller of a hosted
    agent must never choose the server's model client — so an end-to-end test cannot inject through
    the payload. ``run_project`` falls back to the module-level ``_default_factory`` when no factory
    is passed, which makes patching it the only injection point this surface leaves (the argument
    ``test_run_cli_loadbearing`` makes for ``main()``, one layer up).

    Returns the shared prompt sink: non-empty is the proof that the run went through the real
    machinery rather than short-circuiting somewhere before the debate."""
    monkeypatch.delenv("PORTFOLIO_MODEL_MAP", raising=False)
    monkeypatch.delenv("PORTFOLIO_FOUNDRY_PROJECT_ENDPOINT", raising=False)
    sink: list[str] = []
    monkeypatch.setattr(
        run_module,
        "_default_factory",
        lambda _profile: scripted_factory(
            {"proposer": _BUNDLE_PROPOSER_REPLY, "checker": "VERDICT: APPROVE"}, sink
        ),
    )
    return sink


async def test_invocations_answers_through_the_real_run_project(
    served: str, _scripted_backend: list[str]
) -> None:
    """END-TO-END: a bundle payload posted to ``/invocations`` drives a WHOLE real run and comes
    back as a valid response — no stand-in for ``run_project`` anywhere in the path.

    Every other invocations test hands ``invoke`` a recorder that swallows ``**kwargs``, so none of
    them can see whether the whitelist actually composes with ``run_project``'s signature. Two ways
    to break that are invisible to a recorder and fatal in a container: naming a field
    ``run_project`` does not take, and passing it an argument twice. Both surface here as a 500.

    The payload names EVERY whitelisted field, and the coverage assertion below is what keeps that
    true: a field added to the whitelist later cannot slip past this test unexercised."""
    payload = {
        "project_id": _BUNDLE_PID,
        "docs_dir": str(_BUNDLE_DIR),
        "bundle_dir": str(_BUNDLE_DIR),
        "verdict_input": {"decision": "approved", "rationale": "expert reviewed (4e)"},
        # LOCAL, not the hosted default: the AZURE arm resolves a Foundry deployment name from the
        # model map BEFORE any client is built (``run.py`` stamps provenance with it), so it cannot
        # complete offline. The hosted default itself is pinned by the recorder test above.
        "profile": "local",
        "max_rounds": 2,
        "max_tokens": 100_000,
        "top_k": 3,
    }
    assert set(payload) == hosting._ALLOWED_FIELDS, (
        "the payload must exercise every whitelisted field — a field the whitelist accepts but "
        "this test never sends is a field no test proves ``run_project`` accepts"
    )

    status, body = await _post(served, "/invocations", payload)

    assert status == 200, f"a real run did not survive the hosted surface: {body}"
    assert body["outcome_type"] == "validated"
    # The validator's arithmetic, not a scripted string: P90 is computed from the proposal's own
    # cost line, and the claim (30 000) clears it.
    assert body["p90"] == _BUNDLE_P90
    assert body["proposal"]["claimed_saving_nok"] == 30000.0
    assert body["proposal"]["project_id"] == _BUNDLE_PID
    # Provenance carries a first-class citation from the NAVIGATED bundle — the run really read the
    # knowledge base rather than answering from the payload.
    assert body["provenance"]["citations"], "the run produced no citation — the bundle was not read"
    assert body["checker_verdict"] == "approve"
    assert _scripted_backend, "the scripted backend was never called — no real run happened"


def test_deployment_artifacts_pin_the_measured_platform_and_one_start_command() -> None:
    """RAW-TEXT GATE on the two files that decide whether the hosted image runs at all.

    Neither is executed by any test: ``docker build`` and ``azd deploy`` are operator-gated, so a
    regression in them is invisible to the whole suite until it fails in the cloud. What the gate
    pins is exactly what was MEASURED, and nothing about the prose around it:

    * ``--platform linux/amd64`` — the platform requires x86_64 (spike §1.4). Dropping it makes the
      image inherit the builder's architecture, which on an arm64 laptop yields an image that
      builds green locally and cannot start in the cloud. The flag lives in the documented build
      command (a Dockerfile cannot set the build platform for its own invocation), so pinning the
      documented string is the only gate available — and it is worth having precisely because
      nothing else re-derives it.
    * ONE copy of the start command: the image's ``CMD`` names ``main.py``, and ``azure.yaml``
      declares NO ``startupCommand``. Two copies drift (kø-(p)-regelen); this is the pair that
      keeps there being one.

    ``env:`` is checked for the same reason: ``FOUNDRY_PROJECT_ENDPOINT`` is injected by the
    platform, and redeclaring it here could shadow the injected value — which is the failure mode
    Fase 4b's endpoint precedence exists to avoid, undone from the config side."""
    dockerfile = _REPO_ROOT / "Dockerfile"
    azure_yaml = _REPO_ROOT / "azure.yaml"
    assert dockerfile.is_file(), "the hosted image has no build definition"
    assert azure_yaml.is_file(), "azd has no project definition to deploy"

    docker_text = dockerfile.read_text(encoding="utf-8")
    azure_text = azure_yaml.read_text(encoding="utf-8")

    assert "--platform linux/amd64" in docker_text, (
        "the Dockerfile no longer names the measured build platform; an image built without it "
        "inherits the builder's architecture and cannot start on the hosting platform"
    )
    assert "CMD" in docker_text and "main.py" in docker_text, (
        "the image's CMD is the ONE copy of the start command and must name main.py"
    )
    # Line-anchored: a mention inside a comment is prose, a top-level key is a declaration.
    azure_keys = [line.split(":")[0].strip() for line in azure_text.splitlines()]
    assert "startupCommand" not in azure_keys, (
        "azure.yaml declares a startupCommand — a SECOND copy of the start command, free to drift "
        "from the image's CMD"
    )
    assert "env" not in azure_keys, (
        "azure.yaml declares an env block — FOUNDRY_PROJECT_ENDPOINT is injected by the platform "
        "and must never be redeclared here"
    )
    assert "host: azure.ai.agent" in azure_text and "kind: hosted" in azure_text, (
        "azure.yaml no longer declares the hosted-agent host this whole entrypoint targets"
    )