feat: add typed tier assurance guardrails
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

This commit is contained in:
codex 2026-08-19 21:32:07 +02:00
parent 3d0f614f49
commit c65a2f1ff9
13 changed files with 727 additions and 7 deletions

View file

@ -1,5 +1,6 @@
from __future__ import annotations
from dataclasses import replace
from decimal import Decimal
from pathlib import Path
@ -9,6 +10,7 @@ from adaptive_pricing_core.boundary_engine import (
PricingConfiguration,
validate_pricing_configuration,
)
from adaptive_pricing_core.pricing_models import AssuranceClaim
from observatory.load import load_pricing_models
DATA_DIR = Path(__file__).resolve().parent.parent / "data"
@ -18,6 +20,106 @@ def _model(model_id: str):
return next(item for item in load_pricing_models(DATA_DIR) if item.id == model_id)
def _configuration_with_claim(claim: AssuranceClaim) -> PricingConfiguration:
return PricingConfiguration(
model=replace(_model("flat-899-eur-monthly"), assurance_claims=(claim,)),
segment="coulomb-social-members",
allocated_fixed_cost=Decimal("1.00"),
payment_fee_rate_pct=Decimal("5"),
)
def _claim(**changes) -> AssuranceClaim:
defaults = {
"id": "isolation-tier",
"kind": "isolation",
"customer_wording": "Other tenants cannot reach your data.",
"minimum_levels": {"E": 4, "P": 3},
"delivering_service": "tenant-engine",
"evidence_ref": "evidence/isolation-review.json",
}
defaults.update(changes)
return AssuranceClaim(**defaults)
def test_valid_assurance_claim_passes_with_explainable_result() -> None:
result = validate_pricing_configuration(
_configuration_with_claim(_claim()),
BoundaryPolicy(),
)
assurance = next(item for item in result.constraints if item.id == "assurance-claims")
assert assurance.status == "pass"
assert assurance.details["claim_ids"] == ["isolation-tier"]
def test_isolation_wording_requires_e4_and_coupled_p3() -> None:
result = validate_pricing_configuration(
_configuration_with_claim(_claim(minimum_levels={"E": 3, "P": 2})),
BoundaryPolicy(),
)
assurance = next(item for item in result.constraints if item.id == "assurance-claims")
assert assurance.status == "fail"
assert "E §11.4" in assurance.reason
def test_retention_claim_requires_bounded_shared_horizon() -> None:
result = validate_pricing_configuration(
_configuration_with_claim(
_claim(
id="retention-tier",
kind="retention",
customer_wording="Deleted data is gone.",
minimum_levels={"R": 3, "P": 1},
)
),
BoundaryPolicy(),
)
assurance = next(item for item in result.constraints if item.id == "assurance-claims")
assert assurance.status == "fail"
assert "R §4.5.4/§11.1" in assurance.reason
assert "P §4.5.4" in assurance.reason
assert "R §11.4" in assurance.reason
def test_performance_claim_requires_p2_or_resource_governor() -> None:
result = validate_pricing_configuration(
_configuration_with_claim(
_claim(
id="performance-tier",
kind="performance",
customer_wording="Predictable performance under neighbour load.",
minimum_levels={"P": 1},
)
),
BoundaryPolicy(),
)
assurance = next(item for item in result.constraints if item.id == "assurance-claims")
assert assurance.status == "fail"
assert "P §8.3.1/§11.6" in assurance.reason
def test_availability_wording_requires_matching_failure_scope() -> None:
result = validate_pricing_configuration(
_configuration_with_claim(
_claim(
id="availability-tier",
kind="availability",
customer_wording="High availability survives regional loss.",
minimum_levels={"V": 3},
)
),
BoundaryPolicy(),
)
assurance = next(item for item in result.constraints if item.id == "assurance-claims")
assert assurance.status == "fail"
assert "V §11.4" in assurance.reason
def test_commitment_backed_discount_is_accepted_when_economics_stay_strong() -> None:
result = validate_pricing_configuration(
PricingConfiguration(

View file

@ -1,8 +1,12 @@
from __future__ import annotations
from dataclasses import replace
from pathlib import Path
from adaptive_pricing_core.governance import assess_tier_definition_assurance
from adaptive_pricing_core.pricing_models import AssuranceClaim
from observatory.api import build_dashboard_payload
from observatory.load import load_pricing_models
DATA_DIR = Path(__file__).resolve().parent.parent / "data"
@ -46,3 +50,30 @@ def test_recommendations_include_governed_execution_gate() -> None:
assert execution_gate["confidence"] == "0.88"
assert execution_gate["risks"]
assert execution_gate["supporting_observations"]
def test_changed_assurance_claim_requires_one_time_definition_approval() -> None:
previous = next(
item for item in load_pricing_models(DATA_DIR) if item.id == "flat-899-eur-monthly"
)
current = replace(
previous,
assurance_claims=(
AssuranceClaim(
id="availability-tier",
kind="availability",
customer_wording="Survives a zone loss.",
minimum_levels={"V": 3},
delivering_service="railiance-platform",
evidence_ref="evidence/zone-failover.json",
),
),
)
assessment = assess_tier_definition_assurance(current, previous)
unchanged = assess_tier_definition_assurance(current, current)
assert assessment.decision == "approval_required"
assert assessment.approvals[0].evidence_refs == ("evidence/zone-failover.json",)
assert unchanged.decision == "proceed"
assert unchanged.approvals == ()

View file

@ -1,9 +1,16 @@
from __future__ import annotations
import json
from dataclasses import replace
from decimal import Decimal
from pathlib import Path
from adaptive_pricing_core.pricing_models import validate_pricing_catalog
from adaptive_pricing_core.pricing_models import (
AssuranceClaim,
load_pricing_models as load_core_pricing_models,
validate_pricing_catalog,
validate_pricing_model,
)
from observatory.load import load_pricing_models
DATA_DIR = Path(__file__).resolve().parent.parent / "data"
@ -33,3 +40,55 @@ def test_flat_model_still_exposes_access_fee_compatibility_fields() -> None:
assert model.access_fee_amount == Decimal("8.99")
assert model.access_fee_cadence == "monthly"
assert len(model.charge_components) == 1
def test_assurance_claim_schema_is_optional_and_typed() -> None:
model = next(
item for item in load_pricing_models(DATA_DIR) if item.id == "flat-899-eur-monthly"
)
assert model.assurance_claims == ()
invalid = replace(
model,
assurance_claims=(
AssuranceClaim(
id="retention",
kind="retention",
customer_wording="Deleted data is gone.",
minimum_levels={"R": True, "P": 1},
delivering_service="",
evidence_ref="",
maximum_erasure_horizon_days=0,
erasure_mechanism="magic",
),
),
)
issues = validate_pricing_model(invalid)
assert any("invalid R level" in issue for issue in issues)
assert any("requires delivering_service" in issue for issue in issues)
assert any("requires evidence_ref" in issue for issue in issues)
assert any("invalid maximum erasure horizon" in issue for issue in issues)
assert any("unsupported erasure_mechanism" in issue for issue in issues)
def test_assurance_claim_loads_from_json_catalog(tmp_path: Path) -> None:
catalog = json.loads((DATA_DIR / "pricing-models.json").read_text(encoding="utf-8"))
catalog["models"][0]["assurance_claims"] = [
{
"id": "availability-tier",
"kind": "availability",
"customer_wording": "Survives a zone loss.",
"minimum_levels": {"V": "3"},
"delivering_service": "railiance-platform",
"evidence_ref": "evidence/zone-failover.json",
}
]
catalog_path = tmp_path / "pricing-models.json"
catalog_path.write_text(json.dumps(catalog), encoding="utf-8")
models = load_core_pricing_models(catalog_path)
claim = models[0].assurance_claims[0]
assert claim.id == "availability-tier"
assert claim.minimum_levels == {"V": 3}