"""Semantic readiness report DTOs and query-free evidence gating."""
from __future__ import annotations
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, field, replace
from datetime import datetime
from typing import TYPE_CHECKING, ClassVar, Literal, cast
from marivo._authoring.model import AuthoringRepair
from marivo._compat import UTC
from marivo.refs import Ref, RefPayloadV1, SemanticKind, SemanticKindTag
from marivo.refs import ref as ref_factory
from marivo.render import Card, RenderableResult
from marivo.semantic.errors import repair
from marivo.semantic.runtime_metric import RuntimeMetricExpr, replay_payload
if TYPE_CHECKING:
from marivo.semantic._metric_resolution import MetricTemporalContract
from marivo.semantic.reader import SemanticProject
from marivo.semantic.validator import Registry
ReadinessStatus = Literal["ready", "ready_with_warnings", "blocked"]
ReadinessSeverity = Literal["blocker", "warning", "advisory"]
ReadinessIssueKind = Literal[
"load_error",
"unknown_ref",
"cross_datasource_unfederated",
"sql_parity_unverified",
"fragile_string_ref",
"time_dimension_pushdown_advisory",
"undeclared_naive_time_axis",
"metric_graph_invalid",
"snapshot_fold_unobservable",
"state_model_seed_missing",
"period_calendar_artifact_missing",
"period_calendar_artifact_stale",
"period_calendar_artifact_invalid",
"temporal_set_artifact_missing",
"temporal_set_artifact_stale",
"temporal_set_artifact_invalid",
"work_schedule_artifact_missing",
"work_schedule_artifact_stale",
"work_schedule_artifact_invalid",
]
[docs]
@dataclass(frozen=True)
class ReadinessIssue:
kind: ReadinessIssueKind
severity: ReadinessSeverity
refs: tuple[str, ...]
message: str
repair: AuthoringRepair | None = None
details: Mapping[str, object] = field(default_factory=dict)
catalog_definition_fingerprint: str | None = None
def to_dict(self) -> dict[str, object]:
payload: dict[str, object] = {
"kind": self.kind,
"severity": self.severity,
"refs": list(self.refs),
"message": self.message,
"repair": self.repair.model_dump() if self.repair is not None else None,
"catalog_definition_fingerprint": self.catalog_definition_fingerprint,
}
if self.details:
payload["details"] = dict(self.details)
return payload
[docs]
@dataclass(frozen=True, repr=False)
class ReadinessReport(RenderableResult):
scope: ClassVar[Literal["semantic_static"]] = "semantic_static"
status: ReadinessStatus
analysis_ready_inputs: tuple[Ref[SemanticKindTag] | RuntimeMetricExpr, ...]
blockers: tuple[ReadinessIssue, ...]
warnings: tuple[ReadinessIssue, ...]
input_summary: ReadinessInputSummary
checked_at: str
catalog_definition_fingerprint: str | None = None
def _repr_identity(self) -> str:
return (
f"ReadinessReport scope={self.scope} status={self.status} "
f"issues={len(self.blockers) + len(self.warnings)}"
)
def _card(self) -> Card:
card = Card(
identity=self._repr_identity(),
available=(
".show()",
".to_dict()",
".analysis_ready_inputs",
),
)
card = card.field(label="scope", value=self.scope)
if self.blockers:
blocker_items = [
f"{i.kind}: {i.message} -> fix: {i.repair.action if i.repair else ''}"
for i in self.blockers
]
card = card.listing(
label=f"blockers ({len(self.blockers)})",
items=tuple(blocker_items),
)
actual_warnings = tuple(issue for issue in self.warnings if issue.severity == "warning")
advisories = tuple(issue for issue in self.warnings if issue.severity == "advisory")
if actual_warnings:
warning_items = [
f"{i.kind}: {i.message} -> fix: {i.repair.action if i.repair else ''}"
for i in actual_warnings
]
card = card.listing(
label=f"warnings ({len(actual_warnings)})",
items=tuple(warning_items),
)
if advisories:
card = card.listing(
label=f"advisories ({len(advisories)})",
items=tuple(
f"{i.kind}: {i.message} -> optional fix: {i.repair.action if i.repair else ''}"
for i in advisories
),
)
ready_refs = tuple(item for item in self.analysis_ready_inputs if type(item) is Ref)
if ready_refs:
card = card.field(
label="analysis_ready",
value=", ".join(ref.key for ref in ready_refs),
)
runtime_inputs = tuple(item for item in self.analysis_ready_inputs if type(item) is not Ref)
if runtime_inputs:
card = card.field(
label="analysis_ready_runtime",
value=", ".join(item.label for item in runtime_inputs),
)
return card.field(label="checked_at", value=self.checked_at)
def to_dict(self) -> dict[str, object]:
return {
"scope": self.scope,
"status": self.status,
"analysis_ready_inputs": [
RefPayloadV1.from_ref(item).to_dict() if type(item) is Ref else replay_payload(item)
for item in self.analysis_ready_inputs
],
"blockers": [issue.to_dict() for issue in self.blockers],
"warnings": [issue.to_dict() for issue in self.warnings],
"input_summary": self.input_summary.to_dict(),
"checked_at": self.checked_at,
"catalog_definition_fingerprint": self.catalog_definition_fingerprint,
}
def _exact_ref(path: str, kind: SemanticKind) -> Ref[SemanticKindTag]:
factory = {
SemanticKind.DOMAIN: ref_factory.domain,
SemanticKind.DATASOURCE: ref_factory.datasource,
SemanticKind.ENTITY: ref_factory.entity,
SemanticKind.DIMENSION: ref_factory.dimension,
SemanticKind.TIME_DIMENSION: ref_factory.time_dimension,
SemanticKind.MEASURE: ref_factory.measure,
SemanticKind.METRIC: ref_factory.metric,
SemanticKind.RELATIONSHIP: ref_factory.relationship,
SemanticKind.EVENT: ref_factory.event,
SemanticKind.STATE_MODEL: ref_factory.state_model,
SemanticKind.PERIOD_CALENDAR: ref_factory.period_calendar,
SemanticKind.TEMPORAL_SET: ref_factory.temporal_set,
SemanticKind.WORK_SCHEDULE: ref_factory.work_schedule,
}[kind]
return factory(path)
def _exact_key(path: str, kind: SemanticKind) -> str:
return _exact_ref(path, kind).key
def _display_path(ref_key: str) -> str:
prefix, separator, path = ref_key.partition(":")
if separator and prefix in {kind.value for kind in SemanticKind}:
return path
return ref_key
_EXECUTABLE_KINDS = frozenset(
{
SemanticKind.ENTITY,
SemanticKind.DIMENSION,
SemanticKind.TIME_DIMENSION,
SemanticKind.MEASURE,
SemanticKind.METRIC,
SemanticKind.RELATIONSHIP,
SemanticKind.EVENT,
SemanticKind.STATE_MODEL,
}
)
def _checked_at() -> str:
return datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z")
def _status(blockers: list[ReadinessIssue], warnings: list[ReadinessIssue]) -> ReadinessStatus:
if blockers:
return "blocked"
if any(issue.severity == "warning" for issue in warnings):
return "ready_with_warnings"
return "ready"
def _dedupe(values: Iterable[str]) -> tuple[str, ...]:
seen: set[str] = set()
out: list[str] = []
for value in values:
if value not in seen:
seen.add(value)
out.append(value)
return tuple(out)
def _issue(
kind: ReadinessIssueKind,
severity: ReadinessSeverity,
refs: Iterable[str],
message: str,
repair: AuthoringRepair | None = None,
*,
details: Mapping[str, object] | None = None,
) -> ReadinessIssue:
return ReadinessIssue(
kind=kind,
severity=severity,
refs=_dedupe(refs),
message=message,
repair=repair,
details={} if details is None else details,
)
def _parity_passed(project: SemanticProject, ref: str) -> bool:
"""Check whether a metric with SQL provenance has passed parity verification."""
parity_result = project._parity_results.get(ref)
return parity_result is not None and parity_result.ok
def _object_maps(project: SemanticProject) -> tuple[dict[str, SemanticKind], dict[str, object]]:
reg = project._registry
if reg is None:
return {}, {}
kinds: dict[str, SemanticKind] = {}
objects: dict[str, object] = {}
for entity in reg.entities.values():
key = _exact_key(entity.semantic_id, SemanticKind.ENTITY)
kinds[key] = SemanticKind.ENTITY
objects[key] = entity
for dim in reg.dimensions.values():
kind = SemanticKind.TIME_DIMENSION if dim.is_time_dimension else SemanticKind.DIMENSION
key = _exact_key(dim.semantic_id, kind)
kinds[key] = kind
objects[key] = dim
for measure in reg.measures.values():
key = _exact_key(measure.semantic_id, SemanticKind.MEASURE)
kinds[key] = SemanticKind.MEASURE
objects[key] = measure
for metric in reg.metrics.values():
key = _exact_key(metric.semantic_id, SemanticKind.METRIC)
kinds[key] = SemanticKind.METRIC
objects[key] = metric
for relationship in reg.relationships.values():
key = _exact_key(relationship.semantic_id, SemanticKind.RELATIONSHIP)
kinds[key] = SemanticKind.RELATIONSHIP
objects[key] = relationship
for event in reg.events.values():
key = _exact_key(event.semantic_id, SemanticKind.EVENT)
kinds[key] = SemanticKind.EVENT
objects[key] = event
for state_model in reg.state_models.values():
key = _exact_key(state_model.semantic_id, SemanticKind.STATE_MODEL)
kinds[key] = SemanticKind.STATE_MODEL
objects[key] = state_model
for calendar in reg.period_calendars.values():
key = _exact_key(calendar.semantic_id, SemanticKind.PERIOD_CALENDAR)
kinds[key] = SemanticKind.PERIOD_CALENDAR
objects[key] = calendar
for temporal_set in reg.temporal_sets.values():
key = _exact_key(temporal_set.semantic_id, SemanticKind.TEMPORAL_SET)
kinds[key] = SemanticKind.TEMPORAL_SET
objects[key] = temporal_set
for work_schedule in reg.work_schedules.values():
key = _exact_key(work_schedule.semantic_id, SemanticKind.WORK_SCHEDULE)
kinds[key] = SemanticKind.WORK_SCHEDULE
objects[key] = work_schedule
for domain_ir in reg.domains.values():
key = _exact_key(domain_ir.name, SemanticKind.DOMAIN)
kinds[key] = SemanticKind.DOMAIN
objects[key] = domain_ir
for ds_ir in project._datasource_irs or reg.datasources.values():
key = _exact_key(ds_ir.semantic_id, SemanticKind.DATASOURCE)
kinds[key] = SemanticKind.DATASOURCE
objects[key] = ds_ir
return kinds, objects
def _scope_keys(
refs: Iterable[Ref[SemanticKindTag] | str] | None,
kinds: Mapping[str, SemanticKind],
) -> tuple[str, ...] | None:
if refs is None:
return None
keys: list[str] = []
for ref in refs:
if type(ref) is Ref:
keys.append(ref.key)
continue
candidates = tuple(key for key in kinds if _display_path(key) == ref)
keys.append(candidates[0] if len(candidates) == 1 else ref)
return tuple(keys)
_CONTAINER_KINDS = frozenset(
{SemanticKind.RELATIONSHIP, SemanticKind.DOMAIN, SemanticKind.DATASOURCE}
)
def _default_checked_refs(kinds: Mapping[str, SemanticKind]) -> tuple[str, ...]:
return tuple(ref for ref in kinds if kinds[ref] not in _CONTAINER_KINDS) + tuple(
ref for ref in kinds if kinds[ref] in _CONTAINER_KINDS
)
def _dependencies_for_ref(
ref: str,
objects: Mapping[str, object],
kinds: Mapping[str, SemanticKind],
*,
event_predicate_dependencies: Mapping[str, tuple[str, ...]] | None = None,
) -> tuple[str, ...]:
kind = kinds.get(ref)
obj = objects.get(ref)
if obj is None:
return ()
path = _display_path(ref)
if kind == SemanticKind.DOMAIN:
return tuple(
obj_id
for obj_id, other in objects.items()
if kinds.get(obj_id) == SemanticKind.ENTITY and getattr(other, "domain", None) == path
)
if kind == SemanticKind.DATASOURCE:
return tuple(
obj_id
for obj_id, other in objects.items()
if kinds.get(obj_id) == SemanticKind.ENTITY
and getattr(other, "datasource", None) == path
)
if kind in {SemanticKind.DIMENSION, SemanticKind.TIME_DIMENSION}:
entity = getattr(obj, "entity", None)
return (_exact_key(entity, SemanticKind.ENTITY),) if isinstance(entity, str) else ()
if kind == SemanticKind.MEASURE:
entity = getattr(obj, "entity", None)
return (_exact_key(entity, SemanticKind.ENTITY),) if isinstance(entity, str) else ()
if kind == SemanticKind.PERIOD_CALENDAR:
date_field = getattr(obj, "date", None)
levels = tuple(field for _level, field in getattr(obj, "levels", ()))
correspondence_fields = tuple(
field for _name, _level, field in getattr(obj, "correspondences", ())
)
return tuple(
_exact_key(
field,
SemanticKind.TIME_DIMENSION if field == date_field else SemanticKind.DIMENSION,
)
for field in (date_field, *levels, *correspondence_fields)
if isinstance(field, str)
)
if kind == SemanticKind.TEMPORAL_SET:
fields = (
getattr(obj, "occurrence_id", None),
getattr(obj, "start", None),
getattr(obj, "end", None),
getattr(obj, "category", None),
)
return tuple(
_exact_key(
field,
SemanticKind.TIME_DIMENSION
if field in {getattr(obj, "start", None), getattr(obj, "end", None)}
else SemanticKind.DIMENSION,
)
for field in fields
if isinstance(field, str)
)
if kind == SemanticKind.WORK_SCHEDULE:
date_field = getattr(obj, "date", None)
status_field = getattr(obj, "is_working", None)
return tuple(
_exact_key(
field,
SemanticKind.TIME_DIMENSION if field == date_field else SemanticKind.DIMENSION,
)
for field in (date_field, status_field)
if isinstance(field, str)
)
if kind == SemanticKind.METRIC:
deps: list[str] = []
deps.extend(
_exact_key(entity, SemanticKind.ENTITY) for entity in getattr(obj, "entities", ())
)
composition = getattr(obj, "composition", None)
if composition is not None:
from marivo.semantic.ir import composition_components
components = composition_components(composition)
deps.extend(
_exact_key(str(value), SemanticKind.METRIC) for value in components.values()
)
return tuple(deps)
if kind == SemanticKind.RELATIONSHIP:
keys = getattr(obj, "keys", ())
key_refs = tuple(ref for key in keys for ref in key.to_tuple())
entity_deps = (getattr(obj, "from_entity", None), getattr(obj, "to_entity", None))
return tuple(
_exact_key(dep, SemanticKind.ENTITY) for dep in entity_deps if isinstance(dep, str)
) + tuple(
_exact_key(dep, SemanticKind.DIMENSION)
for dep in (*key_refs, *getattr(obj, "from_keys", ()), *getattr(obj, "to_keys", ()))
if isinstance(dep, str)
)
if kind == SemanticKind.EVENT:
from marivo.semantic.ir import EventIR
event = cast("EventIR", obj)
deps = [
_exact_key(event.source_entity, SemanticKind.ENTITY),
_exact_key(event.occurred_at, SemanticKind.TIME_DIMENSION),
]
deps.extend(_exact_key(path, SemanticKind.DIMENSION) for path in event.identity)
for participant in event.participants:
deps.extend(
_exact_key(path, SemanticKind.RELATIONSHIP) for path in (participant.path or ())
)
if event_predicate_dependencies is not None:
deps.extend(event_predicate_dependencies.get(ref, ()))
return tuple(deps)
if kind == SemanticKind.STATE_MODEL:
from marivo.semantic.ir import StateModelIR
model = cast("StateModelIR", obj)
event_refs = {item.trigger.event_ref for item in model.inceptions} | {
item.trigger.event_ref for item in model.transitions
}
return (
_exact_key(model.subject, SemanticKind.ENTITY),
*tuple(_exact_key(event_ref, SemanticKind.EVENT) for event_ref in sorted(event_refs)),
)
return ()
def _expand_checked_refs(
refs: Iterable[str] | None,
kinds: Mapping[str, SemanticKind],
objects: Mapping[str, object],
*,
event_predicate_dependencies: Mapping[str, tuple[str, ...]] | None = None,
) -> tuple[tuple[str, ...], tuple[str, ...]]:
seeds = _dedupe(refs if refs is not None else _default_checked_refs(kinds))
checked: list[str] = []
unknown: list[str] = []
queue = list(seeds)
while queue:
ref = queue.pop(0)
if ref in checked:
continue
checked.append(ref)
if ref not in kinds:
unknown.append(ref)
continue
for dep in _dependencies_for_ref(
ref,
objects,
kinds,
event_predicate_dependencies=event_predicate_dependencies,
):
if dep not in checked and dep not in queue:
queue.append(dep)
return tuple(checked), tuple(unknown)
def _datasource_refs_for_checked_refs(
refs: Iterable[str],
objects: Mapping[str, object],
kinds: Mapping[str, SemanticKind],
) -> tuple[str, ...]:
datasources: list[str] = []
for ref in refs:
if kinds.get(ref) != SemanticKind.ENTITY:
continue
datasource = getattr(objects.get(ref), "datasource", None)
if isinstance(datasource, str):
datasources.append(datasource)
return _dedupe(datasources)
def _dataset_refs(refs: Iterable[str], kinds: Mapping[str, SemanticKind]) -> tuple[str, ...]:
return tuple(_display_path(ref) for ref in refs if kinds.get(ref) == SemanticKind.ENTITY)
def _refs_with_issue(issues: Iterable[ReadinessIssue]) -> set[str]:
return {ref for issue in issues for ref in issue.refs}
def _undeclared_naive_time_axis_issues(
checked_refs: Iterable[str],
kinds: Mapping[str, SemanticKind],
objects: Mapping[str, object],
) -> tuple[list[ReadinessIssue], list[ReadinessIssue]]:
"""Return blockers/warnings for native temporal axes without a source timezone.
An explicit ``parse=ms.datetime()``/``parse=ms.timestamp()`` with no timezone is
certainly a naive datetime/timestamp, so it blocks. Runtime-observed physical
type risks belong to preview and never affect semantic-static readiness.
"""
blockers: list[ReadinessIssue] = []
warnings: list[ReadinessIssue] = []
for ref in checked_refs:
if kinds.get(ref) != SemanticKind.TIME_DIMENSION:
continue
path = _display_path(ref)
time_dimension = objects.get(ref)
parse = getattr(time_dimension, "parse", None)
data_type = getattr(parse, "kind", None)
declared_timezone = getattr(parse, "timezone", None)
entity_ref = getattr(time_dimension, "entity", None)
entity = (
objects.get(_exact_key(entity_ref, SemanticKind.ENTITY))
if isinstance(entity_ref, str)
else None
)
datasource = getattr(entity, "datasource", None)
if data_type in {"datetime", "timestamp"} and declared_timezone is None:
parse_call = f'ms.{data_type}(timezone="Region/City")'
blockers.append(
_issue(
"undeclared_naive_time_axis",
"blocker",
(path,),
f"{path} is a native {data_type} time axis with no declared source timezone; "
"analysis will otherwise interpret naive values using the datasource read timezone.",
repair(
kind="reauthor",
canonical_id="time_dimension_column",
action=f"Declare the source timezone on this time dimension with parse={parse_call}.",
),
details={
"data_type": data_type,
"declared_timezone": None,
"datasource": datasource,
"datasource_read_timezone": "resolved at runtime",
"report_timezone": "resolved by the analysis session",
"window_alignment_risk": "Report-local windows may shift at day or hour boundaries.",
},
)
)
return blockers, warnings
def _temporal_contract_unobservable_issue(
*,
path: str,
temporal_contract: MetricTemporalContract,
root_entity: str,
registry: Registry,
) -> ReadinessIssue | None:
"""Return one blocker when an effective temporal contract cannot be observed."""
from marivo.semantic.ir import SnapshotVersioningIR
time_fold = temporal_contract.fold
status_time_dimension = temporal_contract.status_time_dimension
dim_ir = registry.dimensions.get(status_time_dimension)
if dim_ir is None:
return None
parse = getattr(dim_ir, "parse", None)
sample_interval = getattr(parse, "sample_interval", None)
if sample_interval is not None:
return None
fold_kind = time_fold.kind
partition_field: str | None = None
entity_ir = registry.entities.get(root_entity)
if entity_ir is not None:
versioning = getattr(entity_ir, "versioning", None)
if isinstance(versioning, SnapshotVersioningIR):
partition_field = versioning.partition_field
if fold_kind in {"first", "last"} and partition_field == status_time_dimension:
return None
reason = (
"snapshot_versioning_missing"
if fold_kind in {"first", "last"}
else "unsampled_non_selection_fold"
)
return _issue(
"snapshot_fold_unobservable",
"blocker",
(path,),
(
f"{path} has no legal observe path: time_fold={fold_kind} over "
f"{status_time_dimension} requires a sample_interval on the status "
"time dimension or snapshot versioning bound to it; neither is declared."
),
repair(
kind="reauthor",
canonical_id="objects.metric",
action=(
"Declare sample_interval=(1, 'hour') (or another minute/hour interval) "
"on the status time dimension via parse=ms.timestamp(...), or bind "
"snapshot versioning with versioning=ms.snapshot(partition_field="
"<status_time_dimension>, grain='day') on the root entity for "
"first/last folds."
),
),
details={
"time_fold": fold_kind,
"status_time_dimension": status_time_dimension,
"root_entity": root_entity,
"snapshot_partition_field": partition_field,
"reason": reason,
},
)
def _snapshot_fold_unobservable_issues(
checked_refs: Iterable[str],
kinds: Mapping[str, SemanticKind],
objects: Mapping[str, object],
registry: Registry | None,
) -> list[ReadinessIssue]:
"""Return blockers for metrics whose time fold has no legal observe path.
A semi-additive metric with a time_fold is observable only when either
(a) its status time dimension declares a ``sample_interval`` (sampled fold),
or (b) the fold is first/last and the root entity binds snapshot versioning
to the same status time dimension (snapshot selection). Any other
combination deadlocks the observe planner (reported as a single
``snapshot-fold-deadlock`` error) and must be surfaced at readiness time
rather than reported analysis_ready.
"""
from marivo.semantic._metric_resolution import resolve_metric_temporal_contract
from marivo.semantic.ir import MetricIR
blockers: list[ReadinessIssue] = []
if registry is None:
return blockers
for ref in checked_refs:
if kinds.get(ref) != SemanticKind.METRIC:
continue
metric = objects.get(ref)
if not isinstance(metric, MetricIR):
continue
temporal_contract = resolve_metric_temporal_contract(metric, registry)
if temporal_contract is None:
continue
root_entity = getattr(metric, "root_entity", None)
if not isinstance(root_entity, str) or not root_entity:
entities = tuple(getattr(metric, "entities", ()))
if len(entities) == 1:
root_entity = entities[0]
else:
continue
path = _display_path(ref)
issue = _temporal_contract_unobservable_issue(
path=path,
temporal_contract=temporal_contract,
root_entity=root_entity,
registry=registry,
)
if issue is not None:
blockers.append(issue)
return blockers
def build_readiness_report(
project: SemanticProject,
*,
refs: Iterable[Ref[SemanticKindTag] | str] | None = None,
) -> ReadinessReport:
"""Build a semantic-static readiness report from current loaded state.
Performs pure in-memory checks: load errors, unknown refs,
cross-datasource unfederated metrics, raw SQL requirements,
temporal artifact integrity, and load warnings. It never reads ordinary
discovery or preview history and never executes a datasource query.
Args:
project: A loaded SemanticProject instance.
refs: Semantic refs to scope the check. None checks all loaded objects.
Returns:
ReadinessReport indicating whether the selected refs satisfy the
current certification contract.
"""
blockers: list[ReadinessIssue] = []
warnings: list[ReadinessIssue] = []
if not project.is_ready():
for error in project.errors():
blockers.append(
_issue(
"load_error",
"blocker",
error.semantic_refs,
error.message,
repair(
kind="reload",
canonical_id="load",
action=error.hint or "Fix semantic load errors and reload the project.",
),
)
)
return ReadinessReport(
status="blocked",
analysis_ready_inputs=(),
blockers=tuple(blockers),
warnings=(),
input_summary=ReadinessInputSummary(
datasources=(),
refs=(),
tables=(),
),
checked_at=_checked_at(),
catalog_definition_fingerprint=None,
)
compiled_state = project._compiled_state
if compiled_state is None:
raise RuntimeError("ready semantic project has no compiled state")
catalog_definition_fingerprint = compiled_state.definition_fingerprint
event_predicate_dependencies = {
ref.key: tuple(binding.to_ref().key for binding in body.bindings)
for ref, body in compiled_state.sidecar.bodies.items()
if ref.kind is SemanticKind.EVENT
}
kinds, objects = _object_maps(project)
scoped_keys = _scope_keys(refs, kinds)
direct_refs = _dedupe(scoped_keys if scoped_keys is not None else _default_checked_refs(kinds))
checked_refs, unknown_refs = _expand_checked_refs(
scoped_keys,
kinds,
objects,
event_predicate_dependencies=event_predicate_dependencies,
)
scoped_datasources = _datasource_refs_for_checked_refs(checked_refs, objects, kinds)
reg = project._registry
cross_datasource_refs: list[str] = []
if reg is not None:
from marivo.semantic.preview_scope import dependency_entities_for_ref
for ref in direct_refs:
if kinds.get(ref) not in _EXECUTABLE_KINDS:
continue
entity_ids = dependency_entities_for_ref(_display_path(ref), registry=reg)
datasource_ids = {
reg.entities[entity_id].datasource
for entity_id in entity_ids
if entity_id in reg.entities
}
if len(entity_ids) > 1 and len(datasource_ids) > 1:
cross_datasource_refs.append(ref)
if reg is not None:
from marivo.semantic.metric_graph_canonical import MetricGraphContractError
from marivo.semantic.metric_graph_lowering import (
lower_catalog_metric,
)
for ref in direct_refs:
if kinds.get(ref) != SemanticKind.METRIC:
continue
path = _display_path(ref)
try:
lower_catalog_metric(reg, path, sidecar=project._expression_sidecar)
except (MetricGraphContractError, TypeError, ValueError) as exc:
blockers.append(
_issue(
"metric_graph_invalid",
"blocker",
(path,),
f"{path} cannot lower to the bounded metric expression graph: {exc}",
repair(
kind="reauthor",
canonical_id="objects.metric",
action=(
"Reduce the metric's recursive composition to depth at most 10 "
"and 256 pre-CSE occurrences, or repair the dependency reported "
"at the failing occurrence path."
),
),
details={
"max_depth": 10,
"max_occurrences": 256,
"lowering_error_kind": getattr(exc, "kind", "graph_contract"),
"observed_count": getattr(exc, "observed_count", None),
"limit": getattr(exc, "limit", None),
"dependency_path": getattr(exc, "path", None),
"occurrence_path": getattr(exc, "path", None),
},
)
)
for ref in unknown_refs:
path = _display_path(ref)
blockers.append(
_issue(
"unknown_ref",
"blocker",
(path,),
f"Requested semantic ref {path!r} is not loaded in the project registry.",
repair(
kind="inspect",
canonical_id="load",
action="Browse loaded refs with catalog.domains.show() or catalog.metrics.show(), then inspect a known identity with catalog.require(ms.ref.<kind>(path)).details().show().",
),
)
)
naive_time_axis_blockers, naive_time_axis_warnings = _undeclared_naive_time_axis_issues(
checked_refs,
kinds,
objects,
)
blockers.extend(naive_time_axis_blockers)
warnings.extend(naive_time_axis_warnings)
blockers.extend(_snapshot_fold_unobservable_issues(checked_refs, kinds, objects, reg))
# Period calendars are executable semantic dependencies. Unlike ordinary
# preview output, a missing/stale certified artifact is a hard blocker
# and is checked entirely from project-local state.
if reg is not None:
from marivo._temporal import TemporalSnapshotStore, period_calendar_definition_digest
from marivo.refs import ref as ref_factory
from marivo.semantic._definition_identity import scoped_definition_fingerprint
from marivo.semantic.ir import PeriodCalendarIR
snapshot_store = TemporalSnapshotStore(project._workspace_dir)
for ref in checked_refs:
if kinds.get(ref) is not SemanticKind.PERIOD_CALENDAR:
continue
calendar = objects.get(ref)
if not isinstance(calendar, PeriodCalendarIR):
continue
calendar_ref = ref_factory.period_calendar(_display_path(ref))
definition_digest = period_calendar_definition_digest(
calendar_ref=calendar_ref,
boundary_timezone=calendar.boundary_timezone,
coverage=calendar.coverage,
levels=calendar.levels,
correspondences=calendar.correspondences,
dependency_digest=scoped_definition_fingerprint(
root=calendar_ref,
definitions=compiled_state.definitions,
dependencies=compiled_state.dependencies,
sidecar=compiled_state.sidecar,
),
)
status, _snapshot = snapshot_store.inspect_current(
calendar_ref,
definition_digest=definition_digest,
)
if status != "current":
path = _display_path(ref)
issue_kind = cast(
"ReadinessIssueKind",
f"period_calendar_artifact_{status}",
)
if status == "missing":
issue_kind = "period_calendar_artifact_missing"
blockers.append(
_issue(
issue_kind,
"blocker",
(path,),
(
f"{path} has no current certified period-calendar artifact for its "
f"declaration (state={status})."
),
repair(
kind="rescope",
canonical_id="preview",
action=(
"Run catalog.preview(ref, scope=...) with one exhaustive explicit "
"scope covering the period-calendar declaration."
),
),
details={"artifact_status": status},
)
)
from marivo._temporal import WorkScheduleSnapshotStore, work_schedule_definition_digest
from marivo.semantic.ir import WorkScheduleIR
work_schedule_store = WorkScheduleSnapshotStore(project._workspace_dir)
for ref in checked_refs:
if kinds.get(ref) is not SemanticKind.WORK_SCHEDULE:
continue
work_schedule = objects.get(ref)
if not isinstance(work_schedule, WorkScheduleIR):
continue
work_schedule_ref = ref_factory.work_schedule(_display_path(ref))
definition_digest = work_schedule_definition_digest(
work_schedule_ref=work_schedule_ref,
boundary_timezone=work_schedule.boundary_timezone,
coverage=work_schedule.coverage,
date=work_schedule.date,
is_working=work_schedule.is_working,
dependency_digest=scoped_definition_fingerprint(
root=work_schedule_ref,
definitions=compiled_state.definitions,
dependencies=compiled_state.dependencies,
sidecar=compiled_state.sidecar,
),
)
status, _schedule_snapshot = work_schedule_store.inspect_current(
work_schedule_ref,
definition_digest=definition_digest,
)
if status != "current":
path = _display_path(ref)
issue_kind = cast("ReadinessIssueKind", f"work_schedule_artifact_{status}")
if status == "missing":
issue_kind = "work_schedule_artifact_missing"
blockers.append(
_issue(
issue_kind,
"blocker",
(path,),
f"{path} has no current certified work-schedule artifact for its declaration (state={status}).",
repair(
kind="rescope",
canonical_id="preview",
action=(
"Run catalog.preview(ref, scope=...) with one exhaustive explicit "
"scope covering the work-schedule declaration."
),
),
details={"artifact_status": status},
)
)
from marivo._temporal import TemporalSetSnapshotStore, temporal_set_definition_digest
from marivo.semantic.ir import TemporalSetIR
temporal_set_store = TemporalSetSnapshotStore(project._workspace_dir)
for ref in checked_refs:
if kinds.get(ref) is not SemanticKind.TEMPORAL_SET:
continue
temporal_set = objects.get(ref)
if not isinstance(temporal_set, TemporalSetIR):
continue
temporal_set_ref = ref_factory.temporal_set(_display_path(ref))
definition_digest = temporal_set_definition_digest(
temporal_set_ref=temporal_set_ref,
boundary_timezone=temporal_set.boundary_timezone,
coverage=temporal_set.coverage,
occurrence_id=temporal_set.occurrence_id,
start=temporal_set.start,
end=temporal_set.end,
category=temporal_set.category,
dependency_digest=scoped_definition_fingerprint(
root=temporal_set_ref,
definitions=compiled_state.definitions,
dependencies=compiled_state.dependencies,
sidecar=compiled_state.sidecar,
),
)
status, _temporal_snapshot = temporal_set_store.inspect_current(
temporal_set_ref,
definition_digest=definition_digest,
)
if status != "current":
path = _display_path(ref)
issue_kind = cast("ReadinessIssueKind", f"temporal_set_artifact_{status}")
if status == "missing":
issue_kind = "temporal_set_artifact_missing"
blockers.append(
_issue(
issue_kind,
"blocker",
(path,),
f"{path} has no current certified temporal-set artifact for its declaration (state={status}).",
repair(
kind="rescope",
canonical_id="preview",
action=(
"Run catalog.preview(ref, scope=...) with one exhaustive explicit "
"scope covering the temporal-set declaration."
),
),
details={"artifact_status": status},
)
)
for ref in direct_refs:
if kinds.get(ref) is not SemanticKind.STATE_MODEL:
continue
model = objects.get(ref)
if model is not None and not getattr(model, "inceptions", ()):
path = _display_path(ref)
blockers.append(
_issue(
"state_model_seed_missing",
"blocker",
(path,),
f"{path} has no inception trigger and is not ready for Phase 3 replay.",
repair(
kind="reauthor",
canonical_id="inception",
action="Add at least one ms.inception(on=...) trigger before replay.",
),
)
)
# Cross-datasource unfederated metrics.
if reg is not None:
for ref in cross_datasource_refs:
path = _display_path(ref)
blockers.append(
_issue(
"cross_datasource_unfederated",
"blocker",
(path,),
f"Semantic object {path} spans multiple datasources without federation support.",
repair(
kind="reauthor",
canonical_id="objects.metric",
action="Move integration upstream, enable a federated backend, or split the metric.",
),
)
)
# SQL parity unverified warnings.
for ref in checked_refs:
if kinds.get(ref) != SemanticKind.METRIC:
continue
path = _display_path(ref)
obj = objects.get(ref)
if obj is None:
continue
prov = getattr(obj, "provenance", None)
if prov is None:
continue
provenance_sql = prov.sql
if provenance_sql is None:
continue
if not _parity_passed(project, path):
warnings.append(
_issue(
"sql_parity_unverified",
"warning",
(path,),
f"{path} has provenance SQL but parity has not been confirmed.",
repair(
kind="retry",
canonical_id="parity_check",
action=f"Run ms.parity_check({path!r}) when parity matters, or report the warning as non-blocking when the certification policy allows it.",
),
)
)
# Forward load warnings as readiness warnings.
for sw in project.warnings():
if sw.kind in {"string_ref", "potentially_fragile_reference"}:
warnings.append(
_issue(
"fragile_string_ref",
"warning",
sw.refs,
sw.message,
repair(
kind="reauthor",
canonical_id="ref",
action="Replace fragile string refs with stable object refs where possible.",
),
)
)
if sw.kind == "time_dimension_pushdown_advisory":
warnings.append(
_issue(
"time_dimension_pushdown_advisory",
"warning",
sw.refs,
sw.message,
repair(
kind="reauthor",
canonical_id="time_dimension_column",
action="If the business axis matches the partition field, keep the raw string/integer column and declare date_format; keep the expression when business semantics require it.",
),
)
)
blocked_refs = _refs_with_issue(blockers)
analysis_ready_ids = tuple(
ref
for ref in direct_refs
if blocked_refs.isdisjoint(
_display_path(dependency)
for dependency in _expand_checked_refs(
(ref,),
kinds,
objects,
event_predicate_dependencies=event_predicate_dependencies,
)[0]
)
)
analysis_ready_inputs = tuple(
_exact_ref(_display_path(ref), kinds[ref]) for ref in analysis_ready_ids if ref in kinds
)
datasources_checked: tuple[str, ...] = scoped_datasources if reg is not None else ()
blockers = [
replace(issue, catalog_definition_fingerprint=catalog_definition_fingerprint)
for issue in blockers
]
warnings = [
replace(issue, catalog_definition_fingerprint=catalog_definition_fingerprint)
for issue in warnings
]
return ReadinessReport(
status=_status(blockers, warnings),
analysis_ready_inputs=analysis_ready_inputs,
blockers=tuple(blockers),
warnings=tuple(warnings),
input_summary=ReadinessInputSummary(
datasources=datasources_checked,
refs=_dedupe(_display_path(ref) for ref in checked_refs),
tables=_dataset_refs(checked_refs, kinds),
),
checked_at=_checked_at(),
catalog_definition_fingerprint=catalog_definition_fingerprint,
)