portfolio-optimiser/tests/test_structured_output_loadbearing.py

"""Load-bearing: the proposer call must carry a STRICT structured-output schema (Fase 1b, funn 1b).

The gap, measured on the project's first live run
(``docs/2026-08-14-fase1b-forste-levende-kjoring.md``): ``generate_via_llm`` asked for the typed IR
in PROSE ("Respond with ONLY a JSON object ...") and nothing else. Twelve rounds burned on replies
that did not parse, the run died with ``BudgetExceeded``, and the validator was never reached with a
parseable candidate. ``e371890`` made the raw text visible; this seam removes the cause: the model is
handed a GRAMMAR, not a request.

**The wire form is decided by measurement, not from memory** (MAF 1.9.0, agent-framework-openai
1.8.2, agent-framework-foundry 1.8.2):

* ``ChatOptions`` carries ``response_format: type[BaseModel] | Mapping[str, Any] | None``, and BOTH
  wired profiles honour it. LOCAL (``OpenAIChatCompletionClient``) passes a Mapping through verbatim
  to Chat Completions; AZURE (``FoundryChatClient`` -> ``RawFoundryChatClient`` ->
  ``RawOpenAIChatClient``) converts the SAME mapping into the Responses API's ``text.format``. One
  mapping, both profiles.
* A ``type[BaseModel]`` would be the shorter spelling, and it is REJECTED here on evidence. The
  client then converts it with ``openai.lib._parsing._completions.type_to_response_format_param``,
  which (measured) emits ``minimum`` / ``exclusiveMinimum`` / ``minItems`` / ``prefixItems`` and an
  ``assumptions`` node whose ``additionalProperties`` is a SCHEMA. Azure's documented structured-
  output subset lists every one of those as unsupported and requires ``additionalProperties: false``
  in every object
  (https://learn.microsoft.com/azure/foundry/openai/how-to/structured-outputs). Passing our own
  mapping is the only way to control what actually goes on the wire.

**Stripping the constraints loses nothing**, and that is the point of the split: the schema's job is
SHAPE, the deterministic validator's job is VALUES. ``minItems``/``gt=0`` are re-applied by pydantic
in ``_parse_ir`` and by ``validate_proposal`` — which the module docstring already names as the
reliability mechanism.

**``assumptions`` may NOT simply be dropped, and that is a measurement too.** It is the one IR field
inexpressible in the strict subset (a free-form map of 2-tuples). Dropping it looks harmless because
the field is optional — but ``validator._monte_carlo`` samples ``item.unit_cost`` unchanged when a
code has no band, so with no bands at all the 512 samples are IDENTICAL and P10 == P50 == P90. The
stochastic falsifier would go inert while still reporting percentiles: a gate that can only be green,
which is this repo's cardinal defect class. So the schema carries the bands in a strict-legal
ARRAY-of-entries form and ``_parse_ir`` normalises them back to the IR's map — the IR itself is
untouched.

Five tests, load-bearing as a set. Each detach point is RED on its own:

* T1 — the WIRING: the response_format reaches the client's ``options`` on the generation call
  (RED when ``options=`` is dropped, or when a bare pydantic class is passed instead of the mapping);
* T2 — the schema is inside the DOCUMENTED subset, recursively, with a paired CONTROL proving the
  same walker finds those keywords in the RAW pydantic schema (RED when the sanitiser is detached —
  and non-vacuous, because the control proves the walker reaches nested ``$defs`` at all);
* T3 — the Monte Carlo falsifier SURVIVES: a wire reply with bands yields P90 > P10, with a control
  proving a band-less reply collapses to P90 == P10 (RED when ``assumptions`` is dropped from the
  schema or the bands do not reach the IR);
* T4 — FAIL-CLOSED: a model carrying an inexpressible node with no declared override RAISES rather
  than emitting an illegal schema (RED on a sanitiser that silently skips what it cannot express);
* T5 — the round trip is VERBATIM and ADDITIVE: array-form bands reach the IR with their exact
  values, and the IR's own map form is still accepted unchanged.
"""

from __future__ import annotations

