feat(1b): proposeren får en grammatikk — strict structured output [skip-docs]

Fase 1b, funn 1b. Den første levende kjøringen brant tolv runder på svar som
ikke lot seg parse til IR-formen; e371890 gjorde teksten synlig, dette fjerner
årsaken. generate_via_llm sender nå
options={"response_format": proposal_response_format()} på hvert
genererings-kall.

Formen er MÅLT, ikke valgt. ChatOptions.response_format tar
type[BaseModel] | Mapping, og begge profiler ærer den: LOCAL sender en Mapping
ordrett til Chat Completions, AZURE (FoundryChatClient -> RawFoundryChatClient
-> RawOpenAIChatClient) konverterer samme envelope til Responses-APIets
text.format. Klassen — det korteste svaret — er avvist på bevis: gitt en klasse
konverterer klienten med type_to_response_format_param, som emitterer
minimum/exclusiveMinimum/minItems/prefixItems og et assumptions-node hvis
additionalProperties er et skjema. Azures publiserte subset utelukker alle fire.

assumptions kan ikke bare droppes, og det er også en måling: validator
._monte_carlo faller tilbake på item.unit_cost for hver kode uten bånd, så uten
bånd er alle 512 samples identiske og P10 == P50 == P90. Den stokastiske
falsifisereren ville gått inert mens den fortsatt rapporterte persentiler.
Wire-en bærer derfor et array av navngitte entries som _parse_ir folder tilbake
til IR-ens map — additivt, aldri erstatning. Skjemaet deriveres fra
SavingsProposal; sanitiseren er fail-closed (StructuredOutputUnsupported).

Load-bearing målt mot hele suiten, seks mutasjoner alle røde, grønn kontroll
864/4: detach wiringen (1) · detach sanitiseren (3) · dropp assumptions fra
skjemaet (1) · fail-closed -> stille reparasjon (1) · detach normaliseringen
(3) · erstatning i stedet for tillegg (2, inkl. golden-transkriptet).

T3 ble skrevet vakuøs først og felt av sin egen mutasjon: den påsto å bli rød
når assumptions forsvant fra skjemaet, men den scriptede klienten ignorerer
skjemaet. Testen fikk en direkte assert på skjemaet.

Ærlighets-grense: ingen betalt kjøring gjort. Testene beviser konformitet med
det dokumenterte subsettet, ikke aksept fra det levende endepunktet.

859 -> 864 passed / 4 skipped. ruff + format + mypy rene.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013EQNU4tfAhsBvdefT1jUhk
This commit is contained in:
Kjell Tore Guttormsen 2026-08-14 14:00:32 +02:00
commit 642ce8ae9a
4 changed files with 612 additions and 3 deletions

View file

