"""Deterministic continuous-batching capacity model for invented traffic.""" 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}") def identity(value): if type(value) is not str or not IDENTITY.fullmatch(value): raise ValueError("exact bounded identity 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 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 identities(value, lower, upper): result = sequence(value, lower, upper) for item in result: identity(item) if len(set(result)) != len(result): raise ValueError("duplicate identity") return result def scope_tuple(value): return identities(value, 8, 8) 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 digest(value): encoded = json.dumps( value, sort_keys=True, separators=(",", ":"), allow_nan=False ).encode() 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) 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 ceil_div(numerator, denominator): integer(numerator, 0, 10**30) integer(denominator, 1, 10**30) return (numerator + denominator - 1) // denominator @dataclass(frozen=True) class BatchingContract: scope: tuple max_total_utilization_permille: int max_prefill_utilization_permille: int max_decode_utilization_permille: int max_queue_delay_ms: int max_ttft_ms: int max_tpot_ms: int max_starvation_ms: int max_kv_bytes: int content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) for value in ( self.max_total_utilization_permille, self.max_prefill_utilization_permille, self.max_decode_utilization_permille, ): integer(value, 1, 999) for value in ( self.max_queue_delay_ms, self.max_ttft_ms, self.max_tpot_ms, self.max_starvation_ms, ): integer(value, 1, 86_400_000) integer(self.max_kv_bytes, 1, 10**18) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("batching contract digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class SchedulerProfile: scope: tuple scheduler_id: str token_capacity_per_second: int prefill_share_permille: int max_active_sequences: int max_prefill_chunk_tokens: int iteration_time_ms: int max_consecutive_priority_iterations: int kv_bytes_per_token: int content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) identity(self.scheduler_id) integer(self.token_capacity_per_second, 1, 10**12) integer(self.prefill_share_permille, 1, 999) integer(self.max_active_sequences, 1, 1_000_000) integer(self.max_prefill_chunk_tokens, 1, 10_000_000) integer(self.iteration_time_ms, 1, 86_400_000) integer(self.max_consecutive_priority_iterations, 1, 1_000_000) integer(self.kv_bytes_per_token, 1, 10**12) prefill_capacity = ( self.token_capacity_per_second * self.prefill_share_permille ) // 1_000 prefill_tokens_per_iteration = ( prefill_capacity * self.iteration_time_ms ) // 1_000 if prefill_tokens_per_iteration < 1: raise ValueError("prefill partition cannot serve one token per iteration") if self.max_prefill_chunk_tokens > prefill_tokens_per_iteration: raise ValueError("prefill chunk exceeds declared per-iteration service") expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("scheduler profile digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class RequestClass: scope: tuple class_id: str priority: int arrival_millirps: int prompt_tokens: int output_tokens: int peak_concurrency: int content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) identity(self.class_id) integer(self.priority, 1, 1_000) integer(self.arrival_millirps, 1, 10**12) integer(self.prompt_tokens, 1, 10_000_000) integer(self.output_tokens, 1, 10_000_000) integer(self.peak_concurrency, 1, 1_000_000) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("request class digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class BatchingWorkload: scope: tuple contract_content_id: str scheduler: SchedulerProfile request_classes: tuple admission_control_enabled: bool content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) sha256_text(self.contract_content_id) scheduler = validate_record(self.scheduler, SchedulerProfile) classes = tuple( validate_record(item, RequestClass) for item in sequence(self.request_classes, 1, 64) ) if scheduler.scope != self.scope or any(item.scope != self.scope for item in classes): raise ValueError("scheduler or request class outside workload scope") if len({item.class_id for item in classes}) != len(classes): raise ValueError("duplicate request class identity") exact_bool(self.admission_control_enabled) object.__setattr__(self, "scheduler", scheduler) object.__setattr__(self, "request_classes", classes) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("batching workload digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class BatchingReport: decision: str violations: tuple total_utilization_permille: int prefill_utilization_permille: int decode_utilization_permille: int estimated_queue_delay_ms: int estimated_worst_ttft_ms: int estimated_tpot_ms: int declared_starvation_bound_ms: int peak_kv_bytes: int evidence_id: str claim: str = "LOCAL_FLUID_CAPACITY_MODEL_NOT_QUEUEING_OR_DEVICE_BENCHMARK" def audit_batching( contract: BatchingContract, workload: BatchingWorkload ) -> BatchingReport: """Evaluate one invented fluid traffic model without running a scheduler.""" contract = validate_record(contract, BatchingContract) workload = validate_record(workload, BatchingWorkload) if workload.scope != contract.scope or workload.contract_content_id != contract.content_id: raise ValueError("workload belongs to another batching contract") scheduler = workload.scheduler total_required_milli = sum( item.arrival_millirps * (item.prompt_tokens + item.output_tokens) for item in workload.request_classes ) prefill_required_milli = sum( item.arrival_millirps * item.prompt_tokens for item in workload.request_classes ) decode_required_milli = sum( item.arrival_millirps * item.output_tokens for item in workload.request_classes ) prefill_capacity = ( scheduler.token_capacity_per_second * scheduler.prefill_share_permille ) // 1_000 decode_capacity = scheduler.token_capacity_per_second - prefill_capacity total_utilization = ceil_div(total_required_milli, scheduler.token_capacity_per_second) prefill_utilization = ceil_div(prefill_required_milli, prefill_capacity) decode_utilization = ceil_div(decode_required_milli, decode_capacity) if total_utilization >= 1_000: queue_delay = 86_400_000 else: queue_delay = ceil_div( scheduler.iteration_time_ms * total_utilization, 1_000 - total_utilization, ) worst_prefill_service = max( max( ceil_div(item.prompt_tokens, scheduler.max_prefill_chunk_tokens) * scheduler.iteration_time_ms, ceil_div(item.prompt_tokens * 1_000, prefill_capacity), ) for item in workload.request_classes ) worst_ttft = queue_delay + worst_prefill_service starvation_bound = ( scheduler.max_consecutive_priority_iterations + 1 ) * scheduler.iteration_time_ms active_sequences = sum(item.peak_concurrency for item in workload.request_classes) estimated_tpot = max( scheduler.iteration_time_ms, ceil_div(active_sequences * 1_000, decode_capacity), ) peak_kv_bytes = sum( item.peak_concurrency * (item.prompt_tokens + item.output_tokens) * scheduler.kv_bytes_per_token for item in workload.request_classes ) violations = [] if not workload.admission_control_enabled: violations.append("admission-control-disabled") if active_sequences > scheduler.max_active_sequences: violations.append("active-sequence-capacity") if total_utilization > contract.max_total_utilization_permille: violations.append("total-utilization") if prefill_utilization > contract.max_prefill_utilization_permille: violations.append("prefill-utilization") if decode_utilization > contract.max_decode_utilization_permille: violations.append("decode-utilization") if queue_delay > contract.max_queue_delay_ms: violations.append("queue-delay") if worst_ttft > contract.max_ttft_ms: violations.append("time-to-first-token") if estimated_tpot > contract.max_tpot_ms: violations.append("time-per-output-token") if starvation_bound > contract.max_starvation_ms: violations.append("starvation-bound") if peak_kv_bytes > contract.max_kv_bytes: violations.append("kv-capacity") decision = "REJECT_CAPACITY" if violations else "CAPACITY_WITHIN_DECLARED_BOUND" evidence_id = digest( { "contract": contract.content_id, "workload": workload.content_id, "decision": decision, "violations": violations, "utilization": [total_utilization, prefill_utilization, decode_utilization], "queue_delay_ms": queue_delay, "peak_kv_bytes": peak_kv_bytes, } ) return BatchingReport( decision, tuple(violations), total_utilization, prefill_utilization, decode_utilization, queue_delay, worst_ttft, estimated_tpot, starvation_bound, peak_kv_bytes, evidence_id, ) def illustrative_fixture(): scope = ( "batching-policy-v1", "traffic-mix-v1", "tokenizer-v3", "model-profile-v7", "scheduler-profile-v4", "kv-layout-v2", "latency-slo-v3", "fixture-v1", ) contract = BatchingContract(scope, 800, 700, 700, 50, 250, 30, 100, 10_000_000_000) scheduler = SchedulerProfile( scope, "chunked-prefill-fair-v4", 40_000, 600, 64, 480, 20, 3, 131_072, ) classes = ( RequestClass(scope, "interactive-short", 1, 4_000, 1_000, 200, 24), RequestClass(scope, "interactive-long", 2, 500, 4_000, 800, 8), ) workload = BatchingWorkload( scope, contract.content_id, scheduler, classes, True, ) return contract, workload def main(): report = audit_batching(*illustrative_fixture()) print("example=illustrative_only") print(f"decision={report.decision}") print( f"utilization_permille={report.total_utilization_permille};" f"prefill={report.prefill_utilization_permille};" f"decode={report.decode_utilization_permille}" ) print( f"queue_ms={report.estimated_queue_delay_ms};" f"worst_ttft_ms={report.estimated_worst_ttft_ms};" f"tpot_ms={report.estimated_tpot_ms}" ) print(f"peak_kv_bytes={report.peak_kv_bytes}") print(f"claim={report.claim}") if __name__ == "__main__": main()