#!/usr/bin/env python3 """Typed, standard-library DeltaX Evaluate Developer Kit reference client. Historical beta1 teaching client, not the live beta2 client. Hosted beta2 is in restricted controlled free soak; this client retains the beta1 network guard. The module has no default hostname, performs no automatic retries, refuses `.invalid` design placeholders, and validates the response authority boundary. Running this file is always offline: its CLI only validates and renders a synthetic fixture. Importing ``DeltaXClient`` does not create a network client; call ``from_environment`` only after separate endpoint and credential access is explicitly issued. """ from __future__ import annotations import argparse import dataclasses import datetime as dt import json import math import os import re import sys import urllib.error import urllib.parse import urllib.request import uuid from typing import Any, Dict, List, Mapping, NoReturn, Optional, Sequence, Union API_VERSION = "2026-09-16.beta1" PROFILE_ID = "deltax-hosted-bounded-review-evaluation-v1" ROUTE = "/v1/evaluations" SCOPE = "deltax:evaluate" MAX_REQUEST_BYTES = 32_768 MAX_RESPONSE_BYTES = 262_144 DEFAULT_TIMEOUT_SECONDS = 5.0 PLANNED_UNAVAILABLE_ORIGIN = "https://api.deltaxevaluate.com" _ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._~-]{0,63}$") _IDEMPOTENCY_RE = re.compile(r"^[A-Za-z0-9._~-]{16,128}$") _CONTEXT_KEY_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{0,63}$") _DIGEST_RE = re.compile(r"^[a-f0-9]{64}$") _REASON_RE = re.compile(r"^[a-z0-9_]{1,96}$") SELECTABLE_CLASSES = frozenset({"analyze", "compare", "draft", "validate"}) SEMANTIC_CLASSES = SELECTABLE_CLASSES | {"external"} OUTCOMES = frozenset({"selected", "governed_noop_refusal"}) REASON_CODES = frozenset( { "candidate_selected_within_bounded_evaluation", "no_submitted_candidate_passed_all_gates", "all_submitted_candidates_shadow_only", "insufficient_bounded_support", "identity_or_authority_gate_closed", "resource_or_trace_gate_closed", "policy_conflict", } ) EVIDENCE_LABELS = frozenset( {"operator_reported", "contract_validated", "runtime_observed", "inferred", "unresolved"} ) GATE_IDS = tuple(f"HES-G{index:02d}" for index in range(1, 13)) TRACE_FALSE_FLAGS = ( "submitted_candidate_executed", "external_effect", "learning_applied", "operator_reported_context_verified", ) AUTHORITY_FALSE_FLAGS = ( "selected_is_executed", "selected_is_correct_or_safe", "response_authorizes_downstream_action", "response_is_professional_advice", "trace_is_certification_or_legal_non_repudiation", ) ERROR_STATUS_TO_CODE = { 400: "invalid_request", 401: "unauthenticated", 403: "forbidden", 408: "request_timeout", 409: "idempotency_conflict", 413: "payload_too_large", 415: "unsupported_media_type", 422: "unprocessable_content", 429: "rate_limited", 431: "headers_too_large", 500: "internal_failure", 503: "service_unavailable", } RETRYABLE_STATUSES = frozenset({408, 429, 503}) RETRY_AFTER_REQUIRED_STATUSES = frozenset({429, 503}) Scalar = Union[str, int, float, bool, None] class DeltaXClientError(Exception): """Base exception for local contract or transport failures.""" class ContractViolation(DeltaXClientError): """Raised when a request or response violates the frozen contract.""" class TransportError(DeltaXClientError): """Raised for a single-attempt network or protocol failure.""" class DeltaXAPIError(DeltaXClientError): """A validated API error envelope. The client never retries it.""" def __init__( self, *, status: int, code: str, message: str, retryable: bool, retry_after_seconds: Optional[int], request_id: Optional[str], ) -> None: super().__init__(f"DeltaX API error {status} ({code}): {message}") self.status = status self.code = code self.retryable = retryable self.retry_after_seconds = retry_after_seconds self.request_id = request_id def _closed_keys(value: Mapping[str, Any], required: Sequence[str], label: str) -> None: required_set = set(required) actual = set(value) missing = sorted(required_set - actual) extra = sorted(actual - required_set) if missing or extra: raise ContractViolation(f"{label} keys are not closed; missing={missing}, extra={extra}") def _require_string(value: Any, label: str, minimum: int, maximum: int) -> str: if not isinstance(value, str) or not minimum <= len(value) <= maximum: raise ContractViolation(f"{label} must be a string of {minimum}–{maximum} characters") return value def _require_uuid(value: Any, label: str) -> str: text = _require_string(value, label, 1, 64) try: uuid.UUID(text) except (ValueError, AttributeError) as exc: raise ContractViolation(f"{label} must be a UUID") from exc return text def _reject_json_constant(value: str) -> NoReturn: raise ContractViolation(f"non-finite JSON number is not admitted: {value}") def _reject_duplicate_pairs(pairs: Sequence[tuple[str, Any]]) -> Dict[str, Any]: result: Dict[str, Any] = {} for key, value in pairs: if key in result: raise ContractViolation(f"duplicate JSON object key is not admitted: {key}") result[key] = value return result def strict_json_loads(raw: bytes) -> Any: try: return json.loads( raw.decode("utf-8"), object_pairs_hook=_reject_duplicate_pairs, parse_constant=_reject_json_constant, ) except UnicodeDecodeError as exc: raise ContractViolation("response is not valid UTF-8") from exc except json.JSONDecodeError as exc: raise ContractViolation("response is not valid JSON") from exc @dataclasses.dataclass(frozen=True) class Candidate: candidate_id: str semantic_class: str description: str operator_reported_support: float @classmethod def from_mapping(cls, value: Mapping[str, Any]) -> "Candidate": _closed_keys( value, ("candidate_id", "semantic_class", "description", "operator_reported_support"), "candidate", ) candidate_id = _require_string(value["candidate_id"], "candidate_id", 1, 64) if not _ID_RE.fullmatch(candidate_id): raise ContractViolation("candidate_id does not match the closed pattern") semantic_class = value["semantic_class"] if not isinstance(semantic_class, str) or semantic_class not in SEMANTIC_CLASSES: raise ContractViolation("semantic_class is outside the closed catalog") description = _require_string(value["description"], "description", 1, 512) support = value["operator_reported_support"] if isinstance(support, bool) or not isinstance(support, (int, float)): raise ContractViolation("operator_reported_support must be a finite number") numeric_support = float(support) if not math.isfinite(numeric_support) or not 0 <= numeric_support <= 1: raise ContractViolation("operator_reported_support must be between 0 and 1") return cls(candidate_id, semantic_class, description, numeric_support) def as_dict(self) -> Dict[str, Any]: return dataclasses.asdict(self) @dataclasses.dataclass(frozen=True) class EvaluationRequest: objective: str context: Mapping[str, Scalar] candidates: Sequence[Candidate] profile_id: str = PROFILE_ID @classmethod def from_mapping(cls, value: Mapping[str, Any]) -> "EvaluationRequest": _closed_keys(value, ("profile_id", "objective", "context", "candidates"), "request") if value["profile_id"] != PROFILE_ID: raise ContractViolation("profile_id must equal the one frozen profile") objective = _require_string(value["objective"], "objective", 1, 1024) context = value["context"] if not isinstance(context, dict) or len(context) > 16: raise ContractViolation("context must be an object with at most 16 properties") checked_context: Dict[str, Scalar] = {} for key, scalar in context.items(): if not isinstance(key, str) or not _CONTEXT_KEY_RE.fullmatch(key): raise ContractViolation(f"invalid context key: {key!r}") if scalar is None or isinstance(scalar, bool): checked_context[key] = scalar elif isinstance(scalar, str): if len(scalar) > 256: raise ContractViolation(f"context string {key!r} exceeds 256 characters") checked_context[key] = scalar elif isinstance(scalar, (int, float)): number = float(scalar) if not math.isfinite(number) or not -1_000_000_000 <= number <= 1_000_000_000: raise ContractViolation(f"context number {key!r} is outside the admitted range") checked_context[key] = scalar else: raise ContractViolation(f"context value {key!r} must be scalar") raw_candidates = value["candidates"] if not isinstance(raw_candidates, list) or not 1 <= len(raw_candidates) <= 8: raise ContractViolation("candidates must contain one to eight items") candidates = tuple(Candidate.from_mapping(item) for item in raw_candidates) ids = [candidate.candidate_id for candidate in candidates] if len(set(ids)) != len(ids): raise ContractViolation("candidate_id values must be unique within the request") return cls(objective=objective, context=checked_context, candidates=candidates) def as_dict(self) -> Dict[str, Any]: return { "profile_id": self.profile_id, "objective": self.objective, "context": dict(self.context), "candidates": [candidate.as_dict() for candidate in self.candidates], } def encoded(self) -> bytes: try: body = json.dumps( self.as_dict(), ensure_ascii=False, allow_nan=False, separators=(",", ":"), ).encode("utf-8") except (TypeError, ValueError) as exc: raise ContractViolation("request cannot be encoded as strict JSON") from exc if len(body) > MAX_REQUEST_BYTES: raise ContractViolation(f"encoded request exceeds {MAX_REQUEST_BYTES} bytes") return body class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): def redirect_request(self, request: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> None: return None class DeltaXClient: """Historical beta1 single-attempt teaching client.""" def __init__(self, *, base_url: str, access_token: str, timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS) -> None: parsed = urllib.parse.urlsplit(base_url) hostname = (parsed.hostname or "").lower() if parsed.scheme != "https" or not hostname or parsed.username or parsed.password: raise ContractViolation("base_url must be an HTTPS origin without embedded credentials") if parsed.path not in {"", "/"} or parsed.query or parsed.fragment: raise ContractViolation("base_url must be an HTTPS origin without path, query, or fragment components") if hostname.endswith(".invalid") or hostname == "invalid": raise ContractViolation("the .invalid design placeholder cannot be used as a live endpoint") planned_hostname = urllib.parse.urlsplit(PLANNED_UNAVAILABLE_ORIGIN).hostname if hostname.rstrip(".") == planned_hostname: raise ContractViolation("the pinned historical beta1 client does not admit the live beta2 hostname") if not access_token or access_token.strip() != access_token or "\r" in access_token or "\n" in access_token: raise ContractViolation("access_token is missing or contains unsafe whitespace") if not 0 < timeout_seconds <= DEFAULT_TIMEOUT_SECONDS: raise ContractViolation("timeout_seconds must be greater than zero and at most five seconds") self._base_url = base_url.rstrip("/") self._access_token = access_token self._timeout_seconds = timeout_seconds self._opener = urllib.request.build_opener(_NoRedirectHandler()) @classmethod def from_environment(cls) -> "DeltaXClient": """Require separately issued values; there are deliberately no defaults.""" base_url = os.environ.get("DELTAX_BASE_URL") access_token = os.environ.get("DELTAX_ACCESS_TOKEN") if not base_url or not access_token: raise ContractViolation( "DELTAX_BASE_URL and DELTAX_ACCESS_TOKEN must be separately issued and explicitly set" ) return cls(base_url=base_url, access_token=access_token) def evaluate(self, request: EvaluationRequest, *, idempotency_key: str) -> Mapping[str, Any]: """Perform exactly one POST. This method never executes the selected candidate.""" if not _IDEMPOTENCY_RE.fullmatch(idempotency_key): raise ContractViolation("idempotency_key must match ^[A-Za-z0-9._~-]{16,128}$") checked_request = EvaluationRequest.from_mapping(request.as_dict()) candidate_by_id = { candidate.candidate_id: candidate for candidate in checked_request.candidates } http_request = urllib.request.Request( self._base_url + ROUTE, data=checked_request.encoded(), method="POST", headers={ "Authorization": f"Bearer {self._access_token}", "Content-Type": "application/json", "Accept": "application/json, application/problem+json", "X-DeltaX-API-Version": API_VERSION, "Idempotency-Key": idempotency_key, }, ) try: with self._opener.open(http_request, timeout=self._timeout_seconds) as response: status = response.status content_type = response.headers.get_content_type() response_headers = response.headers raw = response.read(MAX_RESPONSE_BYTES + 1) except urllib.error.HTTPError as exc: raw = exc.read(MAX_RESPONSE_BYTES + 1) self._raise_api_error(exc.code, exc.headers, raw) except (urllib.error.URLError, TimeoutError, OSError) as exc: raise TransportError("single DeltaX request failed; no retry was attempted") from exc if len(raw) > MAX_RESPONSE_BYTES: raise ContractViolation("response exceeds the local safety limit") if status != 200 or content_type != "application/json": raise ContractViolation(f"unexpected success protocol: status={status}, content_type={content_type}") payload = strict_json_loads(raw) if not isinstance(payload, dict): raise ContractViolation("success response must be a JSON object") validate_evaluation_response(payload, candidate_by_id) _validate_common_headers(response_headers, payload["request_id"]) _validate_rate_limit_headers(response_headers) return payload def _raise_api_error(self, status: int, headers: Any, raw: bytes) -> NoReturn: if len(raw) > MAX_RESPONSE_BYTES: raise ContractViolation("error response exceeds the local safety limit") content_type = headers.get_content_type() if status not in ERROR_STATUS_TO_CODE or content_type != "application/problem+json": raise TransportError(f"unexpected HTTP error protocol: status={status}, content_type={content_type}") payload = strict_json_loads(raw) if not isinstance(payload, dict): raise ContractViolation("error response must be a JSON object") error = validate_error_response(payload, status=status) _validate_common_headers(headers, payload["request_id"]) if status == 429: _validate_rate_limit_headers(headers) retry_after_header = headers.get("Retry-After") if status in RETRY_AFTER_REQUIRED_STATUSES: if retry_after_header is None or not retry_after_header.isdigit(): raise ContractViolation("Retry-After is required and must contain whole seconds") if int(retry_after_header) != error["retry_after_seconds"]: raise ContractViolation("Retry-After header disagrees with the error body") raise DeltaXAPIError( status=status, code=error["code"], message=error["message"], retryable=error["retryable"], retry_after_seconds=error["retry_after_seconds"], request_id=payload["request_id"], ) def _validate_common_headers(headers: Any, payload_request_id: Optional[str]) -> None: if headers.get("X-DeltaX-API-Version") != API_VERSION: raise ContractViolation("response X-DeltaX-API-Version is missing or mismatched") header_request_id = headers.get("X-DeltaX-Request-ID") _require_uuid(header_request_id, "X-DeltaX-Request-ID") if payload_request_id is not None and header_request_id != payload_request_id: raise ContractViolation("response header and body request IDs disagree") def _validate_rate_limit_headers(headers: Any) -> None: limit = headers.get("X-RateLimit-Limit") remaining = headers.get("X-RateLimit-Remaining") reset = headers.get("X-RateLimit-Reset") if limit != "60": raise ContractViolation("X-RateLimit-Limit must equal 60") if remaining is None or not remaining.isdigit() or not 0 <= int(remaining) <= 60: raise ContractViolation("X-RateLimit-Remaining must be an integer from 0 through 60") if reset is None or not reset.isdigit() or int(reset) < 0: raise ContractViolation("X-RateLimit-Reset must be a non-negative Unix timestamp") def validate_evaluation_response( payload: Mapping[str, Any], candidate_by_id: Mapping[str, Candidate] ) -> None: _closed_keys( payload, ( "request_id", "api_version", "profile_id", "outcome", "selected_candidate_id", "reason_codes", "trace", "authority", ), "response", ) _require_uuid(payload["request_id"], "request_id") if payload["api_version"] != API_VERSION or payload["profile_id"] != PROFILE_ID: raise ContractViolation("response version or profile does not match the frozen request contract") outcome = payload["outcome"] if outcome not in OUTCOMES: raise ContractViolation("response outcome is outside the closed catalog") selected = payload["selected_candidate_id"] if outcome == "selected": if selected not in candidate_by_id or candidate_by_id[selected].semantic_class == "external": raise ContractViolation("selected_candidate_id must name a submitted non-external candidate") elif selected is not None: raise ContractViolation("governed_noop_refusal requires selected_candidate_id=null") reason_codes = payload["reason_codes"] if ( not isinstance(reason_codes, list) or not 1 <= len(reason_codes) <= 16 or len(set(reason_codes)) != len(reason_codes) or any(code not in REASON_CODES for code in reason_codes) ): raise ContractViolation("reason_codes violate the closed response contract") trace = payload["trace"] if not isinstance(trace, dict): raise ContractViolation("trace must be an object") trace_keys = ( "trace_id", "request_digest", "candidate_field_digest", "release_set_digest", "policy_digest", "lambda_decisions", "evidence_labels", *TRACE_FALSE_FLAGS, "committed_at", ) _closed_keys(trace, trace_keys, "trace") _require_uuid(trace["trace_id"], "trace_id") for field in ("request_digest", "candidate_field_digest", "release_set_digest", "policy_digest"): if not isinstance(trace[field], str) or not _DIGEST_RE.fullmatch(trace[field]): raise ContractViolation(f"{field} must be a lowercase SHA-256 digest") decisions = trace["lambda_decisions"] if not isinstance(decisions, list) or len(decisions) != 12: raise ContractViolation("lambda_decisions must contain exactly 12 entries") seen_gates = set() for decision in decisions: if not isinstance(decision, dict): raise ContractViolation("each lambda decision must be an object") _closed_keys(decision, ("gate_id", "disposition", "reason_code"), "lambda decision") if decision["gate_id"] not in GATE_IDS or decision["gate_id"] in seen_gates: raise ContractViolation("lambda gate IDs must be the 12 unique HES gates") if decision["disposition"] not in {"pass", "fail"}: raise ContractViolation("lambda disposition must be pass or fail") if not isinstance(decision["reason_code"], str) or not _REASON_RE.fullmatch(decision["reason_code"]): raise ContractViolation("lambda reason_code violates the closed pattern") seen_gates.add(decision["gate_id"]) if seen_gates != set(GATE_IDS): raise ContractViolation("lambda_decisions do not cover HES-G01 through HES-G12") labels = trace["evidence_labels"] if ( not isinstance(labels, list) or not 1 <= len(labels) <= 8 or len(set(labels)) != len(labels) or any(label not in EVIDENCE_LABELS for label in labels) ): raise ContractViolation("evidence_labels violate the closed catalog") for flag in TRACE_FALSE_FLAGS: if trace[flag] is not False: raise ContractViolation(f"trace authority flag must remain false: {flag}") committed_at = _require_string(trace["committed_at"], "committed_at", 1, 64) try: parsed_committed_at = dt.datetime.fromisoformat(committed_at.replace("Z", "+00:00")) except ValueError as exc: raise ContractViolation("committed_at must be an RFC 3339 date-time") from exc if parsed_committed_at.tzinfo is None: raise ContractViolation("committed_at must include an RFC 3339 UTC offset") authority = payload["authority"] if not isinstance(authority, dict): raise ContractViolation("authority must be an object") _closed_keys(authority, AUTHORITY_FALSE_FLAGS, "authority") for flag in AUTHORITY_FALSE_FLAGS: if authority[flag] is not False: raise ContractViolation(f"response authority flag must remain false: {flag}") def validate_error_response(payload: Mapping[str, Any], *, status: int) -> Mapping[str, Any]: _closed_keys(payload, ("error", "request_id"), "error response") request_id = payload["request_id"] if request_id is not None: _require_uuid(request_id, "request_id") error = payload["error"] if not isinstance(error, dict): raise ContractViolation("error must be an object") _closed_keys(error, ("code", "message", "retryable", "retry_after_seconds"), "error body") if status not in ERROR_STATUS_TO_CODE or error["code"] != ERROR_STATUS_TO_CODE[status]: raise ContractViolation("error code does not match HTTP status") _require_string(error["message"], "error message", 1, 256) expected_retryable = status in RETRYABLE_STATUSES if error["retryable"] is not expected_retryable: raise ContractViolation("retryable flag does not match the frozen status contract") retry_after = error["retry_after_seconds"] if status in RETRY_AFTER_REQUIRED_STATUSES: if isinstance(retry_after, bool) or not isinstance(retry_after, int) or not 1 <= retry_after <= 60: raise ContractViolation("429 and 503 require retry_after_seconds from 1 through 60") elif retry_after is not None: if isinstance(retry_after, bool) or not isinstance(retry_after, int) or not 1 <= retry_after <= 60: raise ContractViolation("retry_after_seconds must be null or an integer from 1 through 60") return error def _offline_main(argv: Optional[Sequence[str]] = None) -> int: parser = argparse.ArgumentParser(description="Validate and render one synthetic request; never use the network.") parser.add_argument("--fixture", required=True, help="Path to fixtures/cases.json") parser.add_argument("--case", default="selected", help="Success case ID to render") args = parser.parse_args(argv) with open(args.fixture, "rb") as handle: fixture_set = strict_json_loads(handle.read()) matches = [case for case in fixture_set.get("cases", []) if case.get("case_id") == args.case] if len(matches) != 1 or matches[0].get("kind") != "success": raise ContractViolation("--case must name exactly one synthetic success fixture") request = EvaluationRequest.from_mapping(matches[0]["request"]) print(request.encoded().decode("utf-8")) print("offline_only=true", file=sys.stderr) print(f"route=POST {ROUTE}", file=sys.stderr) print(f"X-DeltaX-API-Version={API_VERSION}", file=sys.stderr) print("network_request_sent=false", file=sys.stderr) return 0 if __name__ == "__main__": try: raise SystemExit(_offline_main()) except DeltaXClientError as exc: print(f"contract error: {exc}", file=sys.stderr) raise SystemExit(2)