import json
from collections.abc import Iterator, Mapping, Sequence
from typing import Any

import pytest
from agent_framework import ChatResponse, ChatResponseUpdate, Message
from agent_framework._types import ResponseStream
from pydantic import BaseModel

from portfolio_optimiser.budget import Budget, TokenMeter
from portfolio_optimiser.generate import (
    UNSUPPORTED_SCHEMA_KEYWORDS,
    StructuredOutputUnsupported,
    generate_via_llm,
    proposal_response_format,
    strict_json_schema,
)
from portfolio_optimiser.ir import SavingsProposal
from portfolio_optimiser.reference_domain import load_reference_projects
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.validator import ValidatedProposal

#: The instruction line ``generate._build_messages`` puts in EVERY generation prompt and nowhere
#: else — the one identifier that separates a generation call from a debate turn.
_GENERATION_MARK = "Respond with ONLY a JSON object"

#: One affected line: 100 x 10 = 1000 total -> nominal feasible = 0.30 * 1000 = 300.
#: A band of (8, 12) encloses the unit_cost (the IR's own model_validator requires that) and makes
#: the sampled totals range over 800..1200, i.e. feasible 240..360. A claim of 200 therefore clears
#: BOTH the P90 stage and the nominal stage whether or not bands are present — so the ONLY thing
#: that differs between T3's positive and its control is the band itself.
_CODE = "STRUCTURED-OUT-LINE"
_QUANTITY = 100.0
_UNIT_COST = 10.0
_BAND_LOW = 8.0
_BAND_HIGH = 12.0
_CLAIM = 200.0
#: NOT ``energy_efficiency`` — that measure would additionally hit the method cap (0.15 * 1000 = 150)
#: and reject a claim of 200 for a reason that has nothing to do with this seam.
_MEASURE = "behovsstyrt_drift"
#: P7: the synthetic context must NAME the line the synthetic proposal restates, exactly as a
#: real run's input does. Without it the grounding stage rejects a sentinel code no input ever
#: mentioned — correctly, and for a reason that has nothing to do with this seam.
_CONTEXT = f"Price schedule line {_CODE} covers demand-driven operation."


def _wire_reply(*, with_band: bool) -> str:
    """A reply in exactly the shape the strict schema commissions: ``assumptions`` is an ARRAY of
    entries, never the IR's map. Both arms are byte-identical apart from that array."""
    bands = (
        [{"code": _CODE, "low_unit_cost": _BAND_LOW, "high_unit_cost": _BAND_HIGH}]
        if with_band
        else []
    )
    return json.dumps(
        {
            "project_id": "FV42-GSV-E1",
            "measure": _MEASURE,
            "affected_items": [{"code": _CODE, "quantity": _QUANTITY, "unit_cost": _UNIT_COST}],
            "claimed_saving_nok": _CLAIM,
            "assumptions": bands,
        }
    )


def _meter() -> TokenMeter:
    # Caps well above what one attempt needs, so max_attempts -- not the budget -- is the bound.
    return TokenMeter(Budget(max_tokens=10**9, max_rounds=20))


class _OptionsRecordingChatClient(ScriptedChatClient):
    """Records the ``options`` mapping of every call, then DELEGATES to the canonical scripted body.

    This double cannot live in the ``reply_selector`` seam: that seam is handed
    ``(prompt_blob, role)`` and returns a reply string, and ``options`` — this double's entire
    subject — never reaches it. Registered in ``test_scripted_client_consolidation``'s
    ``_DELEGATING_OVERRIDES`` for exactly that reason.
    """

    def __init__(self, reply: str) -> None:
        super().__init__(reply)
        self.seen_options: list[Mapping[str, Any]] = []

    def _inner_get_response(
        self,
        *,
        messages: Sequence[Message],
        options: Mapping[str, Any],
        stream: bool = False,
        **kwargs: Any,
    ) -> Any | ResponseStream[ChatResponseUpdate, ChatResponse]:
        blob = " ".join(getattr(m, "text", "") or "" for m in messages)
        if _GENERATION_MARK in blob:
            self.seen_options.append(dict(options))
        return super()._inner_get_response(
            messages=messages, options=options, stream=stream, **kwargs
        )


