# License: Apache-2.0 License # Created by: Patrick Lumbantobing, VertoX-AI # Copyright (c) 2026 VertoX-AI. All rights reserved. # # This work is licensed under the Apache-2.0 License. # To view a copy of this license, visit # https://www.apache.org/licenses/LICENSE-2.0 """Define immutable backend-neutral alignment requests, results, and diagnostics. The records in this module are deliberately free of I/O and vendor types. Tuple-based key/value collections keep context and diagnostic values immutable while remaining straightforward to serialize at application boundaries. """ from __future__ import annotations from dataclasses import dataclass, field from enum import StrEnum from pathlib import Path type JsonScalar = str | int | float | bool | None type JsonValue = JsonScalar | tuple[JsonValue, ...] | tuple[tuple[str, JsonValue], ...] type Attributes = tuple[tuple[str, JsonValue], ...] class DiagnosticSeverity(StrEnum): """Classify how strongly a diagnostic affects an alignment outcome.""" INFO = "info" WARNING = "warning" ERROR = "error" @dataclass(frozen=True, slots=True) class Diagnostic: """Describe a stable, serializable event observed during alignment. Parameters ---------- code: Stable machine-facing diagnostic identifier. severity: Importance of the diagnostic. message: Safe human-facing explanation without secrets. stage: Pipeline stage that emitted the diagnostic. details: Immutable JSON-safe detail pairs. """ code: str severity: DiagnosticSeverity message: str stage: str details: Attributes = () @dataclass(frozen=True, slots=True) class AlignmentRequest: """Carry one authoritative transcript and its already-probed WAV metadata. Parameters ---------- request_id: Caller-defined unique identity; metadata services use row ordinal. audio_path: Resolved local WAV path. transcript: Exact authoritative transcript, never an ASR replacement. audio_duration_s: Positive finite WAV duration in seconds. context: Immutable JSON-safe caller identifiers. """ request_id: str audio_path: Path transcript: str audio_duration_s: float context: Attributes = () @dataclass(frozen=True, slots=True) class LexicalWord: """Identify one original-facing lexical word in a transcript. Parameters ---------- index: Zero-based lexical position. text: Exact surface substring from the authoritative transcript. start_char: Inclusive character offset in the transcript. end_char: Exclusive character offset in the transcript. """ index: int text: str start_char: int end_char: int @dataclass(frozen=True, slots=True) class WordAlignment: """Associate one original lexical word with an acoustic interval. Parameters ---------- lexical_word: Original-facing word identity and source span. start_s: Acoustic start in seconds. end_s: Acoustic end in seconds. confidence: Optional finite backend confidence. attributes: Immutable backend diagnostic attributes. """ lexical_word: LexicalWord start_s: float end_s: float confidence: float | None = None attributes: Attributes = () @dataclass(frozen=True, slots=True) class AlignmentProvenance: """Record backend, model, runtime, and artifact identity for an alignment. Parameters ---------- backend_name: Stable backend name. backend_version: Installed backend package or executable version. adapter_version: Project adapter version. model_id: Model/checkpoint identifier. model_revision: Immutable model revision when available. model_sha256: Local model/checkpoint hash when available. dictionary_id: Dictionary identifier when applicable. dictionary_revision: Immutable dictionary revision when available. dictionary_sha256: Local dictionary hash when available. python_version: Runtime interpreter version. torch_version: PyTorch version when applicable. cuda_version: CUDA version when applicable. device: CPU/GPU device description when applicable. config_sha256: Canonical redacted configuration fingerprint. native_command_sha256: Invoked native script/command fingerprint. artifacts: Immutable retained artifact references. attributes: Additional immutable JSON-safe provenance fields. """ backend_name: str backend_version: str | None = None adapter_version: str | None = None model_id: str | None = None model_revision: str | None = None model_sha256: str | None = None dictionary_id: str | None = None dictionary_revision: str | None = None dictionary_sha256: str | None = None python_version: str | None = None torch_version: str | None = None cuda_version: str | None = None device: str | None = None config_sha256: str | None = None native_command_sha256: str | None = None artifacts: tuple[str, ...] = () attributes: Attributes = () @dataclass(frozen=True, slots=True) class AlignmentResult: """Represent a complete successful word alignment for one request. Parameters ---------- request_id: Identity copied from the corresponding request. words: Complete ordered alignment tuple, one per lexical word. provenance: Backend and runtime identity. diagnostics: Non-fatal diagnostics retained with the result. runtime_s: Measured backend runtime in seconds. """ request_id: str words: tuple[WordAlignment, ...] provenance: AlignmentProvenance diagnostics: tuple[Diagnostic, ...] = () runtime_s: float = 0.0 @dataclass(frozen=True, slots=True) class AlignmentFailure: """Represent a structured failure without fabricated timestamps. Parameters ---------- request_id: Identity copied from the failed request. code: Stable machine-facing failure code. category: Stable broad failure category. message: Safe human-facing explanation. retryable: Whether retrying under unchanged inputs may reasonably succeed. diagnostics: Structured evidence collected before failure. provenance: Partial provenance when it was available. """ request_id: str code: str category: str message: str retryable: bool = False diagnostics: tuple[Diagnostic, ...] = () provenance: AlignmentProvenance | None = None @dataclass(frozen=True, slots=True) class AlignmentOutcome: """Contain exactly one successful result or structured failure. Parameters ---------- result: Successful alignment, mutually exclusive with ``failure``. failure: Failed alignment, mutually exclusive with ``result``. Raises ------ ValueError: If neither or both variants are supplied. """ result: AlignmentResult | None = field(default=None) failure: AlignmentFailure | None = field(default=None) def __post_init__(self) -> None: """Enforce the exactly-one variant invariant. Raises: ValueError: If the outcome is empty or contradictory. """ if (self.result is None) == (self.failure is None): raise ValueError("an alignment outcome must contain exactly one variant") @property def request_id(self) -> str: """Return the identity of the contained result or failure. Returns: Stable request identifier. """ variant = self.result if self.result is not None else self.failure assert variant is not None return variant.request_id @property def succeeded(self) -> bool: """Report whether this outcome contains a result. Returns: ``True`` for success and ``False`` for failure. """ return self.result is not None