@ -555,6 +555,44 @@ Python ≥3.10. MAF (`agent-framework-core` 1.9.0). Pakkehåndtering: `uv`. To b
trunker til 100 tegn slik at sentinelen OVERLEVER (1 rød — verbatim-asserten alene, den skarpe
diskriminatoren). Ærlighets-grense: `_charge_usage` kan raise FØR parse, og et svar tapt der er
ikke en parse-feil og fanges ikke.
- **Proposeren får en GRAMMATIKK, og skjemaet er DERIVERT + fail-closed (Fase 1b, funn 1b):**
`generate_via_llm` sender `options={"response_format": proposal_response_format()}` på hvert
genererings-kall. **Formen er MÅLT, ikke valgt:** `ChatOptions.response_format` tar
`type[BaseModel] | Mapping`, og BEGGE profiler ærer den — LOCAL
(`OpenAIChatCompletionClient`) sender en Mapping ordrett til Chat Completions, AZURE
(`FoundryChatClient``RawFoundryChatClient``RawOpenAIChatClient`) konverterer SAMME
envelope til Responses-APIets `text.format`. **Klassen er AVVIST på bevis:** gitt en klasse
konverterer klienten med `type_to_response_format_param`, som (målt) emitterer `minimum` /
`exclusiveMinimum` / `minItems` / `prefixItems` og et `assumptions`-node hvis
`additionalProperties` er et SKJEMA — fire ting Azures publiserte subset utelukker
(Learn: «Unsupported type-specific keywords» + `additionalProperties: false` i hvert objekt).
Vår egen mapping er eneste måte å styre hva som når tråden. **Å stripe beskrankningene koster
ingenting:** skjemaets jobb er FORM, validatorens jobb er VERDIER — `minItems`/`gt=0` gjenreises
av pydantic i `_parse_ir` og av `validate_proposal`. **Skjemaet DERIVERES fra `SavingsProposal`**
(`strict_json_schema`), aldri håndskrevet: en andre kopi av en form som alt bor i `ir.py` drifter
stille, og modellen ville fortsatt blitt bestilt for den gamle. **`assumptions` KAN IKKE bare
droppes, og det er en MÅLING:** feltet er det ene uttrykksløse (fri-form map av 2-tupler), men
`validator._monte_carlo` faller tilbake på `item.unit_cost` for hver kode uten bånd — uten bånd
i det hele tatt er alle 512 samples IDENTISKE og P10 == P50 == P90. Den stokastiske
falsifisereren ville gått inert mens den fortsatt rapporterte persentiler: repoets kardinalklasse
(en gate som bare kan bli grønn). Derfor bærer WIRE-en et array av navngitte entries og
`_parse_ir` folder det tilbake til IR-ens map — **additivt, aldri erstatning** (map-formen
parser uendret; alle scriptede svar i suiten og golden-transkriptet bruker den). Sanitiseren er
**fail-closed** (`StructuredOutputUnsupported`) på `prefixItems`/`oneOf`/`allOf`/fri-form map
uten deklarert override — validering, ALDRI reparasjon (speiler `write_concept_file`).
Prompt-linja «Respond with ONLY a JSON object» + parse-retry + funn-1-fangsten står URØRT: en
leverandør som ignorerer `response_format` må fortsatt få beskjed, og backstoppen er poenget.
Load-bearing MÅLT (`tests/test_structured_output_loadbearing.py`), seks mutasjoner alle røde +
grønn kontroll 864/4: detach wiringen (1 rød) · detach sanitiseren (3 røde) · dropp
`assumptions` fra skjemaet (1 rød) · fail-closed → stille reparasjon (1 rød) · detach
normaliseringen (3 røde) · erstatning i stedet for tillegg (2 røde — T5 PLUSS
golden-transkriptet, et uavhengig vitne). **T3 ble skrevet VAKUØS først** (repoets 08-09-klasse,
sjette gang): den påsto å bli rød når `assumptions` forsvant fra skjemaet, men den scriptede
klienten ignorerer skjemaet — påstanden ble bevist usann av M3 og testen fikk en DIREKTE
assert på skjemaet. **ÆRLIGHETS-GRENSE, UTTALT:** ingen betalt kjøring er gjort, så at det
emitterte skjemaet ER akseptert av det levende endepunktet er IKKE verifisert — testen beviser
konformitet med det DOKUMENTERTE subsettet, ikke aksept. Ollamas oppførsel på
`response_format` er likeledes uverifisert.
- **Overleverings-pakka ER `git archive HEAD`, aldri en kuratert kopi (Fase 5):**
`scripts/make-handover-package.sh` bygger én zip en ekstern organisasjon deployer uten å klone
repoet. **Tracked files only er hele eksponerings-kontrollen**`STATE.md`, `*.local.md` og

View file