def _sub_schemas(node: Any) -> Iterator[Mapping[str, Any]]:
    """Every mapping in the schema tree, including those under ``$defs`` — the walker both T2 and
    its control run, so the control genuinely proves the walker's reach."""
    if isinstance(node, Mapping):
        yield node
        for value in node.values():
            yield from _sub_schemas(value)
    elif isinstance(node, list):
        for value in node:
            yield from _sub_schemas(value)


def _keywords_found(schema: Any) -> set[str]:
    return {
        key for node in _sub_schemas(schema) for key in node if key in UNSUPPORTED_SCHEMA_KEYWORDS
    }


async def test_generation_call_carries_the_strict_schema() -> None:
    """T1 LOAD-BEARING: the strict response_format reaches the client on the generation call.

    RED when ``generate_via_llm`` stops passing ``options=``, and RED when it passes a bare
    ``type[BaseModel]`` instead of the sanitised mapping (the assert is on the mapping's identity,
    not merely on the key's presence)."""
    project = load_reference_projects()[0]
    client = _OptionsRecordingChatClient(_wire_reply(with_band=True))

    await generate_via_llm(client, project, _CONTEXT, _meter(), max_attempts=1)

    # Control FIRST: a positive assert over an empty list would pass vacuously.
    assert client.seen_options, "no generation call was observed — the assert below proves nothing"
    assert client.seen_options[0].get("response_format") == proposal_response_format(), (
        "the generation call did not carry the strict structured-output schema"
    )


def test_schema_stays_inside_the_documented_strict_subset() -> None:
    """T2 LOAD-BEARING: the emitted schema uses only what Azure's structured-output subset allows.

    RED when the sanitiser is detached (the raw pydantic schema's keywords come straight through).
    The CONTROL is what makes it non-vacuous: the SAME walker must FIND those keywords in the raw
    schema — including inside ``$defs``, where ``AffectedItem``'s constraints live. Without it, a
    walker that silently visits nothing would make every assert below green."""
    raw = SavingsProposal.model_json_schema()
    raw_found = _keywords_found(raw)
    assert raw_found, (
        "control failed: the walker found NO unsupported keyword in the raw pydantic schema, so "
        "the positive assert below cannot distinguish a working sanitiser from a dead walker"
    )
    assert "$defs" in raw and _keywords_found(raw["$defs"]), (
        "control failed: the walker does not reach nested $defs, where AffectedItem's constraints "
        "live — a sanitiser that skipped $defs would still pass"
    )

    schema = proposal_response_format()["json_schema"]["schema"]
    assert _keywords_found(schema) == set(), (
        f"emitted schema carries keywords outside the documented strict subset: "
        f"{sorted(_keywords_found(schema))}"
    )
    for node in _sub_schemas(schema):
        if "properties" not in node:
            continue
        assert node.get("additionalProperties") is False, (
            f"object node without ``additionalProperties: false``: {sorted(node.get('properties'))}"
        )
        assert set(node.get("required", [])) == set(node["properties"]), (
            f"strict mode requires EVERY property listed as required; node "
            f"{sorted(node['properties'])} lists {sorted(node.get('required', []))}"
        )


