266 lines
9 KiB
Python
266 lines
9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from decimal import Decimal
|
|
from typing import Any, Literal
|
|
|
|
from .pricing_models import PricingModel
|
|
|
|
GovernanceDecision = Literal["proceed", "approval_required", "blocked"]
|
|
RecommendationType = Literal["research", "simulation", "model_change", "execution"]
|
|
RecommendationPriority = Literal["high", "medium", "low"]
|
|
RiskSeverity = Literal["low", "medium", "high"]
|
|
HealthStatus = Literal["pass", "warn", "fail"]
|
|
|
|
|
|
def _decimal(value: Decimal | str | int | float | None, default: str = "0") -> Decimal:
|
|
if value in (None, ""):
|
|
return Decimal(default)
|
|
return Decimal(str(value))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GovernancePolicy:
|
|
policy_id: str
|
|
max_self_serve_discount_pct: Decimal = Decimal("10")
|
|
max_customer_visible_price_increase_pct: Decimal = Decimal("15")
|
|
max_active_experiments: int = 2
|
|
max_concurrent_candidate_rollouts: int = 1
|
|
require_approval_for_candidate_rollout: bool = True
|
|
require_approval_for_approximate_provider_mapping: bool = True
|
|
block_unsupported_provider_artifacts: bool = True
|
|
drift_blocks_execution: bool = True
|
|
require_approval_for_price_change: bool = True
|
|
require_customer_notice_for_price_increase: bool = True
|
|
customer_notice_days: int = 30
|
|
grandfather_existing_customers: bool = True
|
|
customer_visible_tuning_enabled: bool = False
|
|
customer_visible_tuning_requires_active_model: bool = True
|
|
communication_owner_role: str = "operator"
|
|
default_approver_role: str = "operator"
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ApprovalRequirement:
|
|
id: str
|
|
title: str
|
|
approver_role: str
|
|
reason: str
|
|
blocking: bool = True
|
|
evidence_refs: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GovernanceRisk:
|
|
id: str
|
|
severity: RiskSeverity
|
|
summary: str
|
|
mitigation: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SupportingObservation:
|
|
id: str
|
|
title: str
|
|
summary: str
|
|
source_ref: str
|
|
value: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GovernanceAssessment:
|
|
decision: GovernanceDecision
|
|
summary: str
|
|
approvals: tuple[ApprovalRequirement, ...]
|
|
risks: tuple[GovernanceRisk, ...]
|
|
supporting_observations: tuple[SupportingObservation, ...]
|
|
notes: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SellerRecommendation:
|
|
id: str
|
|
recommendation_type: RecommendationType
|
|
priority: RecommendationPriority
|
|
title: str
|
|
rationale: str
|
|
suggested_action: str
|
|
confidence: Decimal
|
|
governance: GovernanceAssessment
|
|
risks: tuple[GovernanceRisk, ...]
|
|
supporting_observations: tuple[SupportingObservation, ...]
|
|
related_model_ids: tuple[str, ...] = ()
|
|
related_profile_ids: tuple[str, ...] = ()
|
|
notes: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HealthCheck:
|
|
id: str
|
|
title: str
|
|
status: HealthStatus
|
|
summary: str
|
|
value: str | None = None
|
|
threshold: str | None = None
|
|
suggested_action: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SafeTuningParameter:
|
|
key: str
|
|
label: str
|
|
description: str
|
|
data_type: str
|
|
default_value: str | None = None
|
|
min_value: str | None = None
|
|
max_value: str | None = None
|
|
customer_visible: bool = True
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SafeTuningExample:
|
|
id: str
|
|
title: str
|
|
outcome: str
|
|
summary: str
|
|
customer_message: str
|
|
visible_to_customer: bool
|
|
tradeoffs: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SafeTuningContract:
|
|
model_id: str
|
|
model_name: str
|
|
mode: str
|
|
customer_visible: bool
|
|
tunable_parameters: tuple[SafeTuningParameter, ...]
|
|
tradeoff_lexicon: dict[str, str]
|
|
examples: tuple[SafeTuningExample, ...]
|
|
notes: tuple[str, ...] = ()
|
|
|
|
|
|
def assurance_claim_approval_requirements(
|
|
model: PricingModel,
|
|
previous_model: PricingModel | None,
|
|
*,
|
|
approver_role: str = "operator",
|
|
) -> tuple[ApprovalRequirement, ...]:
|
|
"""Return one-time approval gates for assurance changes at tier definition.
|
|
|
|
Callers must pass the previously approved definition. Publication and
|
|
campaign workflows must not call this for an unchanged model: Tenancy
|
|
Posture section 11.3 records the approval once when a tier is defined.
|
|
"""
|
|
|
|
current = {claim.id: claim for claim in model.assurance_claims}
|
|
previous = {
|
|
claim.id: claim
|
|
for claim in (previous_model.assurance_claims if previous_model is not None else ())
|
|
}
|
|
changed_ids = sorted(
|
|
claim_id
|
|
for claim_id in current.keys() | previous.keys()
|
|
if current.get(claim_id) != previous.get(claim_id)
|
|
)
|
|
|
|
requirements: list[ApprovalRequirement] = []
|
|
for claim_id in changed_ids:
|
|
claim = current.get(claim_id)
|
|
prior_claim = previous.get(claim_id)
|
|
evidence_ref = (claim or prior_claim).evidence_ref
|
|
change = "removed" if claim is None else "added" if prior_claim is None else "changed"
|
|
requirements.append(
|
|
ApprovalRequirement(
|
|
id=f"assurance-claim-{claim_id}-approval",
|
|
title=f"Assurance claim '{claim_id}' approval",
|
|
approver_role=approver_role,
|
|
reason=(
|
|
f"Tier '{model.id}' {change} customer assurance claim '{claim_id}'. "
|
|
f"Review its typed minimums and evidence before the tier definition is accepted."
|
|
),
|
|
evidence_refs=(evidence_ref,),
|
|
)
|
|
)
|
|
return tuple(requirements)
|
|
|
|
|
|
def assess_tier_definition_assurance(
|
|
model: PricingModel,
|
|
previous_model: PricingModel | None,
|
|
*,
|
|
approver_role: str = "operator",
|
|
) -> GovernanceAssessment:
|
|
approvals = assurance_claim_approval_requirements(
|
|
model,
|
|
previous_model,
|
|
approver_role=approver_role,
|
|
)
|
|
if approvals:
|
|
return GovernanceAssessment(
|
|
decision="approval_required",
|
|
summary="Approval required for a changed customer assurance claim.",
|
|
approvals=approvals,
|
|
risks=(
|
|
GovernanceRisk(
|
|
id="unapproved-assurance-change",
|
|
severity="high",
|
|
summary="The tier definition changes a customer-visible assurance obligation.",
|
|
mitigation="Validate the typed minima, inspect the attached evidence, and record human approval once for this tier revision.",
|
|
),
|
|
),
|
|
supporting_observations=tuple(
|
|
SupportingObservation(
|
|
id=f"{approval.id}-evidence",
|
|
title="Assurance evidence",
|
|
summary="Evidence attached to the tier-definition approval gate.",
|
|
source_ref=approval.evidence_refs[0],
|
|
)
|
|
for approval in approvals
|
|
),
|
|
notes=("Do not reuse this definition-time gate as a per-campaign approval.",),
|
|
)
|
|
return GovernanceAssessment(
|
|
decision="proceed",
|
|
summary="No assurance claim changed since the approved tier definition.",
|
|
approvals=(),
|
|
risks=(),
|
|
supporting_observations=(),
|
|
)
|
|
|
|
|
|
def governance_policy_from_dict(raw: dict[str, Any]) -> GovernancePolicy:
|
|
return GovernancePolicy(
|
|
policy_id=raw.get("policy_id", "default-governance-policy"),
|
|
max_self_serve_discount_pct=_decimal(raw.get("max_self_serve_discount_pct"), "10"),
|
|
max_customer_visible_price_increase_pct=_decimal(
|
|
raw.get("max_customer_visible_price_increase_pct"),
|
|
"15",
|
|
),
|
|
max_active_experiments=int(raw.get("max_active_experiments", 2)),
|
|
max_concurrent_candidate_rollouts=int(raw.get("max_concurrent_candidate_rollouts", 1)),
|
|
require_approval_for_candidate_rollout=bool(
|
|
raw.get("require_approval_for_candidate_rollout", True)
|
|
),
|
|
require_approval_for_approximate_provider_mapping=bool(
|
|
raw.get("require_approval_for_approximate_provider_mapping", True)
|
|
),
|
|
block_unsupported_provider_artifacts=bool(
|
|
raw.get("block_unsupported_provider_artifacts", True)
|
|
),
|
|
drift_blocks_execution=bool(raw.get("drift_blocks_execution", True)),
|
|
require_approval_for_price_change=bool(raw.get("require_approval_for_price_change", True)),
|
|
require_customer_notice_for_price_increase=bool(
|
|
raw.get("require_customer_notice_for_price_increase", True)
|
|
),
|
|
customer_notice_days=int(raw.get("customer_notice_days", 30)),
|
|
grandfather_existing_customers=bool(raw.get("grandfather_existing_customers", True)),
|
|
customer_visible_tuning_enabled=bool(raw.get("customer_visible_tuning_enabled", False)),
|
|
customer_visible_tuning_requires_active_model=bool(
|
|
raw.get("customer_visible_tuning_requires_active_model", True)
|
|
),
|
|
communication_owner_role=raw.get("communication_owner_role", "operator"),
|
|
default_approver_role=raw.get("default_approver_role", "operator"),
|
|
metadata=dict(raw.get("metadata", {})),
|
|
)
|