from __future__ import annotations from dataclasses import dataclass, replace from datetime import datetime from enum import Enum from typing import Literal # ADR-0013: onboarding-risk / entity-shape grouping. Orthogonal to # capability role (below) -- neither axis constrains the other except # through guardrail policy, which is a reserved, unimplemented concern. GROUPINGS = frozenset( { "trial", "friendly", "single", "small", "medium", "large", "enterprise", "consumer", "family", "community", "association", "agentic", } ) # tenant:platform and tenant:coulomb predate the grouping taxonomy and stay # reserved, ungrouped identifiers (ADR-0013's Tenant Claim rationale). RESERVED_IDENTIFIERS = frozenset({"tenant:platform", "tenant:coulomb"}) class InvalidTenantIdentifierError(ValueError): """A tenant identifier does not match `tenant::` or a reserved form.""" class InvalidGrantError(ValueError): """A role grant violates a domain invariant (ADR-0014).""" class InvalidLifecycleTransitionError(ValueError): """A lifecycle transition is not legal from the tenant's current state.""" class ImmutableFieldError(ValueError): """An update tried to change a field that is immutable by contract.""" class EmptyUpdateError(ValueError): """An update carried no allow-listed field changes.""" class TenantRetiredError(ValueError): """A mutation was attempted on a retired tenant that only active tenants allow.""" class TenantLifecycle(str, Enum): """TEN-WP-0005: tenant existence is reversible, never hard-deleted. Retirement suspends a tenant's ability to take on new capability or plan state; it deliberately preserves the tenant record, its grant history, and its plan history so audit correlation and recovery stay intact. """ ACTIVE = "active" RETIRED = "retired" # The only tenant fields a PATCH may change. tenant_id, identifier, and # grouping are immutable: the identifier is the IAM Profile `tenant` claim # value that key-cape mints into tokens and flex-auth authorizes against, so # mutating it would silently invalidate every issued token referencing it. MUTABLE_METADATA_FIELDS = frozenset({"display_name", "contact_email"}) class CapabilityRole(str, Enum): """ADR-0014: non-exclusive capability roles a tenant may hold.""" PLTF = "PLTF" IAM = "IAM" VEN = "VEN" CUS = "CUS" GrantReason = Literal["plan_assignment", "manual_grant", "platform_default"] def parse_tenant_identifier(identifier: str) -> tuple[str | None, str]: """Return (grouping, name). grouping is None only for reserved identifiers.""" if identifier in RESERVED_IDENTIFIERS: return None, identifier.split(":", 1)[1] parts = identifier.split(":") if len(parts) != 3 or parts[0] != "tenant": raise InvalidTenantIdentifierError( f"Malformed tenant identifier: {identifier!r} (expected tenant::)" ) _, grouping, name = parts if grouping not in GROUPINGS: raise InvalidTenantIdentifierError(f"Unknown tenant grouping: {grouping!r}") if not name: raise InvalidTenantIdentifierError("Tenant name segment is empty") return grouping, name @dataclass(frozen=True, slots=True) class Tenant: tenant_id: str identifier: str grouping: str | None # -- TEN-WP-0005 lifecycle and mutable metadata. All default so that # pre-lifecycle construction sites (and migrated rows) keep working. display_name: str | None = None contact_email: str | None = None lifecycle: TenantLifecycle = TenantLifecycle.ACTIVE version: int = 1 created_at: datetime | None = None updated_at: datetime | None = None retired_at: datetime | None = None reactivated_at: datetime | None = None @classmethod def create( cls, *, tenant_id: str, identifier: str, display_name: str | None = None, contact_email: str | None = None, created_at: datetime | None = None, ) -> "Tenant": grouping, _name = parse_tenant_identifier(identifier) return cls( tenant_id=tenant_id, identifier=identifier, grouping=grouping, display_name=display_name, contact_email=contact_email, created_at=created_at, updated_at=created_at, ) @property def is_reserved(self) -> bool: return self.grouping is None @property def is_active(self) -> bool: return self.lifecycle is TenantLifecycle.ACTIVE def with_metadata(self, changes: dict[str, object], *, at: datetime) -> "Tenant": """Apply an allow-listed metadata change, bumping the record version. Fails closed on anything ambiguous: unknown fields, attempts to change an immutable field, an empty change set, or a no-op change set. A no-op is rejected rather than silently accepted so a caller never reads a version bump as evidence that a value actually changed. """ if self.lifecycle is not TenantLifecycle.ACTIVE: raise InvalidLifecycleTransitionError( "metadata of a retired tenant cannot be updated; reactivate first" ) unknown = set(changes) - MUTABLE_METADATA_FIELDS immutable = unknown & {"tenant_id", "identifier", "grouping", "version", "lifecycle"} if immutable: raise ImmutableFieldError(f"immutable field(s): {', '.join(sorted(immutable))}") if unknown: raise ImmutableFieldError(f"unknown field(s): {', '.join(sorted(unknown))}") if not changes: raise EmptyUpdateError("update carried no fields") if all(getattr(self, field) == value for field, value in changes.items()): raise EmptyUpdateError("update would not change any field") return replace(self, version=self.version + 1, updated_at=at, **changes) # type: ignore[arg-type] def with_grouping(self, grouping: str, *, at: datetime) -> "Tenant": """Reclassify the tenant (TEN-WP-0010). Deliberately *not* part of `with_metadata`. `display_name` and `contact_email` are cosmetic; grouping resolves spend ceilings, so a change here moves money. It gets its own method, its own route, and its own flex-auth action so policy can permit a rename without permitting a reclassification, and so the audit trail shows which one happened. The identifier's grouping segment is *historical* -- onboarding-time, immutable, and not authoritative for current grouping (ADR-0013 amendment proposed under TEN-WP-0010-T01). This field is the authoritative one, which is why it may diverge from the identifier. """ if self.is_reserved: # tenant:platform and tenant:coulomb are ungrouped by design and # resolve guardrails through the reserved profile. Giving one a # grouping would silently move the platform's own identity onto # the grouping ladder. raise ImmutableFieldError( "reserved tenants are ungrouped and cannot be reclassified" ) if self.lifecycle is not TenantLifecycle.ACTIVE: raise InvalidLifecycleTransitionError( "grouping of a retired tenant cannot be changed; reactivate first" ) if grouping not in GROUPINGS: raise InvalidTenantIdentifierError(f"Unknown tenant grouping: {grouping!r}") if grouping == self.grouping: raise EmptyUpdateError("update would not change the grouping") return replace(self, grouping=grouping, version=self.version + 1, updated_at=at) def retire(self, *, at: datetime) -> "Tenant": if self.lifecycle is TenantLifecycle.RETIRED: raise InvalidLifecycleTransitionError("tenant is already retired") return replace( self, lifecycle=TenantLifecycle.RETIRED, version=self.version + 1, updated_at=at, retired_at=at, ) def reactivate(self, *, at: datetime) -> "Tenant": """Return the tenant to active. Deliberately narrow: it restores the tenant's ability to receive new grants and plan changes, and does not resurrect revoked grants or invent plan state -- those stay exactly as retirement left them. """ if self.lifecycle is TenantLifecycle.ACTIVE: raise InvalidLifecycleTransitionError("tenant is already active") return replace( self, lifecycle=TenantLifecycle.ACTIVE, version=self.version + 1, updated_at=at, reactivated_at=at, ) @dataclass(frozen=True, slots=True) class RoleGrant: """Audited role grant record -- the Tenant Role & Plan Grant Contract shape. Append-only: revocation produces a new record via `revoke()`, it never deletes the original. """ grant_id: str tenant_id: str role: CapabilityRole grant_reason: GrantReason plan_id: str | None granted_by: str granted_at: datetime correlation_id: str revoked_at: datetime | None = None @property def active(self) -> bool: return self.revoked_at is None def revoke(self, *, at: datetime) -> "RoleGrant": if self.revoked_at is not None: raise InvalidGrantError(f"Grant {self.grant_id!r} is already revoked") return replace(self, revoked_at=at) @dataclass(frozen=True, slots=True) class PlanAssignment: """A tenant's current plan, referenced by id only -- never resolved or duplicated locally. Plan term definitions belong to adaptive-pricing. """ tenant_id: str plan_id: str assigned_at: datetime def create_role_grant( *, tenant: Tenant, grant_id: str, role: CapabilityRole, grant_reason: GrantReason, plan_id: str | None, granted_by: str, correlation_id: str, granted_at: datetime, ) -> RoleGrant: """Construct a RoleGrant, enforcing ADR-0014's domain invariants. - `plan_assignment` grants always require a `plan_id`. - `platform_default` grants never carry a `plan_id`, and are only valid for `trial`-grouped tenants (ADR-0014: trial may hold any role, unrestricted, for showcase purposes) or the reserved, ungrouped `tenant:platform`/`tenant:coulomb` tenants (their baseline roles were never purchased either). - `manual_grant` carries no grouping restriction and an optional `plan_id`. """ if grant_reason == "plan_assignment" and plan_id is None: raise InvalidGrantError("plan_assignment grants require a plan_id") if grant_reason == "platform_default": if plan_id is not None: raise InvalidGrantError("platform_default grants must not carry a plan_id") if tenant.grouping not in (None, "trial"): raise InvalidGrantError( "platform_default is only valid for trial-grouped or reserved tenants, " f"got grouping={tenant.grouping!r}" ) return RoleGrant( grant_id=grant_id, tenant_id=tenant.tenant_id, role=role, grant_reason=grant_reason, plan_id=plan_id, granted_by=granted_by, granted_at=granted_at, correlation_id=correlation_id, )