async def test_assumption_bands_keep_the_monte_carlo_falsifier_alive() -> None:
    """T3 LOAD-BEARING: the commissioned schema can still carry uncertainty bands, and they reach
    the Monte Carlo.

    Two halves, because a scripted reply cannot be constrained by the schema the way a live model
    is — so the schema half must be asserted DIRECTLY or this test could not tell a schema that
    commissions bands from one that does not:

    * the schema still COMMISSIONS the bands (RED when ``assumptions`` is dropped from it — which
      would leave a live model unable to supply a band at all);
    * a band that does arrive REACHES the Monte Carlo (RED when the array form is not folded into
      the IR).

    The CONTROL is the band-less arm — it proves the spread asserted below is caused by the band
    and not by the Monte Carlo being noisy in general."""
    project = load_reference_projects()[0]

    schema = proposal_response_format()["json_schema"]["schema"]
    assert "assumptions" in schema["properties"], (
        "the schema no longer commissions uncertainty bands — a live model constrained by it could "
        "not supply one, and the Monte Carlo would be degenerate on every generated proposal"
    )
    assert "assumptions" in schema["required"]

    banded = await generate_via_llm(
        _OptionsRecordingChatClient(_wire_reply(with_band=True)),
        project,
        _CONTEXT,
        _meter(),
        max_attempts=1,
    )
    bandless = await generate_via_llm(
        _OptionsRecordingChatClient(_wire_reply(with_band=False)),
        project,
        _CONTEXT,
        _meter(),
        max_attempts=1,
    )

    assert isinstance(banded.outcome, ValidatedProposal), f"fixture: {banded.outcome}"
    assert isinstance(bandless.outcome, ValidatedProposal), f"fixture: {bandless.outcome}"

    # The control: with no band the falsifier is degenerate — this is the state the seam must NOT
    # silently ship.
    assert bandless.outcome.p90 == bandless.outcome.p10, (
        "control failed: a band-less proposal already spreads, so the spread asserted below would "
        "not prove the band arrived"
    )
    assert banded.outcome.p90 > banded.outcome.p10, (
        "the uncertainty band did not reach the Monte Carlo — the stochastic falsifier is inert"
    )


def test_sanitiser_is_fail_closed_on_an_inexpressible_node() -> None:
    """T4 LOAD-BEARING: a node the strict subset cannot express RAISES rather than being emitted or
    silently skipped.

    Without this, a field added to the IR later would quietly produce an illegal schema and turn
    every live generation call into a 400 — or, worse, be dropped so the model is never asked for
    it. Validation, never repair (mirrors ``write_concept_file`` / ``promote_verdict``)."""

    class _FreeFormMap(BaseModel):
        label: str
        bands: dict[str, float]  # free-form map: additionalProperties is a SCHEMA, not ``false``

    with pytest.raises(StructuredOutputUnsupported):
        strict_json_schema(_FreeFormMap)

    class _Expressible(BaseModel):
        label: str
        count: int

    # Control: the raise above is caused by the inexpressible node, not by the sanitiser rejecting
    # everything it is handed.
    assert strict_json_schema(_Expressible)["properties"].keys() == {"label", "count"}


async def test_band_round_trip_is_verbatim_and_the_ir_map_form_still_parses() -> None:
    """T5 LOAD-BEARING: the array form reaches the IR with its EXACT values, and the change is
    ADDITIVE — the IR's own map form is still accepted.

    RED when the normalisation is detached (the array never becomes a map), and RED when it is
    written as a REPLACEMENT rather than an addition (the map form would stop parsing, which would
    break every existing scripted reply in the suite)."""
    project = load_reference_projects()[0]

    from_array = await generate_via_llm(
        _OptionsRecordingChatClient(_wire_reply(with_band=True)),
        project,
        _CONTEXT,
        _meter(),
        max_attempts=1,
    )
    assert isinstance(from_array.outcome, ValidatedProposal)
    assert from_array.outcome.proposal.assumptions == {_CODE: (_BAND_LOW, _BAND_HIGH)}, (
        "the array-form band did not reach the IR verbatim"
    )

    map_form = json.dumps(
        {
            "project_id": "FV42-GSV-E1",
            "measure": _MEASURE,
            "affected_items": [{"code": _CODE, "quantity": _QUANTITY, "unit_cost": _UNIT_COST}],
            "claimed_saving_nok": _CLAIM,
            "assumptions": {_CODE: [_BAND_LOW, _BAND_HIGH]},
        }
    )
    from_map = await generate_via_llm(
        _OptionsRecordingChatClient(map_form), project, _CONTEXT, _meter(), max_attempts=1
    )
    assert isinstance(from_map.outcome, ValidatedProposal), (
        "the IR's own map form stopped parsing — the normalisation replaced rather than extended"
    )
    assert from_map.outcome.proposal.assumptions == {_CODE: (_BAND_LOW, _BAND_HIGH)}