Implement boundary engine and close WP-0004

projects/coulomb-pricing/observatory/api.py

@@ -20,6 +20,7 @@ from .load import (
     load_value_range,
 )
 from .allocation import build_cost_allocation
+from .boundary import build_boundary_validation
 from .credits import build_credit_summary, load_credit_wallets
 from .membership_analytics import build_membership_analytics
 from .pricing_context import build_cost_floor, build_market_price_view, build_value_range_view
@@ -93,6 +94,7 @@ def build_dashboard_payload(data_dir: Path | None = None, period: str | None = N
     cost_allocation = build_cost_allocation(snapshot, usage_records)
     ai_cost_per_member = usage_summary["cost_per_active_user_eur"]
     simulations = build_pricing_simulations(snapshot, models, ai_cost_per_member)
+    boundary_validation = build_boundary_validation(snapshot, models, usage_records)
     credit_wallets = load_credit_wallets(root)
     credit_summary = build_credit_summary(
         credit_wallets,
@@ -125,6 +127,7 @@ def build_dashboard_payload(data_dir: Path | None = None, period: str | None = N
             "usage": usage_summary,
             "cost_allocation": cost_allocation,
             "pricing_simulations": simulations,
+            "boundary_validation": boundary_validation,
             "credit_wallets": credit_summary,
             "recommendations": recommendations,
             "infrastructure": {

projects/coulomb-pricing/observatory/boundary.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+from decimal import Decimal
+from typing import Any
+
+from .models import EconomicsSnapshot, PricingModel
+
+from ._repo_root import ensure_repo_root_on_syspath
+
+ensure_repo_root_on_syspath()
+
+from adaptive_pricing_core.boundary_engine import (  # noqa: E402
+    BoundaryPolicy,
+    PricingConfiguration,
+    validate_pricing_configuration,
+)
+
+
+def _usage_unit_cost(records: list[dict[str, Any]], period: str) -> Decimal:
+    period_rows = [row for row in records if row.get("period") == period]
+    total_tokens = sum(Decimal(str(row.get("tokens", "0"))) for row in period_rows)
+    total_cost = sum(Decimal(str(row.get("cost_eur", "0"))) for row in period_rows)
+    if total_tokens <= Decimal("0"):
+        return Decimal("0")
+    return total_cost / total_tokens
+
+
+def _total_usage_units(records: list[dict[str, Any]], period: str) -> Decimal:
+    return sum(Decimal(str(row.get("tokens", "0"))) for row in records if row.get("period") == period)
+
+
+def build_boundary_policy(snapshot: EconomicsSnapshot) -> BoundaryPolicy:
+    return BoundaryPolicy(
+        minimum_margin_pct=Decimal("0"),
+        target_margin_pct=Decimal("15"),
+        max_payment_fee_pct=Decimal("10"),
+        max_expected_usage_variance_pct=Decimal("50"),
+        approval_discount_pct=Decimal("10"),
+        max_discount_pct=Decimal("25"),
+        minimum_contract_duration_for_discount_months=3,
+        minimum_turnover_multiple_for_discount=Decimal("1"),
+        minimum_prepayment_months_for_discount=Decimal("1"),
+    )
+
+
+def build_boundary_validation(
+    snapshot: EconomicsSnapshot,
+    models: list[PricingModel],
+    usage_records: list[dict[str, Any]],
+    segment: str | None = None,
+) -> dict[str, Any]:
+    fee_rate_pct = Decimal("0")
+    if snapshot.monthly_revenue > Decimal("0"):
+        fee_rate_pct = (
+            snapshot.monthly_payment_processing_cost / snapshot.monthly_revenue
+        ) * Decimal("100")
+
+    unit_cost = _usage_unit_cost(usage_records, snapshot.period)
+    usage_units = _total_usage_units(usage_records, snapshot.period)
+    direct_usage_cost = sum(
+        Decimal(str(row.get("cost_eur", "0")))
+        for row in usage_records
+        if row.get("period") == snapshot.period
+    )
+    allocated_fixed_cost = (
+        snapshot.monthly_infrastructure_cost / snapshot.active_members
+        if snapshot.active_members
+        else snapshot.monthly_infrastructure_cost
+    )
+
+    policy = build_boundary_policy(snapshot)
+    model_results = []
+    for model in models:
+        has_usage_component = any(component.kind == "usage" for component in model.charge_components)
+        configuration = PricingConfiguration(
+            model=model,
+            segment=segment or (model.eligibility[0] if model.eligibility else None),
+            expected_usage_units=usage_units if has_usage_component else Decimal("0"),
+            expected_usage_variance_pct=Decimal("25"),
+            allocated_fixed_cost=allocated_fixed_cost,
+            direct_cost_amount=Decimal("0") if has_usage_component else direct_usage_cost,
+            unit_cost=unit_cost if has_usage_component else Decimal("0"),
+            payment_fee_rate_pct=fee_rate_pct,
+        )
+        model_results.append(validate_pricing_configuration(configuration, policy))
+
+    return {
+        "period": snapshot.period,
+        "policy": policy,
+        "assumptions": {
+            "segment": segment or "model-default-eligibility",
+            "observed_usage_units": usage_units,
+            "observed_usage_unit_cost": unit_cost,
+            "direct_usage_cost_for_flat_models": direct_usage_cost,
+            "allocated_fixed_cost_per_member": allocated_fixed_cost,
+            "payment_fee_rate_pct": fee_rate_pct,
+            "expected_usage_variance_pct": Decimal("25"),
+        },
+        "model_results": model_results,
+        "notes": [
+            "This MVP policy uses current observatory economics and conservative defaults rather than a seller-specific governance file.",
+            "Self-serve discounts above 10% require approval; discounts above 25% are rejected.",
+            "Term-only concessions require at least a 3-month contract to count as meaningful commitment support.",
+        ],
+    }

projects/coulomb-pricing/tests/test_api.py

@@ -24,9 +24,15 @@ def test_dashboard_payload_contains_live_ledger_totals() -> None:
     assert payload["membership_analytics"]["active_members"] == 1
     assert payload["usage"]["record_count"] == 1
     assert len(payload["pricing_simulations"]["scenarios"]) == 3
+    assert len(payload["boundary_validation"]["model_results"]) == 3
+    assert payload["boundary_validation"]["policy"]["target_margin_pct"] == "15"
+    assert any(
+        result["decision"] == "rejected"
+        for result in payload["boundary_validation"]["model_results"]
+    )
     assert payload["recommendations"]
 
 
 def test_payload_json_is_valid() -> None:
     parsed = json.loads(payload_json(DATA_DIR, "2026-06"))
-    assert Decimal(parsed["payments"][0]["fees_amount"]) == Decimal("0.44")
+    assert Decimal(parsed["payments"][0]["fees_amount"]) == Decimal("0.44")

projects/coulomb-pricing/tests/test_boundary_engine.py

@@ -0,0 +1,168 @@
+from __future__ import annotations
+
+from decimal import Decimal
+from pathlib import Path
+
+from adaptive_pricing_core.boundary_engine import (
+    BoundaryPolicy,
+    CommitmentTerms,
+    PricingConfiguration,
+    validate_pricing_configuration,
+)
+from observatory.load import load_pricing_models
+
+DATA_DIR = Path(__file__).resolve().parent.parent / "data"
+
+
+def _model(model_id: str):
+    return next(item for item in load_pricing_models(DATA_DIR) if item.id == model_id)
+
+
+def test_commitment_backed_discount_is_accepted_when_economics_stay_strong() -> None:
+    result = validate_pricing_configuration(
+        PricingConfiguration(
+            model=_model("membership-plus-overage"),
+            segment="coulomb-social-members",
+            expected_usage_units=Decimal("100200"),
+            expected_usage_variance_pct=Decimal("25"),
+            allocated_fixed_cost=Decimal("2.00"),
+            unit_cost=Decimal("0.00000125"),
+            payment_fee_rate_pct=Decimal("5"),
+            usage_unit_price=Decimal("0.0015"),
+            commitment_terms=CommitmentTerms(
+                contract_duration_months=6,
+                minimum_monthly_turnover=Decimal("9.30"),
+            ),
+        ),
+        BoundaryPolicy(),
+    )
+
+    assert result.decision == "accepted"
+    assert result.valid is True
+    assert result.requires_approval is False
+    commitment_result = next(
+        item for item in result.constraints if item.id == "commitment-backed-concession"
+    )
+    assert commitment_result.status == "pass"
+    assert "minimum_monthly_turnover" in commitment_result.details["signals"]
+
+
+def test_discount_without_commitment_is_rejected() -> None:
+    result = validate_pricing_configuration(
+        PricingConfiguration(
+            model=_model("membership-plus-overage"),
+            segment="coulomb-social-members",
+            expected_usage_units=Decimal("100200"),
+            expected_usage_variance_pct=Decimal("25"),
+            allocated_fixed_cost=Decimal("2.00"),
+            unit_cost=Decimal("0.00000125"),
+            payment_fee_rate_pct=Decimal("5"),
+            usage_unit_price=Decimal("0.0015"),
+        ),
+        BoundaryPolicy(),
+    )
+
+    assert result.decision == "rejected"
+    assert result.valid is False
+    failing_ids = {
+        item.id for item in result.constraints if item.status == "fail" and item.severity == "hard"
+    }
+    assert "commitment-backed-concession" in failing_ids
+
+
+def test_weak_commitment_trade_is_rejected() -> None:
+    result = validate_pricing_configuration(
+        PricingConfiguration(
+            model=_model("membership-plus-overage"),
+            segment="coulomb-social-members",
+            expected_usage_units=Decimal("100200"),
+            expected_usage_variance_pct=Decimal("25"),
+            allocated_fixed_cost=Decimal("2.00"),
+            unit_cost=Decimal("0.00000125"),
+            payment_fee_rate_pct=Decimal("5"),
+            usage_unit_price=Decimal("0.0015"),
+            commitment_terms=CommitmentTerms(
+                contract_duration_months=2,
+                minimum_monthly_turnover=Decimal("9.00"),
+            ),
+        ),
+        BoundaryPolicy(),
+    )
+
+    assert result.decision == "rejected"
+    commitment_result = next(
+        item for item in result.constraints if item.id == "commitment-backed-concession"
+    )
+    assert commitment_result.status == "fail"
+
+
+def test_large_but_supported_concession_requires_approval() -> None:
+    result = validate_pricing_configuration(
+        PricingConfiguration(
+            model=_model("flat-899-eur-monthly"),
+            segment="coulomb-social-members",
+            allocated_fixed_cost=Decimal("1.00"),
+            payment_fee_rate_pct=Decimal("5"),
+            access_fee_amount=Decimal("7.50"),
+            commitment_terms=CommitmentTerms(
+                contract_duration_months=6,
+                minimum_monthly_turnover=Decimal("8.00"),
+            ),
+        ),
+        BoundaryPolicy(),
+    )
+
+    assert result.decision == "requires_approval"
+    assert result.valid is True
+    assert result.requires_approval is True
+    review_ids = {item.id for item in result.constraints if item.status == "review"}
+    assert "discount-approval-threshold" in review_ids
+
+
+def test_zero_usage_flat_configuration_can_pass() -> None:
+    result = validate_pricing_configuration(
+        PricingConfiguration(
+            model=_model("flat-899-eur-monthly"),
+            segment="coulomb-social-members",
+            expected_usage_units=Decimal("0"),
+            expected_usage_variance_pct=Decimal("0"),
+            allocated_fixed_cost=Decimal("3.00"),
+            payment_fee_rate_pct=Decimal("5"),
+        ),
+        BoundaryPolicy(),
+    )
+
+    assert result.decision == "accepted"
+    assert result.metrics.billable_usage_units == Decimal("0")
+
+
+def test_high_fee_configuration_is_rejected() -> None:
+    result = validate_pricing_configuration(
+        PricingConfiguration(
+            model=_model("flat-899-eur-monthly"),
+            segment="coulomb-social-members",
+            allocated_fixed_cost=Decimal("1.00"),
+            payment_fee_rate_pct=Decimal("30"),
+        ),
+        BoundaryPolicy(),
+    )
+
+    assert result.decision == "rejected"
+    payment_fee_result = next(item for item in result.constraints if item.id == "payment-fee-limit")
+    assert payment_fee_result.status == "fail"
+
+
+def test_unprofitable_configuration_is_rejected() -> None:
+    result = validate_pricing_configuration(
+        PricingConfiguration(
+            model=_model("flat-899-eur-monthly"),
+            segment="coulomb-social-members",
+            allocated_fixed_cost=Decimal("10.00"),
+            payment_fee_rate_pct=Decimal("5"),
+        ),
+        BoundaryPolicy(),
+    )
+
+    assert result.decision == "rejected"
+    assert result.metrics.monthly_margin < Decimal("0")
+    assert any(item.id == "cost-floor-coverage" and item.status == "fail" for item in result.constraints)