"""Deterministic AI-capability sourcing audit over invented evidence.""" from dataclasses import asdict, dataclass, fields import hashlib import json import re IDENTITY = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:/@-]{0,127}") SHA256 = re.compile(r"[a-f0-9]{64}") MODES = ("build", "buy", "partner") def identity(value): if type(value) is not str or not IDENTITY.fullmatch(value): raise ValueError("exact bounded identity required") return value def sha256_text(value): if type(value) is not str or not SHA256.fullmatch(value): raise ValueError("exact lowercase SHA-256 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 exact_bool(value): if type(value) is not bool: raise ValueError("exact boolean required") return value def bounded_tuple(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 token(label): identity(label) return hashlib.sha256(label.encode()).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): result = bounded_tuple(value, 6, 6) for item in result: identity(item) if len(set(result)) != len(result): raise ValueError("duplicate scope identity") return result def identity_tuple(value, lower, upper): result = bounded_tuple(value, lower, upper) for item in result: identity(item) if len(set(result)) != len(result): raise ValueError("duplicate identity") return result @dataclass(frozen=True) class SourcingContract: scope: tuple = ( "capability-contract-v1", "risk-profile-v1", "evaluation-policy-v1", "security-review-v1", "exit-plan-v1", "fixture-v1", ) required_controls: tuple = ( "audit-export", "data-boundary", "incident-notification", "rollback-or-exit", ) minimum_quality_permille: int = 900 max_p95_latency_ms: int = 500 max_annual_cost_units: int = 1_200 max_launch_days: int = 180 max_vendor_concentration_permille: int = 700 minimum_exit_readiness_permille: int = 800 primary_objective: str = "minimize-annual-cost-subject-to-gates" content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) object.__setattr__( self, "required_controls", identity_tuple(self.required_controls, 1, 32), ) integer(self.minimum_quality_permille, 0, 1_000) integer(self.max_p95_latency_ms, 1, 86_400_000) integer(self.max_annual_cost_units, 1, 1_000_000_000) integer(self.max_launch_days, 1, 36_500) integer(self.max_vendor_concentration_permille, 0, 1_000) integer(self.minimum_exit_readiness_permille, 0, 1_000) if identity(self.primary_objective) != "minimize-annual-cost-subject-to-gates": raise ValueError("unsupported primary objective") expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("sourcing contract digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class ControlCheck: scope: tuple option_id: str sourcing_mode: str option_evidence_digest: str name: str passed: bool evidence_fresh: bool evidence_digest: str content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) identity(self.option_id) if type(self.sourcing_mode) is not str or self.sourcing_mode not in MODES: raise ValueError("control sourcing mode must be build, buy, or partner") sha256_text(self.option_evidence_digest) identity(self.name) exact_bool(self.passed) exact_bool(self.evidence_fresh) sha256_text(self.evidence_digest) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("control-check digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class SourcingOption: scope: tuple option_id: str mode: str option_evidence_digest: str annual_cost_units: int launch_days: int quality_passes: int quality_trials: int p95_latency_ms: int vendor_concentration_permille: int exit_readiness_permille: int controls: tuple content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) identity(self.option_id) if type(self.mode) is not str or self.mode not in MODES: raise ValueError("mode must be build, buy, or partner") sha256_text(self.option_evidence_digest) integer(self.annual_cost_units, 1, 1_000_000_000) integer(self.launch_days, 1, 36_500) integer(self.quality_trials, 1, 1_000_000_000) integer(self.quality_passes, 0, self.quality_trials) integer(self.p95_latency_ms, 1, 86_400_000) integer(self.vendor_concentration_permille, 0, 1_000) integer(self.exit_readiness_permille, 0, 1_000) checks = bounded_tuple(self.controls, 1, 32) canonical = [] for check in checks: rebuilt = validate_record(check, ControlCheck) if rebuilt.scope != self.scope: raise ValueError("control check outside option scope") if ( rebuilt.option_id != self.option_id or rebuilt.sourcing_mode != self.mode or rebuilt.option_evidence_digest != self.option_evidence_digest ): raise ValueError("control check belongs to another sourcing option") canonical.append(rebuilt) if len({check.name for check in canonical}) != len(canonical): raise ValueError("duplicate control check") object.__setattr__(self, "controls", tuple(canonical)) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("sourcing-option digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class SourcingEvidence: scope: tuple contract_content_id: str options: tuple review_fresh: bool content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) sha256_text(self.contract_content_id) options = bounded_tuple(self.options, len(MODES), len(MODES)) canonical = [] for option in options: rebuilt = validate_record(option, SourcingOption) if rebuilt.scope != self.scope: raise ValueError("option outside sourcing scope") canonical.append(rebuilt) if {option.mode for option in canonical} != set(MODES): raise ValueError("exactly one build, buy, and partner option required") if len({option.option_id for option in canonical}) != len(canonical): raise ValueError("duplicate option identity") exact_bool(self.review_fresh) object.__setattr__(self, "options", tuple(canonical)) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("sourcing-evidence digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class SourcingDecision: decision: str selected_option: str selected_mode: str eligible_options: tuple violations: tuple evidence_id: str claim: str = "LOCAL_SOURCING_AUDIT_NOT_PROCUREMENT_APPROVAL" def decide_sourcing(contract: SourcingContract, evidence: SourcingEvidence): """Select by one declared objective after hard gates; mutate nothing.""" contract = validate_record(contract, SourcingContract) evidence = validate_record(evidence, SourcingEvidence) if evidence.scope != contract.scope or evidence.contract_content_id != contract.content_id: raise ValueError("evidence belongs to another sourcing contract") eligible = [] violations = [] required = set(contract.required_controls) for option in sorted(evidence.options, key=lambda item: item.option_id): reasons = [] checks = {check.name: check for check in option.controls} if set(checks) != required: reasons.append("control-set") for name in sorted(required & set(checks)): if not checks[name].passed: reasons.append(f"control-{name}") elif not checks[name].evidence_fresh: reasons.append(f"stale-{name}") if option.quality_passes * 1_000 < contract.minimum_quality_permille * option.quality_trials: reasons.append("quality") if option.p95_latency_ms > contract.max_p95_latency_ms: reasons.append("latency") if option.annual_cost_units > contract.max_annual_cost_units: reasons.append("annual-cost") if option.launch_days > contract.max_launch_days: reasons.append("launch-time") if option.vendor_concentration_permille > contract.max_vendor_concentration_permille: reasons.append("vendor-concentration") if option.exit_readiness_permille < contract.minimum_exit_readiness_permille: reasons.append("exit-readiness") if not evidence.review_fresh: reasons.append("stale-review") if reasons: violations.extend(f"{option.option_id}/{reason}" for reason in reasons) else: eligible.append(option) if not eligible: return SourcingDecision( decision="NO_ELIGIBLE_OPTION", selected_option="none", selected_mode="none", eligible_options=(), violations=tuple(violations), evidence_id=evidence.content_id, ) selected = min( eligible, key=lambda item: ( item.annual_cost_units, item.launch_days, item.p95_latency_ms, item.option_id, ), ) return SourcingDecision( decision="SELECT_ELIGIBLE_OPTION", selected_option=selected.option_id, selected_mode=selected.mode, eligible_options=tuple(sorted(option.option_id for option in eligible)), violations=tuple(violations), evidence_id=evidence.content_id, ) def illustrative_fixture(): contract = SourcingContract() def controls(prefix, option_id, mode, option_evidence_digest): return tuple( ControlCheck( scope=contract.scope, option_id=option_id, sourcing_mode=mode, option_evidence_digest=option_evidence_digest, name=name, passed=True, evidence_fresh=True, evidence_digest=token(f"{prefix}-{name}"), ) for name in contract.required_controls ) build = SourcingOption( scope=contract.scope, option_id="internal-platform", mode="build", option_evidence_digest=token("build-option-profile"), annual_cost_units=1_100, launch_days=150, quality_passes=940, quality_trials=1_000, p95_latency_ms=420, vendor_concentration_permille=100, exit_readiness_permille=950, controls=controls( "build", "internal-platform", "build", token("build-option-profile") ), ) buy = SourcingOption( scope=contract.scope, option_id="managed-service", mode="buy", option_evidence_digest=token("buy-option-profile"), annual_cost_units=700, launch_days=60, quality_passes=925, quality_trials=1_000, p95_latency_ms=340, vendor_concentration_permille=600, exit_readiness_permille=850, controls=controls( "buy", "managed-service", "buy", token("buy-option-profile") ), ) partner = SourcingOption( scope=contract.scope, option_id="delivery-partner", mode="partner", option_evidence_digest=token("partner-option-profile"), annual_cost_units=900, launch_days=90, quality_passes=890, quality_trials=1_000, p95_latency_ms=390, vendor_concentration_permille=500, exit_readiness_permille=820, controls=controls( "partner", "delivery-partner", "partner", token("partner-option-profile"), ), ) return contract, SourcingEvidence( scope=contract.scope, contract_content_id=contract.content_id, options=(build, buy, partner), review_fresh=True, ) def main(): contract, evidence = illustrative_fixture() report = decide_sourcing(contract, evidence) print("example=illustrative_only") print(f"decision={report.decision}") print(f"selected={report.selected_option};mode={report.selected_mode}") print(f"eligible={','.join(report.eligible_options)}") print(f"violations={','.join(report.violations)}") print(f"claim={report.claim}") if __name__ == "__main__": main()