"""recoil.lib.exceptions — Canonical exception types per Tenet 6. Tenet 6 (Errors Must Be Visible) requires that the engine fails loudly by default. This module is the single home for the engine's typed exception families. Provider-local exception families (LipSyncError in execution/providers/sync_so.py, AudioSynthesisError in elevenlabs.py, EvalProviderError in gemini_vision.py) remain in their provider modules — those are localized to one transport / one upstream API and do not benefit from consolidation. The types here are the engine-wide canonical errors. Importers across recoil/lib, recoil/core, recoil/pipeline, recoil/execution, recoil/workspace all bind to these names. Provenance: - RecoilError, SchemaValidationError, SidecarFieldError, CrossConfigDriftError: introduced as Phase A.1 temp at recoil/lib/_phase_a_exceptions.py; canonicalized here in Phase E.4. - CostMissingError: introduced as Phase C temp at recoil/pipeline/core/cost.py; canonicalized here in Phase E.5. - UnknownFailureEscalation: introduced as Phase C temp at recoil/pipeline/core/failure_mode.py; canonicalized here in Phase E.5. - PayloadHintsValidationError: introduced as Phase C temp at recoil/execution/providers/payload_hints.py; canonicalized here in Phase E.5. - RetryExhaustedError, SanctionedFallbackError, RefDimensionUnknownError: introduced new in Phase E.2. - SidecarCorruptError, VerdictCorruptError, WorkspaceStateCorruptError (+ subclasses CastingFragmentCorruptError, EditorialConfigCorruptError, RecommendationsCorruptError, PromptCompilerOverridesCorruptError): introduced new in Phase E.2 from Phase E.1 inventory's corruption-family cluster (8 of top-30 sites). - ConfigParseError, PromptValidatorConfigError, KeyframeContextLookupError, MediaProbeError, ExecutionStoreUnavailableError, VerdictAutofillError: introduced new in Phase E.2 from Phase E.1 inventory's lookup/config cluster. CP-9 lock: production_loop.py is byte-untouched. Phase E does not import these types from production_loop. """ from __future__ import annotations from typing import Literal, Optional WorkspaceStateKind = Literal[ "state", "casting fragment", "editorial config", "recommendations", "prompt-compiler overrides", "chat sessions", ] # --------------------------------------------------------------------------- # Base # --------------------------------------------------------------------------- class RecoilError(Exception): """Base for all Recoil typed exceptions. Originated in Phase A.1's _phase_a_exceptions.py. Canonicalized here in Phase E.2; the temp module re-exports for one cycle (Phase E.4 closes the loop). """ pass # --------------------------------------------------------------------------- # Validation / shape errors (Phase A.1 family — preserve signatures exactly) # --------------------------------------------------------------------------- class SchemaValidationError(RecoilError): """Raised when a config file fails schema validation at load time. Replaces the 'json.loads then .get(key, default)' pattern with fail-loud validation per Tenet 6. """ pass class SidecarFieldError(RecoilError, ValueError): """Raised when a sidecar write attempts an out-of-contract field. Frozen-contract field set per data-contracts.md §1a. Inherits from ValueError for API compatibility with existing callers. """ pass class CrossConfigDriftError(RecoilError): """Raised when cross-config validation detects drift. e.g. a model id in provider_strategy.json that doesn't exist in model_profiles.json. Per Tenet 6. """ pass # --------------------------------------------------------------------------- # Cost / metadata errors (Phase C family — preserve signatures exactly) # --------------------------------------------------------------------------- class CostMissingError(RuntimeError): """Raised when a RunResult or record has no readable cost_usd. Tenet 6: the default disposition for missing cost is to fail loud. Callers that legitimately tolerate missing cost (display, aggregation) use read_cost_from_result_safe() instead, which logs WARNING and returns 0.0. """ def __init__( self, *, result_id: Optional[str] = None, source: str = "unknown", ): self.result_id = result_id self.source = source msg = ( f"CostMissingError: result_id={result_id!r} source={source!r} " f"has no cost_usd. If this is a legitimate billing-zero case, " f"the upstream writer must store cost_usd=0.0 explicitly. If " f"missing is tolerable for this read site, use " f"read_cost_from_result_safe() and accept the WARNING log." ) super().__init__(msg) # --------------------------------------------------------------------------- # Failure-mode classifier (Phase C — preserve signature exactly) # --------------------------------------------------------------------------- class UnknownFailureEscalation(RuntimeError): """Raised by classify_failure() when an error cannot be classified. Tenet 6 (Errors Must Be Visible): unknown error shapes ESCALATE rather than silently defaulting to TRANSIENT (retry forever) or PERMANENT (give up silently). The exception carries the full classification context so the operator can grep for "UnknownFailureEscalation" in logs and immediately see what input was unclassifiable. Callers that legitimately need to handle unknown-as-data (e.g., an aggregation pass over historical errors) opt in via classify_failure(..., escalate_unknown=False); this returns (FailureMode.UNKNOWN, 0.0) without raising. """ def __init__( self, *, error_text: Optional[str] = None, gate_verdict: Optional[object] = None, http_status: Optional[int] = None, caller: Optional[str] = None, ): self.error_text = error_text self.gate_verdict = gate_verdict self.http_status = http_status self.caller = caller msg = ( f"UnknownFailureEscalation: classify_failure could not " f"classify input. caller={caller!r} error_text={error_text!r} " f"http_status={http_status!r} gate_verdict={gate_verdict!r}. " f"Add the missing pattern to " f"pipeline.core.failure_mode and re-run." ) super().__init__(msg) # --------------------------------------------------------------------------- # Payload hints (Phase C — preserve signature exactly) # --------------------------------------------------------------------------- class PayloadHintsValidationError(ValueError): """Raised when a PayloadHints subclass receives unknown keys. Reserved for the follow-on sprint that flips extra="allow" → "forbid". Not raised in Phase C. """ # --------------------------------------------------------------------------- # Retry exhaustion # --------------------------------------------------------------------------- class RetryExhaustedError(RuntimeError): """A retry helper exhausted its attempt budget without success. Per Tenet 6: a retry exhausting all attempts must NOT return None or a sentinel "give up" RunResult. It either raises this error (caller does not bind to a `RunResult`) or returns a `RunResult` with explicit `failure_mode = RETRY_EXHAUSTED` (caller does bind). Phase E ships the raise variant. """ def __init__( self, *, attempts: int, last_error: BaseException | None = None, operation: str = "", ): self.attempts = attempts self.last_error = last_error self.operation = operation msg = f"retry exhausted after {attempts} attempts" if operation: msg = f"{msg} for {operation}" if last_error: msg = f"{msg}; last error: {last_error.__class__.__name__}: {last_error}" super().__init__(msg) # --------------------------------------------------------------------------- # Sanctioned-fallback signaling # --------------------------------------------------------------------------- class SanctionedFallbackError(RuntimeError): """Base class for errors raised from within sanctioned-fallback paths. Most sanctioned fallbacks DO NOT raise — they log + return a substituted value. This class is for the rare case where a fallback path itself cannot complete (e.g., the fallback resolver also failed). Subclasses in this module enumerate the specific cases. Tenet 6 carve-out: a fallback firing is observable via log.warning("FALLBACK_FIRED ..."); a fallback FAILING is observable via this exception type. """ # --------------------------------------------------------------------------- # Asset / probe errors # --------------------------------------------------------------------------- class RefDimensionUnknownError(ValueError): """A ref image's pixel dimensions could not be probed. Replaces silent `return None` from `recoil/core/ref_resolver.py:113` (and its alias re-export at `recoil/pipeline/lib/ref_resolver.py:71`). Callers either propagate (preferred — the prompt builder's aspect-ratio computation cannot proceed without dimensions) or catch and substitute a documented default (rare; only call sites that explicitly choose to skip the ref). """ def __init__(self, path: str, message: str = ""): self.path = path full = f"could not probe dimensions for {path}" if message: full += f": {message}" super().__init__(full) class MediaProbeError(RuntimeError): """Raised when ffprobe / image probe / metadata extraction fails on a media file (video / image / audio) and the caller cannot recover via fallback. """ def __init__(self, path: str, probe_kind: str = "media", message: str = ""): self.path = path self.probe_kind = probe_kind full = f"{probe_kind} probe failed for {path}" if message: full += f": {message}" super().__init__(full) # --------------------------------------------------------------------------- # Sidecar / verdict / workspace-state corruption family # (introduced from Phase E.1 inventory's corruption cluster — 8 of top-30) # # SidecarCorruptError and VerdictCorruptError are intentionally siblings # (not subclasses) of WorkspaceStateCorruptError. Sidecars and verdicts are # engine-level frozen-shape artifacts (per data-contracts.md §1a/§1d); # WorkspaceStateCorruptError covers editor-level UI state files. Different # layers, different blast radius — keep the corruption family split. # --------------------------------------------------------------------------- class SidecarCorruptError(ValueError): """Raised when a sidecar JSON file is unreadable or malformed. Replaces silent ``except Exception: return None`` in workspace/sidecar.py and reuse sites in workspace/server.py / workspace/verdict.py. """ def __init__(self, path: str, message: str = ""): self.path = path full = f"corrupt sidecar at {path}" if message: full += f": {message}" super().__init__(full) class VerdictCorruptError(ValueError): """Raised when a verdict YAML / JSON file is unreadable or malformed.""" def __init__(self, path: str, message: str = ""): self.path = path full = f"corrupt verdict at {path}" if message: full += f": {message}" super().__init__(full) class WorkspaceStateCorruptError(ValueError): """Raised when a workspace state JSON file (recommendations, casting fragment, editorial config, etc.) is unreadable or malformed. Common base for narrower subclasses below; usable directly when no narrower class fits. """ def __init__( self, path: str, kind: WorkspaceStateKind = "state", message: str = "", ): self.path = path self.kind = kind full = f"corrupt {kind} at {path}" if message: full += f": {message}" super().__init__(full) class CastingFragmentCorruptError(WorkspaceStateCorruptError): """Casting-fragment subclass.""" def __init__(self, path: str, message: str = ""): super().__init__(path=path, kind="casting fragment", message=message) class EditorialConfigCorruptError(WorkspaceStateCorruptError): """Editorial-config subclass.""" def __init__(self, path: str, message: str = ""): super().__init__(path=path, kind="editorial config", message=message) class RecommendationsCorruptError(WorkspaceStateCorruptError): """Recommendations-state subclass.""" def __init__(self, path: str, message: str = ""): super().__init__(path=path, kind="recommendations", message=message) class PromptCompilerOverridesCorruptError(WorkspaceStateCorruptError): """Prompt-compiler overrides subclass.""" def __init__(self, path: str, message: str = ""): super().__init__(path=path, kind="prompt-compiler overrides", message=message) class ChatSessionsCorruptError(WorkspaceStateCorruptError): """Chat-sessions store subclass (~/.recoil/chat-sessions.json).""" def __init__(self, path: str, message: str = ""): super().__init__(path=path, kind="chat sessions", message=message) class ProposalCorruptError(WorkspaceStateCorruptError): """Proposal JSON file subclass (~/.recoil/proposals//.json).""" def __init__(self, path: str, message: str = ""): super().__init__(path=path, kind="proposal", message=message) class ExecutionStoreCorruptError(RecoilError): """Raised when a shot JSON file in ExecutionStore fails to decode. Replaces the prior silent `json.JSONDecodeError → log + return None` behavior. The operator must triage the corrupt file (rm + re-emit, or git restore). """ def __init__(self, path: str, parse_error: str) -> None: self.path = path self.parse_error = parse_error super().__init__( f"ExecutionStoreCorruptError: {path}: {parse_error} — " "manual triage required (rm + regenerate or git restore)" ) # --------------------------------------------------------------------------- # Config / lookup errors # (introduced from Phase E.1 inventory's lookup cluster) # --------------------------------------------------------------------------- class ConfigParseError(ValueError): """Raised when a config file (json/yaml) parses but its shape is invalid. Distinct from SchemaValidationError (raised by Pydantic schemas); this is raised by hand-written validators that pre-date Phase A.1. """ def __init__(self, path: str, message: str = ""): self.path = path full = f"could not parse config {path}" if message: full += f": {message}" super().__init__(full) class PromptValidatorConfigError(ConfigParseError): """Prompt-validator-config subclass — surfaces the offending validator key.""" def __init__(self, path: str, validator_name: str = "", message: str = ""): self.validator_name = validator_name prefix = f"validator={validator_name!r} " if validator_name else "" super().__init__(path=path, message=f"{prefix}{message}".strip()) class KeyframeContextLookupError(KeyError): """Raised when a keyframe context lookup fails (missing shot id, missing pass id, etc.) where the prior code returned None silently. """ def __init__(self, lookup_kind: str, key: str, message: str = ""): self.lookup_kind = lookup_kind self.key = key full = f"keyframe context {lookup_kind} lookup failed for key={key!r}" if message: full += f": {message}" super().__init__(full) class ExecutionStoreUnavailableError(RuntimeError): """Raised by `verdict_executionstore_*` callers when the ExecutionStore is unreachable in the current process (typical: cross-process readers). Callers may catch + fire `verdict_executionstore_unavailable` sanctioned fallback. For raisers, this is the canonical type. """ def __init__(self, message: str = ""): full = "execution store unavailable in current process" if message: full += f": {message}" super().__init__(full) class VerdictAutofillError(RuntimeError): """Raised when a verdict-autofill operation fails for a reason other than ExecutionStore unavailability (e.g., the autofill source itself raised, or a downstream merge step rejected the autofill). """ def __init__(self, *, source: str = "", message: str = ""): self.source = source full = "verdict autofill failed" if source: full += f" (source={source})" if message: full += f": {message}" super().__init__(full) class ModelProfileLookupError(KeyError): """Raised when a model-profile lookup fails (missing model id, missing rate field, or schema mismatch) where the prior code returned a hard- coded default cost / feature flag silently. Inherits from KeyError so existing callers that catch KeyError on ``get_profile(...)`` still trip; the typed name lets callers narrow when they want to distinguish profile lookup failure from any other KeyError in the call site. """ def __init__(self, model_id: str, message: str = ""): self.model_id = model_id full = f"model profile lookup failed for model_id={model_id!r}" if message: full += f": {message}" super().__init__(full) # --------------------------------------------------------------------------- # Public surface # --------------------------------------------------------------------------- __all__ = [ # Base "RecoilError", # Phase A.1 family "SchemaValidationError", "SidecarFieldError", "CrossConfigDriftError", # Phase C family "CostMissingError", "UnknownFailureEscalation", "PayloadHintsValidationError", # Phase E.2 new "RetryExhaustedError", "SanctionedFallbackError", "RefDimensionUnknownError", "MediaProbeError", # Corruption family (Phase E.2 from inventory) "SidecarCorruptError", "VerdictCorruptError", "WorkspaceStateCorruptError", "CastingFragmentCorruptError", "EditorialConfigCorruptError", "RecommendationsCorruptError", "PromptCompilerOverridesCorruptError", "ChatSessionsCorruptError", "ProposalCorruptError", "ExecutionStoreCorruptError", # Config/lookup family (Phase E.2 from inventory) "ConfigParseError", "PromptValidatorConfigError", "KeyframeContextLookupError", "ExecutionStoreUnavailableError", "VerdictAutofillError", "ModelProfileLookupError", ]