"""Error hierarchy for marivo.semantic v1.1.
All errors flow through a single raise helper and share a common
string template. ErrorKind enum centralises every known error kind
with its associated hint factory.
"""
from __future__ import annotations
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from typing import Any, Literal, NoReturn
from marivo._authoring.errors import ContractScopeErrorPayload
from marivo._authoring.model import AuthoringRepair
from marivo._compat import StrEnum
from marivo.introspection.live.errors import HelpTargetErrorPayload
from marivo.introspection.live.model import LiveHelpTarget
from marivo.semantic.constraints import (
ConstraintId,
default_constraint_for_error_kind,
default_hint_for_error_kind,
get_constraint,
)
from marivo.semantic.ir import SourceLocation
__all__ = [
"HINTS",
"ErrorKind",
"SemanticContractScopeError",
"SemanticDecoratorError",
"SemanticDefinitionReadError",
"SemanticError",
"SemanticHelpTargetError",
"SemanticLoadError",
"SemanticLoadFailed",
"SemanticParityError",
"SemanticRuntimeError",
"StructuredWarning",
"WarningKind",
"_raise",
"repair",
]
# ---------------------------------------------------------------------------
# Repair helper
# ---------------------------------------------------------------------------
def repair(
*,
kind: Literal[
"retry",
"configure",
"register",
"reconnect",
"inspect",
"rescope",
"reacquire",
"reauthor",
"reload",
"environment",
"user_choice",
],
canonical_id: str,
action: str,
snippet: str | None = None,
candidates: tuple[str, ...] = (),
preserves_evidence: bool | None = None,
) -> AuthoringRepair:
"""Construct a semantic-owned typed repair."""
return AuthoringRepair(
kind=kind,
help_target=LiveHelpTarget(surface="semantic", canonical_id=canonical_id),
action=action,
snippet=snippet,
candidates=candidates,
preserves_evidence=preserves_evidence,
)
# ---------------------------------------------------------------------------
# ErrorKind enum
# ---------------------------------------------------------------------------
class ErrorKind(StrEnum):
"""Canonical error kind identifiers.
Grouped by phase (decorator-time, assembly-time, runtime, parity).
Every ``SemanticError.kind`` must be a member of this enum.
"""
# decorator-time
DUPLICATE_NAME = "duplicate_name"
MISSING_DOMAIN = "missing_domain"
MISSING_ENTITIES = "missing_entities"
INVALID_REF = "invalid_ref"
INVALID_FILTER = "invalid_filter"
INVALID_COMPOSITION = "invalid_composition"
INVALID_COMPONENT_BODY = "invalid_component_body"
OUTSIDE_LOADER_CONTEXT = "outside_loader_context"
METRIC_BODY_NOT_SINGLE_RETURN = "metric_body_not_single_return"
INVALID_AI_CONTEXT = "invalid_ai_context"
INVALID_DOMAIN_OWNER = "invalid_domain_owner"
SQL_ESCAPE_HATCH = "sql_escape_hatch"
IBIS_ATTR_SHADOW = "ibis_attr_shadow"
INVALID_EVENT_IDENTITY = "invalid_event_identity"
INVALID_EVENT_SOURCE = "invalid_event_source"
INVALID_EVENT_TIME = "invalid_event_time"
INVALID_EVENT_PREDICATE = "invalid_event_predicate"
INVALID_EVENT_PARTICIPANT_PATH = "invalid_event_participant_path"
INVALID_EVENT_PARTICIPANT_CARDINALITY = "invalid_event_participant_cardinality"
INVALID_STATE_MODEL = "invalid_state_model"
AMBIGUOUS_PARTICIPANT_ROLE = "ambiguous_participant_role"
MODEL_STATE_MISMATCH = "model_state_mismatch"
# assembly-time
DOMAIN_FILE_MISSING = "domain_file_missing"
DOMAIN_FILE_MISMATCH = "domain_file_mismatch"
MISSING_ENTITY_REF = "missing_entity_ref"
MISSING_DIMENSION_REF = "missing_dimension_ref"
MISSING_METRIC_REF = "missing_metric_ref"
CROSS_MODEL_CYCLE = "cross_model_cycle"
INVALID_RELATIONSHIP_ENDPOINT = "invalid_relationship_endpoint"
ORGANIZATION_ERROR = "organization_error"
INVALID_PROJECT = "invalid_project"
MISSING_METRIC_ADDITIVITY = "missing_metric_additivity"
MISSING_METRIC_ROOT_ENTITY = "missing_metric_root_entity"
INVALID_METRIC_ROOT_ENTITY = "invalid_metric_root_entity"
UNKNOWN_MEASURE = "unknown_measure"
MISSING_MEASURE_ADDITIVITY = "missing_measure_additivity"
INVALID_MEASURE_AGGREGATION = "invalid_measure_aggregation"
INCOMMENSURABLE_LINEAR_UNITS = "incommensurable_linear_units"
INVALID_VERIFICATION_MODE = "invalid_verification_mode"
INVALID_ENTITY_VERSIONING = "invalid_entity_versioning"
NON_ROOT_METRIC_AGGREGATE = "non_root_metric_aggregate"
INVALID_METRIC_FANOUT_POLICY = "invalid_metric_fanout_policy"
DERIVED_METRIC_FANOUT_POLICY = "derived_metric_fanout_policy"
DUPLICATE_DEFAULT_TIME_DIMENSION = "duplicate_default_time_dimension"
INVALID_SAMPLE_INTERVAL = "invalid_sample_interval"
INVALID_TIME_FOLD = "invalid_time_fold"
TIME_FOLD_REQUIRES_SEMI_ADDITIVE = "time_fold_requires_semi_additive"
TIME_FOLD_REQUIRES_SAMPLED_TIME_FIELD = "time_fold_requires_sampled_time_field"
MISSING_TIME_FOLD = "missing_time_fold"
MISSING_STATUS_TIME_DIMENSION = "missing_status_time_dimension"
INVALID_STATUS_TIME_DIMENSION = "invalid_status_time_dimension"
# runtime
NOT_FOUND = "not_found"
ENTITY_NOT_FOUND = "entity_not_found"
DIMENSION_NOT_FOUND = "dimension_not_found"
METRIC_NOT_FOUND = "metric_not_found"
MATERIALIZE_FAILED = "materialize_failed"
BACKEND_MISMATCH = "backend_mismatch"
COMPILE_ERROR = "compile_error"
DEFINITION_READ_FAILED = "definition_read_failed"
AMBIGUOUS_REFERENCE = "ambiguous_reference"
CROSS_DATASOURCE_NOT_SUPPORTED = "cross_datasource_not_supported"
BACKEND_FACTORY_REQUIRED = "backend_factory_required"
INSPECT_SOURCE_REQUIRED = "inspect_source_required"
PROJECT_NOT_LOADED = "project_not_loaded"
BINDING_CONTEXT_MISSING = "binding_context_missing"
INVALID_BINDING_REF = "invalid_binding_ref"
BINDING_ALIAS_NOT_DIRECT = "binding_alias_not_direct"
BINDING_ALIAS_AMBIGUOUS = "binding_alias_ambiguous"
BINDING_ENTITY_MISMATCH = "binding_entity_mismatch"
BINDING_NOT_DECLARED = "binding_not_declared"
BINDING_TARGET_MISSING = "binding_target_missing"
BINDING_CYCLE = "binding_cycle"
BINDING_RESULT_INVALID = "binding_result_invalid"
FILTER_VALUE_RUNTIME_INCOMPATIBLE = "filter_value_runtime_incompatible"
# parity
PROVENANCE_DIALECT_MISSING = "provenance_dialect_missing"
UNVERIFIED_PROVENANCE = "unverified_provenance"
PARITY_VALUE_MISMATCH = "parity_value_mismatch"
PARITY_NOT_SCALAR = "parity_not_scalar"
# ---------------------------------------------------------------------------
# Catalog-backed hint factories
# ---------------------------------------------------------------------------
def _hint_from_catalog(kind: ErrorKind, **_kwargs: Any) -> str:
hint = default_hint_for_error_kind(kind.value)
if hint is not None:
return hint
return 'Run marivo.help("semantic.constraints") to inspect semantic constraints.'
HINTS: dict[ErrorKind, Callable[..., str]] = {
kind: (lambda _kind=kind, **kwargs: _hint_from_catalog(_kind, **kwargs)) for kind in ErrorKind
}
# ---------------------------------------------------------------------------
# Error classes
# ---------------------------------------------------------------------------
class SemanticError(Exception):
"""Base class for all semantic errors.
Shared template for ``__str__``::
[kind] message
refs: ref1, ref2
at: file:line
hint: ...
"""
kind: str
semantic_refs: tuple[str, ...]
location: SourceLocation | None
hint: str | None
details: dict[str, Any]
constraint_id: str | None
repair: AuthoringRepair | None
expected: str | None
received: str | None
location_label: str | None
def __init__(
self,
*,
kind: str,
message: str,
refs: tuple[str, ...] = (),
location: SourceLocation | None = None,
hint: str | None = None,
details: dict[str, Any] | None = None,
constraint_id: ConstraintId | str | None = None,
repair: AuthoringRepair | None = None,
expected: str | None = None,
received: str | None = None,
location_label: str | None = None,
) -> None:
if constraint_id is None:
default_constraint = default_constraint_for_error_kind(kind)
constraint_id = default_constraint.id if default_constraint is not None else None
constraint = get_constraint(constraint_id) if constraint_id is not None else None
if hint is None and constraint is not None:
hint = constraint.hint
if repair is None:
from marivo.semantic._capabilities.registry import REGISTRY
repair_contract = REGISTRY.error_repair_contract(kind)
if repair_contract is not None:
repair = AuthoringRepair(
kind=repair_contract.kind,
help_target=repair_contract.help_target,
action=repair_contract.action,
snippet=repair_contract.snippet,
preserves_evidence=repair_contract.preserves_evidence,
)
self.kind = kind
self.message = message
self.semantic_refs = refs
self.location = location
self.hint = hint
self.details = details or {}
self.constraint_id = str(constraint_id) if constraint_id is not None else None
self.repair = repair
self.expected = expected
self.received = received
self.location_label = location_label
super().__init__(str(self))
def __str__(self) -> str:
lines: list[str] = [f"[{self.kind}] {self.message}"]
if self.semantic_refs:
lines.append(f" refs: {', '.join(self.semantic_refs)}")
if self.location is not None:
lines.append(f" at: {self.location.file}:{self.location.line}")
if self.location_label is not None:
lines.append(f" at: {self.location_label}")
if self.expected is not None:
lines.append(f" expected: {self.expected}")
if self.received is not None:
lines.append(f" received: {self.received}")
if self.hint is not None:
lines.append(f" hint: {self.hint}")
dym = self.details.get("did_you_mean")
if isinstance(dym, list) and dym:
lines.append(f" Did you mean: {', '.join(dym)}")
if self.repair is not None:
lines.extend(("", "Repair:", f" {self.repair.action}"))
if self.repair.snippet is not None:
lines.extend(f" {line}" for line in self.repair.snippet.splitlines())
if self.repair.candidates:
lines.append(f" Candidates: {', '.join(self.repair.candidates)}")
target = self.repair.help_target
if target.canonical_id is not None:
qualified = f"{target.surface}.{target.canonical_id}"
lines.append(f"Help: marivo.help({qualified!r})")
return "\n".join(lines)
class SemanticDecoratorError(SemanticError):
"""Error raised during decorator-time validation."""
[docs]
class SemanticDefinitionReadError(SemanticError):
"""A loaded definition cannot be faithfully projected to the public schema."""
def __init__(self, *, ref: str, location: SourceLocation, received: str) -> None:
"""Report the owning ref, declaration location and safe failure category.
Returns an error carrying expected/received facts and a reauthor repair.
Example: ``raise SemanticDefinitionReadError(ref=key, location=loc,
received="unknown metric composition")``. Never pass raw source or secrets.
"""
super().__init__(
kind=ErrorKind.DEFINITION_READ_FAILED,
message="The loaded semantic definition cannot be described safely.",
refs=(ref,),
location=location,
expected="a supported loaded declaration with JSON-safe parameters",
received=received,
hint="Inspect the named declaration and replace unsupported or non-JSON-safe parameters.",
repair=repair(
kind="reauthor",
canonical_id="authoring",
action=(
"Inspect the declaration at the reported location, replace unsupported or "
"non-JSON-safe parameters, reload the Catalog, and retry definition reading."
),
),
)
class SemanticLoadError(SemanticError):
"""Error raised during assembly-time (loader Pass 2) validation."""
class SemanticRuntimeError(SemanticError):
"""Error raised during runtime operations (materialize, compile)."""
class SemanticParityError(SemanticError):
"""Error raised during parity checking."""
class SemanticHelpTargetError(SemanticError):
"""Semantic-owned rejection of an unsupported live help target."""
def __init__(self, payload: HelpTargetErrorPayload) -> None:
owning_surface = payload.surface or "unknown"
super().__init__(
kind="not_found",
message=payload.message,
expected=f"accepted semantic help target ({', '.join(payload.accepted_kinds)})",
received=payload.received,
location_label=f"{owning_surface} help surface",
repair=AuthoringRepair(
kind="retry",
help_target=LiveHelpTarget(surface="semantic"),
action="Retry with a registered semantic help target.",
candidates=payload.candidates,
),
)
class SemanticContractScopeError(SemanticError):
"""Semantic-owned rejection of an over-broad contract request."""
def __init__(self, payload: ContractScopeErrorPayload) -> None:
super().__init__(
kind="ambiguous_reference",
message=payload.message,
expected=f"at most {payload.allowed_maximum} semantic subjects",
received=", ".join(payload.requested_subjects),
location_label="semantic contract scope",
repair=AuthoringRepair(
kind="retry",
help_target=payload.repair_target,
action="Narrow subject_refs to semantic-owned candidates.",
candidates=payload.owned_subjects,
),
)
class SemanticLoadFailed(Exception): # noqa: N818
"""Raised when reader methods are called on an errored project.
Wraps one or more SemanticError instances that prevented the
project from reaching the ready state.
"""
def __init__(self, errors: Sequence[SemanticError]) -> None:
self.errors = tuple(errors)
joined = "; ".join(str(error) for error in self.errors)
super().__init__(joined)
# ---------------------------------------------------------------------------
# Warning types
# ---------------------------------------------------------------------------
class WarningKind(StrEnum):
"""Canonical warning kind identifiers for non-fatal issues."""
STRING_REF = "string_ref"
UNVERIFIED_PROVENANCE = "unverified_provenance"
POTENTIALLY_FRAGILE_REFERENCE = "potentially_fragile_reference"
TIME_DIMENSION_PUSHDOWN_ADVISORY = "time_dimension_pushdown_advisory"
TIME_DIMENSION_DTYPE_ADVISORY = "time_dimension_dtype_advisory"
FILTERED_DOMAIN_REF = "filtered_domain_ref"
@dataclass(frozen=True)
class StructuredWarning:
"""Non-fatal warning produced during assembly validation.
Frozen dataclass matching the spec's structure.
"""
kind: Literal[
"string_ref",
"unverified_provenance",
"potentially_fragile_reference",
"time_dimension_pushdown_advisory",
"time_dimension_dtype_advisory",
"filtered_domain_ref",
]
message: str
refs: tuple[str, ...]
location: SourceLocation | None
def __str__(self) -> str:
lines: list[str] = [f"[{self.kind}] {self.message}"]
if self.refs:
lines.append(f" refs: {', '.join(self.refs)}")
if self.location is not None:
lines.append(f" at: {self.location.file}:{self.location.line}")
return "\n".join(lines)
def __repr__(self) -> str:
return (
f"StructuredWarning(kind={self.kind!r}, message={self.message!r}, "
f"refs={self.refs!r}, location={self.location!r})"
)
# ---------------------------------------------------------------------------
# Single raise helper
# ---------------------------------------------------------------------------
def _raise(
kind: ErrorKind,
message: str,
*,
cls: type[SemanticError] = SemanticDecoratorError,
refs: Sequence[str] = (),
location: SourceLocation | None = None,
hint: str | None = None,
details: dict[str, Any] | None = None,
expected: object | None = None,
received: object | None = None,
constraint_id: ConstraintId | str | None = None,
repair_value: AuthoringRepair | None = None,
) -> NoReturn:
"""Raise a structured SemanticError with hint from the HINTS registry."""
if hint is None:
constraint = get_constraint(constraint_id) if constraint_id is not None else None
if constraint is not None:
hint = constraint.hint
else:
hint_fn = HINTS.get(kind)
if hint_fn is not None:
hint = hint_fn()
resolved_details = dict(details or {})
if repair_value is None and kind in {
ErrorKind.INVALID_EVENT_IDENTITY,
ErrorKind.INVALID_EVENT_SOURCE,
ErrorKind.INVALID_EVENT_TIME,
ErrorKind.INVALID_EVENT_PREDICATE,
ErrorKind.INVALID_EVENT_PARTICIPANT_PATH,
ErrorKind.INVALID_EVENT_PARTICIPANT_CARDINALITY,
}:
participant_error = kind in {
ErrorKind.INVALID_EVENT_PARTICIPANT_PATH,
ErrorKind.INVALID_EVENT_PARTICIPANT_CARDINALITY,
}
repair_value = repair(
kind="reauthor",
canonical_id="participant" if participant_error else "objects.event",
action=hint
or (
"Reauthor the participant declaration."
if participant_error
else "Reauthor the Event definition."
),
snippet=(
"ms.participant(name='subject', cardinality='one')"
if participant_error
else "\n".join(
(
"@ms.event(",
" identity=(event_id,),",
" occurred_at=event_time,",
" participants=(",
" ms.participant(name='subject', cardinality='one'),",
" ),",
")",
"def business_event(rows):",
" return ms.all_rows()",
)
)
),
)
raise cls(
kind=kind.value,
message=message,
refs=tuple(refs),
location=location,
hint=hint,
details=resolved_details or None,
constraint_id=constraint_id,
repair=repair_value,
expected=str(expected) if expected is not None else None,
received=str(received) if received is not None else None,
)