@ -22,11 +22,12 @@ Two entry points, because the LLM call is async while ``validator.self_repair``
from __future__ import annotations
import json
from collections.abc import Callable
from collections.abc import Callable, Mapping
from dataclasses import dataclass, field
from typing import Any
from agent_framework import BaseChatClient, Message
from pydantic import ValidationError
from pydantic import BaseModel, ValidationError
from portfolio_optimiser.budget import TokenMeter
from portfolio_optimiser.ir import CostBaseline, SavingsProposal
@ -44,6 +45,195 @@ class GenerationError(RuntimeError):
"""No parseable proposal could be produced within the attempt budget."""
class StructuredOutputUnsupported(TypeError):
"""A schema node cannot be expressed in the provider's strict structured-output subset.
Fail-closed, and deliberately so (mirrors ``write_concept_file`` / ``promote_verdict``:
validation, never repair). The alternative silently dropping what cannot be expressed would
stop commissioning a field without saying so, and the field it would have dropped first is
``assumptions``, whose absence makes the Monte Carlo falsifier inert while it still reports
percentiles. A schema this module cannot express is a decision for a human, not a default.
"""
#: Type-specific JSON Schema keywords the provider's structured-output subset does NOT support,
#: transcribed from Azure's published table (Structured outputs -> "Unsupported type-specific
#: keywords", https://learn.microsoft.com/azure/foundry/openai/how-to/structured-outputs), which
#: states it is the same subset OpenAI accepts.
#:
#: ``exclusiveMinimum``/``exclusiveMaximum`` are NOT literally in that table — it names
#: ``minimum maximum multipleOf`` — but they are the same family, and pydantic emits them for
#: ``Field(gt=...)``/``Field(lt=...)``, which is exactly how this repo's IR spells its bounds. Being
#: stricter than the table costs nothing here: every constraint stripped is re-applied by pydantic in
#: ``_parse_ir`` and by ``validate_proposal``. The schema's job is SHAPE; the validator's job is
#: VALUES. ``default`` is stripped for a different reason — strict mode requires every property to be
#: required, so a default can never apply.
UNSUPPORTED_SCHEMA_KEYWORDS = frozenset(
{
# String
"minLength",
"maxLength",
"pattern",
"format",
# Number
"minimum",
"maximum",
"multipleOf",
"exclusiveMinimum",
"exclusiveMaximum",
# Objects
"patternProperties",
"unevaluatedProperties",
"propertyNames",
"minProperties",
"maxProperties",
# Arrays
"unevaluatedItems",
"contains",
"minContains",
"maxContains",
"minItems",
"maxItems",
"uniqueItems",
# Meaningless once every property is required
"default",
}
)
#: The strict-legal stand-in for ``SavingsProposal.assumptions``.
#:
#: The IR spells the uncertainty bands as ``dict[str, tuple[float, float]]`` — a free-form map whose
#: values are tuples. Neither half is expressible: strict mode requires ``additionalProperties:
#: false`` in every object (so a map with arbitrary keys cannot be described), and tuples arrive as
#: ``prefixItems``, which is outside the supported type list. Dropping the field instead would be
#: silent damage: ``validator._monte_carlo`` falls back to the item's stated ``unit_cost`` for every
#: code with no band, so with no bands at all the samples are identical and P10 == P50 == P90 — the
#: stochastic falsifier goes inert while still reporting percentiles.
#:
#: So the WIRE carries an array of named entries and ``_parse_ir`` folds it back into the IR's map.
#: The IR itself is untouched; the entry names spell out what the tuple positions mean, which the
#: model would otherwise have to guess.
_ASSUMPTIONS_WIRE_NODE: dict[str, Any] = {
"type": "array",
"description": (
"Uncertainty band per affected cost line: the low and high unit cost the true price is "
"expected to fall between. The band MUST enclose that item's own unit_cost. Omit an entry "
"for a line whose unit cost is certain; an empty list means no uncertainty is claimed."
),
"items": {
"type": "object",
"properties": {
"code": {"type": "string"},
"low_unit_cost": {"type": "number"},
"high_unit_cost": {"type": "number"},
},
},
}
#: Dotted paths (from the root model's own properties) whose node is replaced before sanitising.
_PROPOSAL_SCHEMA_OVERRIDES: Mapping[str, dict[str, Any]] = {"assumptions": _ASSUMPTIONS_WIRE_NODE}
def _sanitise_schema_node(node: Any, *, path: str, overrides: Mapping[str, dict[str, Any]]) -> Any:
"""Rewrite one JSON Schema node into the strict subset, or raise ``StructuredOutputUnsupported``.
An override is applied FIRST, so a declared replacement is what gets checked and emitted that
is how the one inexpressible node in this repo's IR (``assumptions``) is expressed rather than
excused. The replacement is then sanitised by the same code as everything else, so an override
cannot smuggle in an illegal node.
"""
if not isinstance(node, Mapping):
return node
if path in overrides:
node = overrides[path]
if "prefixItems" in node:
raise StructuredOutputUnsupported(
f"{path or '<root>'}: tuple types (prefixItems) are outside the strict subset"
)
for combinator in ("oneOf", "allOf"):
if combinator in node:
raise StructuredOutputUnsupported(
f"{path or '<root>'}: {combinator} is outside the strict subset (anyOf is the "
"only supported combinator)"
)
if isinstance(node.get("additionalProperties"), Mapping):
raise StructuredOutputUnsupported(
f"{path or '<root>'}: a free-form map cannot be expressed — strict mode requires "
"additionalProperties: false in every object. Declare an override that spells the "
"entries out as an array."
)
out: dict[str, Any] = {}
for key, value in node.items():
if key in UNSUPPORTED_SCHEMA_KEYWORDS:
continue
if key == "properties" and isinstance(value, Mapping):
out[key] = {
name: _sanitise_schema_node(
sub, path=f"{path}.{name}" if path else name, overrides=overrides
)
for name, sub in value.items()
}
elif key == "$defs" and isinstance(value, Mapping):
out[key] = {
name: _sanitise_schema_node(sub, path=f"$defs.{name}", overrides=overrides)
for name, sub in value.items()
}
elif key == "items":
out[key] = _sanitise_schema_node(value, path=f"{path}[]", overrides=overrides)
elif key == "anyOf" and isinstance(value, list):
out[key] = [_sanitise_schema_node(sub, path=path, overrides=overrides) for sub in value]
else:
out[key] = value
if "properties" in out:
# Strict mode's two structural demands, applied to EVERY object rather than the root only:
# no undeclared keys, and every declared key required.
out["additionalProperties"] = False
out["required"] = sorted(out["properties"])
return out
def strict_json_schema(
model: type[BaseModel], *, overrides: Mapping[str, dict[str, Any]] | None = None
) -> dict[str, Any]:
"""Derive a strict-structured-output schema from ``model``'s own pydantic schema.
DERIVED rather than hand-written on purpose: a hand-written copy of a shape that already exists
in ``ir.py`` is the second copy that drifts (-(p)), and it drifts silently the model would
keep being commissioned for the old shape. ``$defs``/``$ref`` are kept (the published subset
supports definitions), so nested models need no inlining.
"""
schema = _sanitise_schema_node(model.model_json_schema(), path="", overrides=overrides or {})
assert isinstance(schema, dict) # a model's root schema is always an object
return schema
def proposal_response_format() -> dict[str, Any]:
"""The ``response_format`` mapping commissioning a ``SavingsProposal`` from the proposer.
A MAPPING, not the ``type[BaseModel]`` the option also accepts, and the reason is measured: given
a class, the client converts it with ``type_to_response_format_param``, which emits ``minimum`` /
``exclusiveMinimum`` / ``minItems`` / ``prefixItems`` and an ``assumptions`` node whose
``additionalProperties`` is a schema four things the published subset rules out. Our own
mapping is the only way to control what reaches the wire.
ONE mapping serves both wired profiles (measured against agent-framework-openai 1.8.2 /
agent-framework-foundry 1.8.2): the Chat Completions client passes it through verbatim, and the
Responses client which ``FoundryChatClient`` delegates to converts this exact envelope into
``text.format``.
"""
return {
"type": "json_schema",
"json_schema": {
"name": SavingsProposal.__name__,
"strict": True,
"schema": strict_json_schema(SavingsProposal, overrides=_PROPOSAL_SCHEMA_OVERRIDES),
},
}
@dataclass(frozen=True)
class ParseFailure:
"""One model reply that did NOT parse into the typed IR, kept VERBATIM (Fase 1b, funn 1).
@ -136,6 +326,37 @@ def _build_messages(
return [Message(role="user", contents=[prompt])]
def _normalise_assumptions(data: dict[str, Any]) -> None:
"""Fold the WIRE's array-of-entries assumption bands back into the IR's ``code -> (low, high)``
map, in place.
ADDITIVE, never a replacement: a reply that already uses the IR's map form (every scripted reply
in the suite, and any model that answers without honouring the schema) is left untouched. A
malformed entry is raised as ``ValueError`` rather than ``KeyError`` on purpose ``ValueError``
is what ``_fetch_parsed`` catches, so a bad band is captured as the parse failure it is instead
of escaping the loop and killing the run.
"""
entries = data.get("assumptions")
if not isinstance(entries, list):
return
bands: dict[str, tuple[Any, Any]] = {}
for entry in entries:
if (
not isinstance(entry, Mapping)
or not {
"code",
"low_unit_cost",
"high_unit_cost",
}
<= entry.keys()
):
raise ValueError(
f"each assumption entry needs code, low_unit_cost and high_unit_cost; got {entry!r}"
)
bands[entry["code"]] = (entry["low_unit_cost"], entry["high_unit_cost"])
data["assumptions"] = bands
def _parse_ir(text: str, project: Project) -> SavingsProposal:
"""Parse the model's structured reply into the typed IR. Raises on malformed/text-leaked
output (JSON error or Pydantic ``ValidationError``)."""
@ -143,6 +364,7 @@ def _parse_ir(text: str, project: Project) -> SavingsProposal:
if not isinstance(data, dict):
raise ValueError("reply is not a JSON object")
data.setdefault("project_id", project.id)
_normalise_assumptions(data)
return SavingsProposal(**data)
@ -231,7 +453,13 @@ async def generate_via_llm(
# Parse-robust: a malformed/text-leaked reply is retried; the meter caps total work.
while True:
meter.tick_round() # between-attempt bound (BudgetExceeded over cap)
reply = await chat_client.get_response(messages) # non-streaming
# Fase 1b, funn 1b: hand the model a GRAMMAR, not a prose request. The prompt's
# "Respond with ONLY a JSON object" line stays — a provider that ignores
# ``response_format`` (or a local model that does not implement it) must still be told
# what is wanted, and the parse-retry below remains the backstop either way.
reply = await chat_client.get_response( # non-streaming
messages, options={"response_format": proposal_response_format()}
)
_charge_usage(meter, reply)
try:
return _parse_ir(reply.text, project)

View file

@ -39,6 +39,11 @@ _DELEGATING_OVERRIDES = [
# It cannot live in the reply-selector seam either — that seam returns a reply STRING, and a
# response that is not text is precisely this double's subject.
"tests/test_b4_mcp_call_trace_loadbearing.py",
# Fase 1b structured-output probe: records the ``options`` mapping of every generation call,
# then delegates. It cannot live in the reply-selector seam either — that seam is handed
# ``(prompt_blob, role)`` and returns a reply string, and ``options`` (this double's whole
# subject) never reaches it.
"tests/test_structured_output_loadbearing.py",
]
# Doubles in a DIFFERENT lineage (``spikes._harness.FakeChatClient``). There is no canonical

View file

@ -0,0 +1,338 @@
"""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"
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, "", _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,
"",
_meter(),
max_attempts=1,
)
bandless = await generate_via_llm(
_OptionsRecordingChatClient(_wire_reply(with_band=False)),
project,
"",
_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,
"",
_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, "", _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)}