generated from john/python-template
155 lines
5.1 KiB
Python
155 lines
5.1 KiB
Python
"""Versioned, provider-neutral contracts for processing evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import platform
|
|
from importlib.metadata import PackageNotFoundError
|
|
from importlib.metadata import version
|
|
from typing import Any
|
|
from typing import Literal
|
|
from uuid import UUID
|
|
|
|
from pydantic import BaseModel
|
|
from pydantic import ConfigDict
|
|
from pydantic import Field
|
|
from pydantic import JsonValue
|
|
|
|
REQUEST_MANIFEST_SCHEMA = "transcription.request-manifest"
|
|
REQUEST_MANIFEST_VERSION = "1"
|
|
SOFTWARE_CONTEXT_SCHEMA = "transcription.software-context"
|
|
SOFTWARE_CONTEXT_VERSION = "1"
|
|
TRANSPORT_EVIDENCE_SCHEMA = "transcription.transport-evidence"
|
|
TRANSPORT_EVIDENCE_VERSION = "1"
|
|
CANONICAL_JSON_ALGORITHM = "transcription-canonical-json-v1"
|
|
|
|
SAFE_RESPONSE_HEADERS = frozenset(
|
|
{
|
|
"content-type",
|
|
"content-encoding",
|
|
"date",
|
|
"retry-after",
|
|
"x-request-id",
|
|
"x-openrouter-generation-id",
|
|
"x-ratelimit-limit",
|
|
"x-ratelimit-remaining",
|
|
"x-ratelimit-reset",
|
|
}
|
|
)
|
|
|
|
|
|
class EvidenceModel(BaseModel):
|
|
"""Strict immutable base for persisted evidence contracts."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
|
|
class SourceEvidenceReference(EvidenceModel):
|
|
"""Secret-safe identity for source content used by one execution."""
|
|
|
|
source_id: UUID
|
|
digest_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
|
|
byte_size: int = Field(ge=0)
|
|
media_type: str = Field(min_length=1)
|
|
page_number: int = Field(ge=1)
|
|
width: int | None = Field(default=None, ge=1)
|
|
height: int | None = Field(default=None, ge=1)
|
|
derivative_id: UUID | None = None
|
|
transformation: str | None = None
|
|
|
|
|
|
class SoftwareContext(EvidenceModel):
|
|
"""Versions needed to interpret a provider execution."""
|
|
|
|
schema_name: Literal["transcription.software-context"] = SOFTWARE_CONTEXT_SCHEMA
|
|
schema_version: Literal["1"] = SOFTWARE_CONTEXT_VERSION
|
|
application_version: str
|
|
application_commit: str | None = None
|
|
adapter_name: str
|
|
adapter_version: str
|
|
client_library: str
|
|
client_library_version: str
|
|
python_version: str
|
|
|
|
|
|
class RequestManifest(EvidenceModel):
|
|
"""Frozen, secret-safe representation of one concrete provider request."""
|
|
|
|
schema_name: Literal["transcription.request-manifest"] = REQUEST_MANIFEST_SCHEMA
|
|
schema_version: Literal["1"] = REQUEST_MANIFEST_VERSION
|
|
provider: str = Field(min_length=1)
|
|
requested_model: str = Field(min_length=1)
|
|
request: dict[str, JsonValue]
|
|
source: SourceEvidenceReference
|
|
explicitly_supplied_parameters: tuple[str, ...] = ()
|
|
omitted_optional_parameters: tuple[str, ...] = ()
|
|
optional_parameter_states: dict[str, Literal["omitted", "null", "value"]]
|
|
prompt_content: str = Field(min_length=1)
|
|
prompt_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
|
|
timeout_seconds: float = Field(gt=0)
|
|
retry_policy: str = Field(min_length=1)
|
|
software: SoftwareContext
|
|
canonicalization: Literal["transcription-canonical-json-v1"] = CANONICAL_JSON_ALGORITHM
|
|
|
|
def canonical_bytes(self) -> bytes:
|
|
return canonical_json_bytes(self.model_dump(mode="json"))
|
|
|
|
def digest(self) -> str:
|
|
return hashlib.sha256(self.canonical_bytes()).hexdigest()
|
|
|
|
|
|
class TransportEvidence(EvidenceModel):
|
|
"""Exact response captured at the application/router HTTP boundary."""
|
|
|
|
schema_name: Literal["transcription.transport-evidence"] = TRANSPORT_EVIDENCE_SCHEMA
|
|
schema_version: Literal["1"] = TRANSPORT_EVIDENCE_VERSION
|
|
response_received: bool
|
|
status_code: int | None = Field(default=None, ge=100, le=599)
|
|
body: bytes | None = None
|
|
safe_headers: dict[str, str] = Field(default_factory=dict)
|
|
content_type: str | None = None
|
|
content_encoding: str | None = None
|
|
request_id: str | None = None
|
|
generation_id: str | None = None
|
|
|
|
|
|
def canonical_json_bytes(value: Any) -> bytes:
|
|
"""Serialize JSON deterministically for evidence integrity hashes."""
|
|
return json.dumps(
|
|
value,
|
|
ensure_ascii=False,
|
|
allow_nan=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
|
|
|
|
def filter_safe_response_headers(headers: Any) -> dict[str, str]:
|
|
"""Return only explicitly allowlisted response headers."""
|
|
return {
|
|
str(name).lower(): str(value) for name, value in headers.items() if str(name).lower() in SAFE_RESPONSE_HEADERS
|
|
}
|
|
|
|
|
|
def package_version(package: str) -> str:
|
|
"""Return an installed package version without failing evidence capture."""
|
|
try:
|
|
return version(package)
|
|
except PackageNotFoundError:
|
|
return "unknown"
|
|
|
|
|
|
def build_software_context(*, adapter_name: str, adapter_version: str, client_library: str) -> SoftwareContext:
|
|
"""Build the runtime software identity for an execution."""
|
|
return SoftwareContext(
|
|
application_version=package_version("transcription"),
|
|
application_commit=os.environ.get("TRANSCRIPTION_COMMIT") or None,
|
|
adapter_name=adapter_name,
|
|
adapter_version=adapter_version,
|
|
client_library=client_library,
|
|
client_library_version=package_version(client_library),
|
|
python_version=platform.python_version(),
|
|
)
|