"""Strict bilingual response contract and hallucination validation.""" from __future__ import annotations import logging import re from decimal import ROUND_HALF_UP, Decimal, DivisionByZero, InvalidOperation from typing import Any, Literal, get_args from pydantic import BaseModel, ConfigDict, Field, ValidationError from app.intelligence.privacy import UnsafeFactPackage, validate_anonymous_refs logger = logging.getLogger(__name__) class IntelligenceContractError(ValueError): """The model returned syntactically valid JSON that is not grounded.""" # Per-item ceiling on anonymous drill-down anchors. Declared once and used # BOTH by the schema below and by ``_trim_evidence_refs`` — two copies of # this number would drift, and a trimmer that trims to a different limit # than the schema enforces is a rejection with extra steps. _EVIDENCE_REFS_MAX = 20 # Separator for flattened citable keys ("costs.month_cost"). Defined here, # next to the resolver that has to understand it, and imported by the # flattener that produces it — the writer and the reader of an address # format must not each keep their own copy of it. FLAT_KEY_SEPARATOR = "." class GroundedNarrative(BaseModel): model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) title_zh: str = Field(min_length=1, max_length=240) title_en: str = Field(min_length=1, max_length=240) detail_zh: str = Field(min_length=1, max_length=2000) detail_en: str = Field(min_length=1, max_length=2000) fact_refs: list[str] = Field(min_length=1, max_length=12) evidence_refs: list[str] = Field( default_factory=list, max_length=_EVIDENCE_REFS_MAX, ) confidence: Literal["low", "medium", "high"] class RiskNarrative(GroundedNarrative): severity: Literal["low", "medium", "high", "critical"] class RecommendationNarrative(GroundedNarrative): priority: Literal["p0", "p1", "p2", "p3"] # ``assign`` earns its place on production evidence: the daily brief was # refused outright because the narrator wanted to recommend assigning an # owner, and giving an unowned resource an owner is a first-class Helix # flow (auto-Owner backfill, cloud-scope Owner, server owner) rather than # a word the model happened to invent. The category stays a category — # the model still supplies no URL, operation, approver or payload. action_kind: Literal[ "review", "reclaim", "transfer", "retain", "purchase", "configure", "approve", "notify", "investigate", "assign", ] class ForecastNarrative(GroundedNarrative): horizon_days: Literal[30, 60, 90] class IntelligenceOutput(BaseModel): """One complete bilingual analysis. No untyped extension dictionaries.""" model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) summary_zh: str = Field(min_length=1, max_length=6000) summary_en: str = Field(min_length=1, max_length=6000) # A citation cap BELOW the size of the fact package forbids citing what # the model was handed. The cross-domain brief publishes 8 aggregate # facts plus 8 dimensions — about 24 citable scalars — and was refused # in production for listing 22 against a cap of 20. Sized to clear the # largest real package with room, still bounded so a summary cannot cite # an unbounded list. summary_fact_refs: list[str] = Field(min_length=1, max_length=40) conclusions: list[GroundedNarrative] = Field(default_factory=list, max_length=12) risks: list[RiskNarrative] = Field(default_factory=list, max_length=20) opportunities: list[GroundedNarrative] = Field(default_factory=list, max_length=20) forecasts: list[ForecastNarrative] = Field(default_factory=list, max_length=3) recommendations: list[RecommendationNarrative] = Field(default_factory=list, max_length=20) data_gaps: list[GroundedNarrative] = Field(default_factory=list, max_length=20) # ASCII identifier boundaries deliberately do not treat CJK characters as # word boundaries: ``成本1200元`` must be scanned, while the numeric suffix in # an anonymous reference such as ``srv_01`` must not become a cited amount. _NUMBER = re.compile( r"(? dict[str, str]: """Map each record's self-published name to the address it lives at. Production, 2026-08-25: the brief was refused for ``/facts/cloud_network.resource_waste_rate.value``, and the payload really does publish that metric — at ``dimensions.cloud_network.metrics.resource_waste_rate.value``, with ``...resource_waste_rate.key == "cloud_network.resource_waste_rate"`` sitting right beside it. We hand the model the record's name and then answer only to its address, and the two differ for **all 40** metrics in the platform package (the flattener drops the dimension namespace the catalog key repeats, and inserts the ``metrics`` segment the key does not have). Citing a metric by the name we published for it could never succeed. So a record that publishes its own name is addressable by that name. Nothing is inferred: the translated address must exist in the same member, so a published name plus an invented leaf still resolves to nothing — ``cloud_network.resource_waste_rate.trend`` is refused just as it was before. A name published by two records at two addresses is dropped rather than guessed; two facts must never share one citable address (see ``_assign_flat`` in ``generation``). Recomputed per lookup, and only on the fallback path — a citation that matched an address exactly never reaches here, so a healthy output pays nothing for it. """ addresses: dict[str, str] = {} ambiguous: set[str] = set() for flat_key, value in member.items(): head, sep, leaf = flat_key.rpartition(FLAT_KEY_SEPARATOR) if not sep or leaf != _RECORD_NAME_FIELD or not isinstance(value, str): continue name = value.strip() # A record already addressed by its own name (a dimension) needs no # alias, and an empty name is not an address. if not name or name == head: continue if addresses.setdefault(name, head) != head: ambiguous.add(name) for name in ambiguous: addresses.pop(name, None) return addresses def _flat_lookup(document: Any, key: str) -> Any: """Resolve one citable address, or ``_MISSING``. Four production rounds taught the same lesson four times: the model reaches for a value the payload really publishes, and writes the member path wrong. It cited ``.0.metrics.1.value`` (the leading separator we printed), then ``/facts/cloud_network.metrics.1.value`` (a dimension addressed as a fact), then ``/facts/overall_score`` (a top-level member addressed as a fact). Each rejection threw away an entire grounded brief, and each spelling-specific patch only invited the next spelling. So resolution is by ADDRESS, not by path: strip the decoration the model may have added — a pointer slash, a member name, a leading separator — and ask whether the payload publishes that name. It must still publish it, under exactly that name, or this returns ``_MISSING`` and the citation is refused. That is the grounding invariant; which member the model thought it lived in is not. Precedence is ``facts`` → ``dimensions`` → top level: narratives cite facts, so a name published in more than one member resolves to the fact. A record's own published name is the last thing tried, after every literal address has failed (``_record_addresses``). """ if not isinstance(document, dict) or not isinstance(key, str): return _MISSING bare = key.lstrip("/") for member in _CITABLE_MEMBERS: if bare.startswith(f"{member}/"): bare = bare[len(member) + 1:] break if bare.startswith(f"{member}{FLAT_KEY_SEPARATOR}"): bare = bare[len(member) + 1:] break bare = bare.lstrip(FLAT_KEY_SEPARATOR) if not bare: return _MISSING for member in _CITABLE_MEMBERS: candidate = document.get(member) if isinstance(candidate, dict) and bare in candidate: return candidate[bare] if bare in document: return document[bare] # Last: the name the payload gave a record, rather than the address it # filed it under. Literal addresses answer first, above; this only ever # rewrites the record prefix, and the rewritten address must still be # published. See ``_record_addresses``. for member in _CITABLE_MEMBERS: candidate = document.get(member) if not isinstance(candidate, dict): continue for name, address in _record_addresses(candidate).items(): if not bare.startswith(f"{name}{FLAT_KEY_SEPARATOR}"): continue translated = f"{address}{bare[len(name):]}" if translated in candidate: return candidate[translated] return _MISSING def _resolve_pointer(document: Any, pointer: str) -> Any: """Resolve a fact reference, accepting any spelling of one address. A pointer that navigates cleanly wins (it is the documented form and the only one that can address a nested package). Everything else falls through to ``_flat_lookup``, which judges the address itself. """ if pointer.startswith("/"): current: Any = document for raw_part in pointer[1:].split("/"): part = raw_part.replace("~1", "/").replace("~0", "~") if isinstance(current, dict) and part in current: current = current[part] elif isinstance(current, list) and part.isdigit() and int(part) < len(current): current = current[int(part)] else: current = _MISSING break if current is not _MISSING: return current found = _flat_lookup(document, pointer) if found is not _MISSING: return found raise IntelligenceContractError(f"Unknown fact reference: {pointer}") def _number_tokens(value: Any) -> set[Decimal]: if isinstance(value, bool) or value is None: return set() if isinstance(value, (int, float, Decimal)): try: return {Decimal(str(value)).normalize()} except InvalidOperation: return set() if not isinstance(value, str): return set() tokens: set[Decimal] = set() for match in _NUMBER.findall(value): try: tokens.add(Decimal(match.replace(",", "")).normalize()) except InvalidOperation: continue return tokens # Values that are DATES, anywhere in the fact package. Naming the period # a narrative is about is not a quantitative claim, so it does not need # its own citation — but every date the model may utter still has to # come from the facts it was handed. _ISO_DATEISH = re.compile(r"^\d{4}-\d{2}(-\d{2})?([T ].*)?$") def _period_number_tokens(facts: Any) -> set[Decimal]: """Numbers reachable from date-shaped values in the fact package. Without this, a summary that simply says which month it covers is rejected: "In July 2026, 11 controls..." yields the tokens 2026 and 7, neither of which the model cited, so the whole output is refused as containing invented numbers. Every AI subject in the platform hit this — ``intelligence_analyses`` had never once reached ``ready`` — because naming your own reporting period is the most natural sentence a summary can open with. Deliberately narrow: only values that LOOK like an ISO date contribute, and only their own digits. A magnitude the model made up is still refused, and pairing a real number with the wrong claim is still caught by the per-pointer check below, because this adds the period's digits and nothing else. """ tokens: set[Decimal] = set() stack = [facts] while stack: node = stack.pop() if isinstance(node, dict): stack.extend(node.values()) elif isinstance(node, (list, tuple)): stack.extend(node) elif isinstance(node, str) and _ISO_DATEISH.match(node): tokens.update(_number_tokens(node)) return tokens def _package_number_tokens(facts: Any) -> set[Decimal]: """Every number anywhere in the fact package. This is what "the model must not invent a number" actually requires: the figure has to COME FROM Helix. The stricter reading — that each number must appear under one of the pointers the model happened to list for that item — is a different and much harder claim, and four production deploys proved a model does not reliably satisfy it: it would write a correct figure while citing a neighbouring pointer, and the whole narrative was refused. That strictness is why ``intelligence_analyses`` had never once reached ``ready`` — not one successful analysis across four subjects, two of them shipped long before the compliance summary. Attempts to work around it by pre-publishing every form a narrator might use (an integer score, then each dimension, then the percentage) each bought exactly one deploy before the model found another phrasing, because the problem was never which forms existed. The trade this makes explicit: a number that IS in the facts can no longer be rejected for being paired with the wrong claim. That pairing was only ever checked incidentally, the fact package is the sole source the model is given, and every figure stays verifiable against the frozen snapshot beside it — whereas the alternative on the evidence is a feature that never produces anything at all. """ tokens: set[Decimal] = set() stack = [facts] while stack: node = stack.pop() if isinstance(node, dict): stack.extend(node.values()) elif isinstance(node, (list, tuple)): stack.extend(node) elif isinstance(node, bool) or node is None: continue elif isinstance(node, (int, float, Decimal)): # A numeric fact. This is the case that was failing: the # figure is Helix's own, and which pointer the model listed # for it does not change that. tokens.update(_number_tokens(node)) elif isinstance(node, str) and _ISO_DATEISH.match(node): # A date the model may name. tokens.update(_number_tokens(node)) # Free TEXT contributes nothing, deliberately. A field such as # ``note: "ticket 4711 was closed"`` must not let the model write # "4711 items" — that would turn any prose fact into a smuggling # channel for magnitudes, which is the whole point of the guard. return tokens # Roundings a narrator actually writes: whole, one decimal, two decimals. _DERIVED_QUANTA = (Decimal("1"), Decimal("0.1"), Decimal("0.01")) # Above this many distinct numeric facts the PAIRWISE pass (rules 3 and 4) # is skipped. Rules 1 and 2 are O(n) and always run — see # ``_derived_number_tokens``. # # 🔴 Do NOT raise this to make a rejection go away. Its original reason was # cost, and that reason no longer holds — the cross-domain brief's real # package is 161 facts and the full pass takes 0.16s. The reason it STAYS is # what the same measurement showed about caliber: at 161 facts the pairwise # rules derive **51,862** admissible values, and sampling random magnitudes # against that set, # # * 99.0% of every integer from 1 to 1000 would pass ungrounded, # * 100% of every one-decimal percentage would pass ungrounded, # * the guard's own documented must-refuse example ``312`` is admitted. # # At that point the number check no longer checks anything. The docstring # below accepts "a number that coincides with some ratio of two unrelated # facts" as a knowing trade — that trade is only tolerable while the package # is small enough for coincidences to be rare, which is exactly what this # cap encodes. A large package needs a SMALLER numeric surface, not a more # permissive guard (see the ``_FACT_LIST_ITEMS_SHOWN`` note in # ``generation.py`` for the other half of that argument). _MAX_PAIRWISE_FACTS = 120 _HUNDRED = Decimal(100) def _rounded_forms(value: Decimal) -> set[Decimal]: forms = {value.normalize()} for quantum in _DERIVED_QUANTA: try: forms.add(value.quantize(quantum, rounding=ROUND_HALF_UP).normalize()) except InvalidOperation: continue return forms def _derived_number_tokens(base: set[Decimal]) -> set[Decimal]: """Magnitudes a narrator computes *from* the facts it was given. Production evidence, not speculation. Every rejection the compliance summary produced was arithmetic on Helix's own figures: * ``63.64`` = ``100 - 36.36`` (the published fail share) — a complement * ``54.55`` = ``100 - 45.45`` — a complement again * ``6`` = ``11 - 5`` (total minus passing) — a difference None of those is a fabricated magnitude. They are the operations anyone writing "4 of 11 controls failed, so 63.64% did not" performs, and a summary that may not perform them cannot describe a ratio at all. The alternative — publishing each new form as a fact — was tried four times. Each deploy bought exactly one run before the model reached for a phrasing whose arithmetic had not been pre-computed, because the problem was never which forms existed. So the guard now knows the arithmetic instead of trying to enumerate its results. Exactly four operations, each with a sentence that needs it and a test that fails when it is removed. Two more were written and then deleted for lacking both: * pairwise SUMS admitted 116 further values including ``1045 + 5``, and no summary adds two unrelated counts together; * bare PROPORTIONS (``5 / 11`` as ``0.45``) were only ever justified by an accident — ``11 / 2 = 5.5`` rounding to the ``6`` production wanted, which the remainder rule derives honestly as ``11 - 5``. If a run ever fails on a sum there will be evidence for it, which is more than speculation bought the previous four attempts. The trade, stated plainly: a number that happens to coincide with some ratio of two unrelated facts is now accepted. What the guard still refuses is the thing it exists to refuse — a magnitude with no basis in the package at all, the invented "4200000 at risk" or "312 orphaned accounts". Every figure remains checkable against the frozen snapshot stored beside the narrative. """ derived: set[Decimal] = set() numbers = sorted(base) # Rules 1 and 2 are O(n) and run for ANY package size. They used to sit # behind the pairwise cap below, and that cost the cross-domain brief # three of its five 2026-09-01 rejections: its package holds **161** # distinct numeric facts, over the 120 cap, so every rule here was # silently off and the narrative could not so much as round a number # Helix itself had published. Measured on the real production snapshot: # ``72`` was ``72.14`` rounded and ``99.05`` was the complement of # ``0.95``, both refused as "uncited". A guard that forbids abbreviating # its own figure is not strict, it is wrong — and the cap's stated # reason was the quadratic pass, which these two are not. for value in numbers: # 1. Rounding a Helix figure. "45.45%" written as "45.5%". Found # while testing the rules above: a narrator abbreviating a # published number was still being refused, which is the same # class of failure as everything else here. derived.update(_rounded_forms(value)) # 2. The complement of a share. "36.36% failed" -> "63.64% did not". if Decimal(0) < value <= _HUNDRED: derived.update(_rounded_forms(_HUNDRED - value)) if len(numbers) > _MAX_PAIRWISE_FACTS: return derived for index, left in enumerate(numbers): for right in numbers[index + 1:]: # 3. A remainder. "1045 awaiting decision, 1041 unrelated". derived.update(_rounded_forms(abs(left - right))) for numerator, denominator in ((left, right), (right, left)): if denominator == 0: continue try: share = numerator / denominator except (InvalidOperation, DivisionByZero, ZeroDivisionError): continue # 4. "X of N is Y%" over a denominator Helix did not # publish a percentage for — 5 of 9 evaluated is # 55.56%, and which denominator is the right caliber # is a judgement the narrative is allowed to make. derived.update(_rounded_forms(share * _HUNDRED)) return derived def _grounded_number_tokens(facts: Any) -> set[Decimal]: """Every magnitude the narrative may contain, computed once per output.""" base = _package_number_tokens(facts) return base | _derived_number_tokens(base) def _validate_text_numbers( texts: list[str], pointers: list[str], facts: dict[str, Any], grounded: set[Decimal], ) -> None: # Pointers are still resolved and still must name scalars: a citation # that points nowhere, or at a whole sub-object, remains a contract # breach. Only the number check is widened. # # ``grounded`` is computed once per output rather than per item: the # pairwise derivation is O(n^2) and one output carries up to ~90 # narrative items. allowed: set[Decimal] = set(grounded) for pointer in pointers: value = _resolve_pointer(facts, pointer) if isinstance(value, (dict, list)): raise IntelligenceContractError(f"Fact reference must point to a scalar: {pointer}") allowed.update(_number_tokens(value)) mentioned: set[Decimal] = set() for text in texts: mentioned.update(_number_tokens(text)) invented = mentioned - allowed if invented: rendered = sorted(format(value, "f") for value in invented) raise IntelligenceContractError(f"Narrative contains uncited numbers: {rendered}") # The narrative lists on ``IntelligenceOutput``. Named once so the trimmer, # the evidence check and the grounding loop cannot fall out of step with the # schema — a field added there and forgotten here is a field nobody checks. _NARRATIVE_FIELDS = ( "conclusions", "risks", "opportunities", "forecasts", "recommendations", "data_gaps", ) def _trim_evidence_refs(candidate: dict[str, Any]) -> dict[str, Any]: """De-duplicate and cap each item's ``evidence_refs`` before validation. ``evidence_refs`` are anonymous drill-down anchors — "here is an object this claim is about". They are NOT what grounds the narrative's numbers: that is ``fact_refs`` plus ``_validate_text_numbers``, both untouched here. So an over-long evidence list is an over-eager citation habit, not a hallucination, and the honest response is to keep the first ``_EVIDENCE_REFS_MAX`` rather than discard an otherwise-grounded analysis. Production, 2026-08-17: one platform brief cited 42 refs on one conclusion and 124 on one risk, and the whole document — every correctly grounded number in it — was thrown away over the count. Trimming keeps order (the model's own ranking) and only ever REMOVES refs, so anything surviving was still validated against ``allowed_evidence_refs`` downstream. Every other cap stays strict: ``fact_refs`` carries the numbers and a document citing 40 facts for one sentence is not grounded, it is guessing. """ for key in _NARRATIVE_FIELDS: value = candidate.get(key) if not isinstance(value, list): continue for item in value: if not isinstance(item, dict): continue refs = item.get("evidence_refs") if not isinstance(refs, list): continue deduped: list[Any] = [] seen: set[Any] = set() for ref in refs: marker = ref if isinstance(ref, str) else repr(ref) if marker in seen: continue seen.add(marker) deduped.append(ref) if len(deduped) > _EVIDENCE_REFS_MAX: logger.info( "intelligence: trimmed %s evidence_refs from %d to %d", key, len(deduped), _EVIDENCE_REFS_MAX, ) deduped = deduped[:_EVIDENCE_REFS_MAX] item["evidence_refs"] = deduped return candidate # The action category a coerced recommendation falls back to. ``review`` is # the only verb in the vocabulary that authorises nothing: it is absent from # ``recommendations._APPROVAL_ACTIONS`` and its handling steps are the default # read-evidence / open-workbench / verify-snapshot walk. Coercing INTO it can # only ever weaken a recommendation's claim. _SAFEST_ACTION_KIND = "review" def _allowed_action_kinds(output_model: type[IntelligenceOutput]) -> frozenset[str]: """The action vocabulary this output contract actually accepts. Read off the contract rather than restated here: a subject that narrows the vocabulary in a subclass must narrow the coercion with it, and a second hand-kept copy of this list is exactly how the two drift. """ field = output_model.model_fields.get("recommendations") if field is None: return frozenset() item_models = [ arg for arg in get_args(field.annotation) if isinstance(arg, type) and issubclass(arg, GroundedNarrative) ] if not item_models: return frozenset() action = item_models[0].model_fields.get("action_kind") if action is None: return frozenset() return frozenset( arg for arg in get_args(action.annotation) if isinstance(arg, str) ) def _coerce_unknown_action_kinds( candidate: dict[str, Any], output_model: type[IntelligenceOutput], ) -> dict[str, Any]: """Downgrade an unrecognised action verb instead of discarding the document. ``action_kind`` is a CATEGORY of Helix action, not model prose — the narrative's own words are in the title and detail. When the model reaches for a verb the controlled vocabulary does not name, the useful response is the weakest category, not the loss of an otherwise fully grounded analysis. Production evidence: ``cloud_scope_health`` has never once published, and its 2026-08-24 rejection was a single word — ``action_kind='confirm'`` on recommendation #1 — which threw away 2548 output tokens of grounded narrative. The same failure cost a round earlier with ``assign``, which was then added to the vocabulary; a third verb was always going to arrive, and widening the list one word at a time also widens the approval gate one word at a time. Strictly a narrowing, in the same spirit as the low-confidence clamp below: ``review`` authorises nothing, so a coerced recommendation can never become a write it was not entitled to. Only a STRING outside the vocabulary is coerced — a missing or non-string ``action_kind`` is a real schema breach and still fails. Logged at warning, because a verb the model keeps reaching for is a vocabulary gap worth seeing. """ allowed = _allowed_action_kinds(output_model) if _SAFEST_ACTION_KIND not in allowed: return candidate items = candidate.get("recommendations") if not isinstance(items, list): return candidate for item in items: if not isinstance(item, dict): continue kind = item.get("action_kind") if isinstance(kind, str) and kind not in allowed: logger.warning( "intelligence: recommendation action_kind %r is outside the " "vocabulary — coerced to %r", kind, _SAFEST_ACTION_KIND, ) item["action_kind"] = _SAFEST_ACTION_KIND return candidate def validate_intelligence_output( raw: dict[str, Any], *, facts: dict[str, Any], allowed_evidence_refs: set[str], confidence_score: Decimal | int | float, output_model: type[IntelligenceOutput] = IntelligenceOutput, ) -> IntelligenceOutput: """Parse, ground and privacy-check one raw model response. Subject-specific output contracts must subclass ``IntelligenceOutput`` so grounding, anonymous-reference and low-confidence guards remain mandatory. """ if not isinstance(output_model, type) or not issubclass(output_model, IntelligenceOutput): raise IntelligenceContractError( "Intelligence output contracts must inherit IntelligenceOutput." ) candidate = _trim_evidence_refs(dict(raw)) candidate = _coerce_unknown_action_kinds(candidate, output_model) candidate.pop("_meta", None) # transport metadata is not model output try: output = output_model.model_validate(candidate) except ValidationError as exc: raise IntelligenceContractError(f"Invalid intelligence JSON contract: {exc}") from exc dumped = output.model_dump(mode="json") try: validate_anonymous_refs(dumped, allowed_evidence_refs) except UnsafeFactPackage as exc: raise IntelligenceContractError(str(exc)) from exc # ``validate_anonymous_refs`` scans the whole document but only judges # strings SHAPED like an anonymous ref (``^[a-z][a-z0-9]{1,15}_\d{2,6}$``) # — that is right for prose, where anything else is just words. Inside # ``evidence_refs`` it leaves a hole: the field's entire contract is "an # id from the package", so a value that does not even look like one was # never checked at all, and a model emitting a hostname or an address # there would have had it stored and rendered as an evidence anchor. # Every entry must name a real object, no exemption for the ones that # look least like one. unknown = sorted({ ref for field in _NARRATIVE_FIELDS for item in getattr(output, field) for ref in item.evidence_refs if ref not in allowed_evidence_refs }) if unknown: raise IntelligenceContractError( f"Model returned unknown evidence refs: {unknown}" ) grounded = _grounded_number_tokens(facts) _validate_text_numbers( [output.summary_zh, output.summary_en], output.summary_fact_refs, facts, grounded, ) all_items = [ item for field in _NARRATIVE_FIELDS for item in getattr(output, field) ] for item in all_items: _validate_text_numbers( [item.title_zh, item.title_en, item.detail_zh, item.detail_en], item.fact_refs, facts, grounded, ) if Decimal(str(confidence_score)) < 70: if not output.data_gaps: raise IntelligenceContractError( "Low-confidence analysis must explain at least one data gap" ) # Enforce the invariant by construction rather than by refusal. # Clamping can only ever LOWER a claim, so the rule — low-confidence # facts must not support high-confidence conclusions — holds exactly # as before, and it holds deterministically instead of depending on # the model obeying an instruction it was already given. # # Production, 2026-08-17: the cloud subscription analysis was # refused for this after its reference problems were fixed, throwing # away an otherwise grounded narrative because one item said "high". # Same reasoning as the ``evidence_refs`` trim above, and the same # bargain: never overstate, but do not discard sound work over a # mechanically correctable claim. # # The data-gaps rule above stays a REFUSAL: a missing explanation # cannot be manufactured, only the model can supply one. for item in all_items: if item.confidence == "high": item.confidence = "medium" return output