"""Plan one reversible, evidence-gated AI stack migration transition. The artifact models expand, migrate, verify, and separately authorized contract steps over invented evidence. It never contacts, changes, or certifies a system. """ from dataclasses import asdict, dataclass, fields import hashlib import json import math PHASES = ("EXPAND", "MIGRATE", "VERIFY") def identity(value, *, maximum=160): if type(value) is not str or not value or value != value.strip(): raise ValueError("exact non-empty identity required") if len(value) > maximum or any(ord(character) < 32 for character in value): raise ValueError("bounded printable identity required") return value def sha256_text(value): identity(value, maximum=64) if len(value) != 64 or any(character not in "0123456789abcdef" for character in value): raise ValueError("lowercase sha256 digest required") return value def integer(value, lower, upper): if type(value) is not int or not lower <= value <= upper: raise ValueError("bounded non-boolean integer required") return value def finite_float(value, lower, upper, *, lower_inclusive=True): if type(value) is not float or not math.isfinite(value): raise ValueError("exact finite float required") lower_ok = value >= lower if lower_inclusive else value > lower if not lower_ok or value > upper: raise ValueError("bounded finite float required") return value def exact_bool(value): if type(value) is not bool: raise ValueError("exact boolean required") return value def bounded_sequence(value, lower, upper): if type(value) not in (tuple, list): raise ValueError("bounded tuple or list required") integer(len(value), lower, upper) return tuple(value) def digest(value): try: encoded = json.dumps( value, sort_keys=True, separators=(",", ":"), allow_nan=False ).encode() except (TypeError, ValueError, OverflowError) as error: raise ValueError("canonical JSON value required") from error return hashlib.sha256(encoded).hexdigest() def seal(record): if type(record.content_id) is not str: raise ValueError("content digest requires exact string") data = asdict(record) data.pop("content_id") return digest(data) def validate_record(record, cls): if type(record) is not cls: raise ValueError("concrete frozen record required") try: rebuilt = cls( **{field.name: getattr(record, field.name) for field in fields(cls)} ) except (AttributeError, TypeError, ValueError, OverflowError) as error: raise ValueError("malformed record") from error if record != rebuilt or record.content_id != rebuilt.content_id: raise ValueError("noncanonical or modified record") return rebuilt def scope_tuple(value): scope = bounded_sequence(value, 6, 6) for item in scope: identity(item) if len(scope) != len(set(scope)): raise ValueError("duplicate scope identity") return scope @dataclass(frozen=True) class MigrationContract: scope: tuple = ( "interface-contract-v1", "data-contract-v1", "evaluation-contract-v1", "routing-policy-v1", "rollback-policy-v1", "fixture-v1", ) minimum_shadow_requests: int = 10_000 minimum_agreement_rate: float = 0.995 maximum_error_rate_increase: float = 0.002 maximum_latency_ratio: float = 1.10 traffic_step: float = 0.25 maximum_target_traffic: float = 1.0 minimum_verification_windows: int = 3 minimum_rollback_replicas: int = 2 require_explicit_contraction: bool = True content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) integer(self.minimum_shadow_requests, 1, 1_000_000_000) finite_float(self.minimum_agreement_rate, 0.0, 1.0) finite_float(self.maximum_error_rate_increase, 0.0, 1.0) finite_float(self.maximum_latency_ratio, 1.0, 100.0) finite_float(self.traffic_step, 0.0, 1.0, lower_inclusive=False) finite_float( self.maximum_target_traffic, 0.0, 1.0, lower_inclusive=False ) if self.traffic_step > self.maximum_target_traffic: raise ValueError("traffic step exceeds the migration ceiling") integer(self.minimum_verification_windows, 1, 100_000) integer(self.minimum_rollback_replicas, 1, 100_000) if exact_bool(self.require_explicit_contraction) is not True: raise ValueError("destructive contraction requires separate authorization") expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("migration contract digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class MigrationEvidence: scope: tuple contract_content_id: str source_revision: str target_revision: str phase: str current_target_fraction: float source_ready_replicas: int target_ready_replicas: int dual_write_coverage: float backfill_coverage: float shadow_requests: int agreement_rate: float source_error_rate: float target_error_rate: float source_p95_latency_ms: float target_p95_latency_ms: float backward_compatible: bool old_readers_supported: bool rollback_route_ready: bool idempotent_replay_passed: bool observability_complete: bool verification_windows: int contraction_approved: bool rollback_retention_windows: int content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) sha256_text(self.contract_content_id) identity(self.source_revision) identity(self.target_revision) if self.source_revision == self.target_revision: raise ValueError("source and target revisions must differ") identity(self.phase) if self.phase not in PHASES: raise ValueError("unknown migration phase") finite_float(self.current_target_fraction, 0.0, 1.0) if self.phase == "EXPAND" and self.current_target_fraction != 0.0: raise ValueError("expand phase cannot expose target traffic") integer(self.source_ready_replicas, 0, 100_000) integer(self.target_ready_replicas, 0, 100_000) finite_float(self.dual_write_coverage, 0.0, 1.0) finite_float(self.backfill_coverage, 0.0, 1.0) integer(self.shadow_requests, 0, 1_000_000_000) finite_float(self.agreement_rate, 0.0, 1.0) finite_float(self.source_error_rate, 0.0, 1.0) finite_float(self.target_error_rate, 0.0, 1.0) finite_float( self.source_p95_latency_ms, 0.0, 86_400_000.0, lower_inclusive=False ) finite_float( self.target_p95_latency_ms, 0.0, 86_400_000.0, lower_inclusive=False ) exact_bool(self.backward_compatible) exact_bool(self.old_readers_supported) exact_bool(self.rollback_route_ready) exact_bool(self.idempotent_replay_passed) exact_bool(self.observability_complete) integer(self.verification_windows, 0, 100_000) exact_bool(self.contraction_approved) integer(self.rollback_retention_windows, 0, 100_000) if self.phase != "VERIFY" and self.contraction_approved: raise ValueError("contraction can be approved only after entering verify") expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("migration evidence digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class MigrationPlan: decision: str violations: tuple phase: str source_revision: str target_revision: str current_target_fraction: float next_target_fraction: float evidence_id: str claim: str = "LOCAL_MIGRATION_PLAN_NOT_EXECUTION_OR_PRODUCTION_CERTIFICATION" def plan_migration( contract: MigrationContract, evidence: MigrationEvidence ) -> MigrationPlan: """Return one replay-safe proposal without mutating a source or target stack.""" contract = validate_record(contract, MigrationContract) evidence = validate_record(evidence, MigrationEvidence) if evidence.scope != contract.scope or evidence.contract_content_id != contract.content_id: raise ValueError("evidence belongs to another migration contract") if evidence.current_target_fraction > contract.maximum_target_traffic: raise ValueError("current traffic exceeds the contract ceiling") regressions = [] holds = [] if evidence.agreement_rate < contract.minimum_agreement_rate: regressions.append("agreement-regression") if ( evidence.target_error_rate - evidence.source_error_rate > contract.maximum_error_rate_increase ): regressions.append("error-regression") if ( evidence.target_p95_latency_ms / evidence.source_p95_latency_ms > contract.maximum_latency_ratio ): regressions.append("latency-regression") if not evidence.backward_compatible: holds.append("backward-compatibility") if not evidence.old_readers_supported: holds.append("old-reader-support") if evidence.dual_write_coverage < 1.0: holds.append("dual-write-coverage") if evidence.backfill_coverage < 1.0: holds.append("backfill-coverage") if evidence.shadow_requests < contract.minimum_shadow_requests: holds.append("shadow-evidence") if evidence.source_ready_replicas < contract.minimum_rollback_replicas: holds.append("source-capacity") if evidence.target_ready_replicas < 1: holds.append("target-capacity") if not evidence.rollback_route_ready: holds.append("rollback-route") if not evidence.idempotent_replay_passed: holds.append("idempotent-replay") if not evidence.observability_complete: holds.append("observability") decision = "HOLD_MIGRATION" next_fraction = evidence.current_target_fraction rollback_available = ( evidence.source_ready_replicas >= contract.minimum_rollback_replicas and evidence.rollback_route_ready and evidence.backward_compatible and evidence.old_readers_supported ) rollback_unavailable = bool( regressions and evidence.current_target_fraction > 0.0 and not rollback_available ) violations = tuple( regressions + (["rollback-unavailable"] if rollback_unavailable else []) + holds ) if regressions and evidence.current_target_fraction > 0.0: if rollback_available: decision = "ROLLBACK_TO_SOURCE" next_fraction = 0.0 else: decision = "STOP_AND_ESCALATE" elif violations: decision = "HOLD_MIGRATION" elif evidence.phase == "EXPAND": decision = "START_BOUNDED_MIGRATION" next_fraction = min( contract.traffic_step, contract.maximum_target_traffic ) elif evidence.phase == "MIGRATE": if evidence.current_target_fraction < contract.maximum_target_traffic: decision = "ADVANCE_BOUNDED_TRAFFIC" next_fraction = min( evidence.current_target_fraction + contract.traffic_step, contract.maximum_target_traffic, ) else: decision = "ENTER_VERIFY" else: verify_holds = [] if evidence.current_target_fraction != contract.maximum_target_traffic: verify_holds.append("target-traffic-incomplete") if evidence.verification_windows < contract.minimum_verification_windows: verify_holds.append("verification-windows") if not evidence.contraction_approved: verify_holds.append("contraction-approval-required") if evidence.rollback_retention_windows < contract.minimum_verification_windows: verify_holds.append("rollback-retention") if verify_holds: decision = "HOLD_MIGRATION" violations = tuple(verify_holds) else: decision = "AUTHORIZE_SEPARATE_CONTRACT_CHANGE" return MigrationPlan( decision=decision, violations=violations, phase=evidence.phase, source_revision=evidence.source_revision, target_revision=evidence.target_revision, current_target_fraction=evidence.current_target_fraction, next_target_fraction=next_fraction, evidence_id=evidence.content_id, ) def illustrative_fixture(): contract = MigrationContract() evidence = MigrationEvidence( scope=contract.scope, contract_content_id=contract.content_id, source_revision="ai-stack-v4", target_revision="ai-stack-v5", phase="MIGRATE", current_target_fraction=0.25, source_ready_replicas=4, target_ready_replicas=4, dual_write_coverage=1.0, backfill_coverage=1.0, shadow_requests=25_000, agreement_rate=0.999, source_error_rate=0.005, target_error_rate=0.005, source_p95_latency_ms=180.0, target_p95_latency_ms=184.0, backward_compatible=True, old_readers_supported=True, rollback_route_ready=True, idempotent_replay_passed=True, observability_complete=True, verification_windows=0, contraction_approved=False, rollback_retention_windows=0, ) return contract, evidence def main(): contract, evidence = illustrative_fixture() report = plan_migration(contract, evidence) print("example=illustrative_only") print(f"decision={report.decision}") print(f"phase={report.phase};source={report.source_revision};target={report.target_revision}") print( "target_traffic=" f"{report.current_target_fraction:.2f}->{report.next_target_fraction:.2f}" ) print(f"claim={report.claim}") if __name__ == "__main__": main()