"""Base typed analysis frame contracts."""
from __future__ import annotations
# mypy: disable-error-code=import-untyped
import importlib
from collections.abc import Iterator
from dataclasses import dataclass, field
from datetime import date, datetime, time
from decimal import Decimal
from pathlib import PurePath
from typing import TYPE_CHECKING, Any, Literal
import pandas as pd
from pydantic import BaseModel, ConfigDict, Field, computed_field, model_validator
from marivo._temporal import ComparisonTemporalContractV1, FrameTemporalContractV1
from marivo.analysis._contract_budget import ARTIFACT_CONTRACT_RENDER_BUDGET
from marivo.analysis.attribution_contract import (
AttributeAdmissionV1,
CumulativeAttributionCapabilityV1,
)
from marivo.analysis.candidate_lineage import CandidateOrigin, merge_candidate_origins
from marivo.analysis.errors import (
AnalysisRepair,
FrameMutationError,
SemanticKindMismatchError,
)
from marivo.analysis.evidence.types import (
ArtifactDigest,
ArtifactIssue,
DataQualityIssue,
EvidenceScope,
QualitySummary,
)
from marivo.analysis.lineage import Lineage
from marivo.introspection.live.model import LiveHelpTarget
from marivo.refs import SemanticKind
from marivo.render import Card, RenderableResult, result_repr
from marivo.semantic._capabilities.catalog_members import CATALOG_MEMBER_CONTRACTS
if TYPE_CHECKING:
from marivo.analysis.evidence.artifact_reads import Finding, FindingPage
CURRENT_ARTIFACT_SCHEMA_VERSION: Literal["analysis-artifact/v13"] = "analysis-artifact/v13"
_ARTIFACT_SEMANTIC_INPUT_LIMIT = 12
_DEFAULT_FRAME_PREVIEW_ROWS = 50
def _display_column_names(columns: pd.Index) -> list[str]:
display_columns: list[str] = []
used_columns: set[str] = set()
for column in columns:
column_name = str(column)
display_name = column_name
suffix = 2
while display_name in used_columns:
display_name = f"{column_name}#{suffix}"
suffix += 1
used_columns.add(display_name)
display_columns.append(display_name)
return display_columns
def _is_decimal_column(series: pd.Series) -> bool:
"""Return True for object-dtype columns whose non-null values are all Decimal."""
if not pd.api.types.is_object_dtype(series.dtype):
return False
non_null = series.dropna()
return bool(len(non_null)) and all(isinstance(value, Decimal) for value in non_null)
def _coerce_decimal_columns(df: pd.DataFrame) -> pd.DataFrame:
"""Cast object columns whose values are ``decimal.Decimal`` to ``float64``.
Some backends (e.g. DuckDB) materialize DECIMAL columns as object-dtype
``decimal.Decimal`` values. Once a frame exits the typed boundary, terminal
pandas arithmetic over such columns (``float - Decimal``) raises
``TypeError``, and Decimal precision is meaningless downstream. Coerce those
columns to ``float64`` while leaving genuine string, integer, float, and
temporal columns untouched.
"""
for column in df.columns:
if _is_decimal_column(df[column]):
df[column] = df[column].astype("float64")
return df
def _is_missing(value: Any) -> bool:
try:
missing = pd.isna(value)
except (TypeError, ValueError):
return False
if isinstance(missing, bool):
return missing
item = getattr(missing, "item", None)
if callable(item):
try:
scalar = item()
except (TypeError, ValueError):
return False
return scalar if isinstance(scalar, bool) else False
return False
def _preview_cell(value: Any) -> Any:
if _is_missing(value):
return None
if isinstance(value, pd.Timestamp):
return value.isoformat()
if isinstance(value, (datetime, date, time)):
return value.isoformat()
if isinstance(value, pd.Timedelta):
return str(value)
item = getattr(value, "item", None)
if callable(item):
try:
return item()
except (TypeError, ValueError):
return value
return value
def assert_semantic_shape(*, got: str, expected: str, frame_kind: str) -> None:
"""Raise SemanticKindMismatchError unless ``got`` semantic shape matches ``expected``."""
if got != expected:
raise SemanticKindMismatchError(
message=f"{frame_kind} semantic_shape is {got!r}, expected {expected!r}",
context={
"got_semantic_shape": got,
"expected_semantic_shape": expected,
"frame_kind": frame_kind,
},
)
def assert_attribution_shape(*, got: str, expected: str, frame_kind: str) -> None:
"""Raise SemanticKindMismatchError unless ``got`` attribution shape matches ``expected``."""
if got != expected:
raise SemanticKindMismatchError(
message=f"{frame_kind} attribution_shape is {got!r}, expected {expected!r}",
context={
"got_attribution_shape": got,
"expected_attribution_shape": expected,
"frame_kind": frame_kind,
},
)
ArtifactColumnRole = Literal["time", "dimension", "value", "measure", "unknown"]
ArtifactMaterialization = Literal["materialized", "recomputed", "partial"]
ArtifactPreconditionStatus = Literal["pass", "fail"]
[docs]
class ArtifactColumn(BaseModel):
"""Column-level schema fact for an analysis artifact."""
model_config = ConfigDict(extra="forbid", frozen=True)
name: str
dtype: str
nullable: bool
role: ArtifactColumnRole = "unknown"
[docs]
class ArtifactSchema(BaseModel):
"""Bounded deterministic schema descriptor embedded in an artifact contract."""
model_config = ConfigDict(extra="forbid", frozen=True)
columns: list[ArtifactColumn]
semantic_shape: str | None = None
class ArtifactSemanticInput(BaseModel):
"""One exact semantic input retained by an artifact with its acquisition path."""
model_config = ConfigDict(extra="forbid", frozen=True)
role: str
semantic_kind: SemanticKind
semantic_path: str
output_column: str | None = None
acquisition: str
help_target: LiveHelpTarget
@model_validator(mode="after")
def _validate_help_target(self) -> ArtifactSemanticInput:
if self.help_target.canonical_id is None:
raise ValueError("semantic input help_target must be exact")
return self
[docs]
class ArtifactPrecondition(BaseModel):
"""Mechanical precondition attached to an affordance."""
model_config = ConfigDict(extra="forbid", frozen=True)
check: str
status: ArtifactPreconditionStatus
reason: str | None = None
repair: AnalysisRepair | None = None
repair_options: tuple[AnalysisRepair, ...] = ()
@model_validator(mode="after")
def _validate_repairs(self) -> ArtifactPrecondition:
has_repair = self.repair is not None
has_options = bool(self.repair_options)
if has_repair and has_options:
raise ValueError("repair and repair_options are mutually exclusive")
if self.status == "fail" and not (has_repair or has_options):
raise ValueError("failed precondition requires repair or repair_options")
repairs = (self.repair,) if self.repair is not None else self.repair_options
for item in repairs:
if not item.action.strip():
raise ValueError("precondition repair action must be non-empty")
if item.kind == "retry" and not (item.snippet and item.snippet.strip()):
raise ValueError("retry precondition repair requires a runnable snippet")
return self
class ArtifactCallOption(BaseModel):
"""One exact runnable call option owned by an artifact affordance."""
model_config = ConfigDict(extra="forbid", frozen=True)
label: str
semantic_refs: tuple[str, ...] = ()
snippet: str
[docs]
class ArtifactAffordance(BaseModel):
"""Mechanical compatibility entry, not a recommendation."""
model_config = ConfigDict(extra="forbid", frozen=True)
capability_id: str
public_entrypoint: str
help_target: LiveHelpTarget
input_requirements: tuple[ArtifactInputRequirement, ...] = ()
preconditions: tuple[ArtifactPrecondition, ...] = ()
expected_output_family: str | None = None
call_options: tuple[ArtifactCallOption, ...] = ()
@model_validator(mode="after")
def _validate_help_target(self) -> ArtifactAffordance:
if self.help_target.canonical_id is None:
raise ValueError("affordance help_target must be exact")
return self
class ArtifactBoundaryPort(BaseModel):
"""Terminal exit boundary port derived from the capability registry."""
model_config = ConfigDict(extra="forbid", frozen=True)
kind: Literal["terminal_exit"]
capability_id: Literal["boundary.to_pandas"]
public_entrypoint: str
help_target: LiveHelpTarget
preserves: tuple[str, ...]
does_not_preserve: tuple[str, ...]
@model_validator(mode="after")
def _validate_help_target(self) -> ArtifactBoundaryPort:
expected = LiveHelpTarget(surface="analysis", canonical_id="boundary.to_pandas")
if self.help_target != expected:
raise ValueError("boundary port must target analysis.boundary.to_pandas")
return self
[docs]
class ArtifactContract(BaseModel):
"""Mechanical consumption contract for an artifact."""
model_config = ConfigDict(extra="forbid", frozen=True)
kind: str
ref: str
is_canonical: bool
artifact_schema: ArtifactSchema
semantic_inputs: tuple[ArtifactSemanticInput, ...] = ()
semantic_inputs_omitted: int = Field(default=0, ge=0)
issues: tuple[ArtifactIssue, ...] = ()
affordances: tuple[ArtifactAffordance, ...] = ()
boundary_ports: tuple[ArtifactBoundaryPort, ...] = ()
attribute_admission: AttributeAdmissionV1 | None = None
cumulative_attribution: CumulativeAttributionCapabilityV1 | None = None
row_arithmetic: (
Literal[
"not_additive_across_resolutions",
"additive_once_per_comparison_bucket",
]
| None
) = None
temporal_contract: FrameTemporalContractV1 | ComparisonTemporalContractV1 | None = None
@model_validator(mode="after")
def _validate_affordance_budget(self) -> ArtifactContract:
if len(self.affordances) > ARTIFACT_CONTRACT_RENDER_BUDGET.max_affordances:
raise ValueError(
"ArtifactContract exceeds the complete affordance budget: "
f"{len(self.affordances)} > "
f"{ARTIFACT_CONTRACT_RENDER_BUDGET.max_affordances}"
)
return self
@computed_field # type: ignore[prop-decorator] # Pydantic wraps this property.
@property
def output_columns(self) -> tuple[str, ...]:
"""Return the exact ordered public names from the artifact schema."""
return tuple(column.name for column in self.artifact_schema.columns)
def _repr_identity(self) -> str:
return (
f"ArtifactContract kind={self.kind} ref={self.ref} affordances={len(self.affordances)}"
)
[docs]
def render(self, *, max_output_bytes: int | None = None) -> str:
"""Render the complete bounded contract without dropping continuations."""
rendered = _artifact_contract_card(self).render(max_output_bytes=None)
_enforce_artifact_contract_budget(self, rendered)
if max_output_bytes is not None and len(rendered.encode("utf-8")) > max_output_bytes:
required = len(rendered.encode("utf-8"))
raise ValueError(
"max_output_bytes is too small for the complete ArtifactContract; "
f"required={required}, received={max_output_bytes}; "
"ArtifactContract rendering never truncates affordances or repairs"
)
return rendered
[docs]
def show(self, *, max_output_bytes: int | None = None) -> None:
"""Print the complete bounded mechanical contract."""
print(self.render(max_output_bytes=max_output_bytes))
def __repr__(self) -> str:
return result_repr(self._repr_identity())
def __str__(self) -> str:
return self.render()
def _enforce_artifact_contract_budget(contract: ArtifactContract, rendered: str) -> None:
"""Fail explicitly when a complete contract exceeds its native limits."""
budget = ARTIFACT_CONTRACT_RENDER_BUDGET
violations: list[str] = []
line_count = len(rendered.splitlines())
if line_count > budget.max_lines:
violations.append(f"lines={line_count}>{budget.max_lines}")
if len(rendered) > budget.max_codepoints:
violations.append(f"codepoints={len(rendered)}>{budget.max_codepoints}")
if len(contract.affordances) > budget.max_affordances:
violations.append(f"affordances={len(contract.affordances)}>{budget.max_affordances}")
if violations:
raise RuntimeError("ArtifactContract render budget exceeded: " + ", ".join(violations))
[docs]
class ArtifactState(BaseModel):
"""Baseline runtime facts for a materialized artifact."""
model_config = ConfigDict(extra="forbid", frozen=True)
materialization: ArtifactMaterialization
content_hash: str | None = None
@dataclass(frozen=True, slots=True)
class _FrameAuxiliaryTable:
"""One private frame-owned table written beside the public artifact rows."""
filename: str
dataframe: pd.DataFrame
def __post_init__(self) -> None:
path = PurePath(self.filename)
if (
not self.filename
or path.is_absolute()
or len(path.parts) != 1
or path.name != self.filename
):
raise ValueError("auxiliary table filename must be one safe basename")
@dataclass(frozen=True, slots=True)
class _FrameAuxiliaryReceipt:
"""Persistence receipt bound into shape-specific frame metadata."""
filename: str
row_count: int
byte_size: int
content_hash: str
@dataclass(frozen=True, slots=True)
class _ArtifactSemanticBinding:
"""Private role-preserving semantic input used to build public guidance."""
role: str
semantic_kind: SemanticKind
semantic_path: str
output_column: str | None = None
_CATALOG_COLLECTION_BY_KIND: dict[SemanticKind, str] = {
member.kind: member.property_name for member in CATALOG_MEMBER_CONTRACTS
}
def _artifact_semantic_input(binding: _ArtifactSemanticBinding) -> ArtifactSemanticInput:
collection = _CATALOG_COLLECTION_BY_KIND[binding.semantic_kind]
return ArtifactSemanticInput(
role=binding.role,
semantic_kind=binding.semantic_kind,
semantic_path=binding.semantic_path,
output_column=binding.output_column,
acquisition=(f'session.catalog.{collection}.get("{binding.semantic_path}")'),
help_target=LiveHelpTarget(
surface="analysis",
canonical_id=f"catalog.{collection}",
),
)
def _visible_precondition(precondition: ArtifactPrecondition) -> bool:
"""Return True when a precondition is visible (has actionable content).
A passing precondition is visible only when it carries a non-empty reason.
A failing precondition is visible only when it carries a repair with a
non-empty action.
"""
if precondition.status == "pass":
return bool(precondition.reason and precondition.reason.strip())
if precondition.repair is not None:
return _visible_repair(precondition.repair)
return bool(precondition.repair_options) and all(
_visible_repair(repair) for repair in precondition.repair_options
)
def _visible_repair(repair: AnalysisRepair) -> bool:
if not repair.action.strip():
return False
return repair.kind != "retry" or bool(repair.snippet and repair.snippet.strip())
def _affordance_visible(affordance: ArtifactAffordance) -> bool:
"""Suppress only failed preconditions that lack executable or inspectable repair."""
return not any(
precondition.status == "fail" and not _visible_precondition(precondition)
for precondition in affordance.preconditions
)
def _capability_public_entrypoint(capability_id: str) -> str:
"""Return the registry-owned public entrypoint for one capability."""
registry_module = importlib.import_module("marivo.analysis._capabilities.registry")
return str(registry_module.REGISTRY.by_id(capability_id).public_entrypoint)
def _read_artifact_affordance(
capability_id: str,
*,
artifact_shape: str | None = None,
) -> ArtifactAffordance:
"""Build one receiver-owned continuation solely from its exact descriptor."""
registry_module = importlib.import_module("marivo.analysis._capabilities.registry")
model_module = importlib.import_module("marivo.analysis._capabilities.model")
descriptor = registry_module.REGISTRY.by_id(capability_id)
if not isinstance(descriptor, model_module.ReadCapability):
raise TypeError(f"{capability_id} is not a receiver-owned Read capability")
output_family = descriptor.output_type
if descriptor.artifact_output_by_shape:
if artifact_shape is None or artifact_shape not in descriptor.artifact_output_by_shape:
raise ValueError(f"{capability_id} has no output family for shape {artifact_shape!r}")
output_family = descriptor.artifact_output_by_shape[artifact_shape]
return ArtifactAffordance(
capability_id=descriptor.id,
public_entrypoint=descriptor.public_entrypoint,
help_target=LiveHelpTarget(
surface="analysis",
canonical_id=descriptor.help_target,
),
input_requirements=(
ArtifactInputRequirement(
parameter="receiver",
accepted_families=(descriptor.receiver_family,),
bindable_from_current_artifact=True,
),
),
expected_output_family=output_family,
)
def _help_call(target: LiveHelpTarget) -> str:
"""Render one exact typed Help target as a directly callable public path."""
if target.canonical_id is None:
raise ValueError("help target must be exact")
return f'marivo.help("{target.surface}.{target.canonical_id}")'
def _repair_help_call(repair: AnalysisRepair) -> str:
"""Render a structured repair help target as a public marivo.help call."""
return _help_call(repair.help_target)
def _grouped_affordances(
affordances: tuple[ArtifactAffordance, ...],
) -> tuple[tuple[LiveHelpTarget, tuple[ArtifactAffordance, ...]], ...]:
"""Group readable affordances by registry ownership without mutating them."""
registry_module = importlib.import_module("marivo.analysis._capabilities.registry")
registry = registry_module.REGISTRY
grouped: dict[LiveHelpTarget, list[ArtifactAffordance]] = {}
for affordance in affordances:
owner = registry.continuation_group(affordance.help_target)
grouped.setdefault(owner, []).append(affordance)
return tuple(
(
owner,
tuple(sorted(items, key=lambda item: item.public_entrypoint)),
)
for owner, items in sorted(grouped.items(), key=lambda item: item[0].display)
)
def _contract_receiver(contract: ArtifactContract) -> str:
return "candidates" if contract.kind == "candidate_set" else "frame"
def _artifact_contract_card(contract: ArtifactContract) -> Card:
"""Build the bounded contract card from already-materialized typed facts."""
semantic_shape = contract.artifact_schema.semantic_shape or "unspecified"
card = (
Card(
identity=contract._repr_identity(),
available=(
".artifact_schema",
".output_columns",
".semantic_inputs",
".semantic_inputs_omitted",
".issues",
".affordances",
".boundary_ports",
".model_dump()",
".show()",
),
)
.field("canonical_state", "canonical" if contract.is_canonical else "non_canonical")
.field("receiver", _contract_receiver(contract))
.field("semantic_shape", semantic_shape)
.field("output_columns", repr(list(contract.output_columns)))
.listing(
"columns",
(
f"{column.name}: dtype={column.dtype} nullable={str(column.nullable).lower()} "
f"role={column.role}"
for column in contract.artifact_schema.columns
),
)
)
if contract.semantic_inputs:
card.listing(
"semantic inputs",
(
f"{item.role}: {item.semantic_kind.value}:{item.semantic_path}"
+ (f" output_column={item.output_column}" if item.output_column is not None else "")
+ f"; acquire={item.acquisition}; help={_help_call(item.help_target)}"
for item in contract.semantic_inputs
),
)
if contract.semantic_inputs_omitted:
card.field("semantic_inputs_omitted", str(contract.semantic_inputs_omitted))
if contract.issues:
from marivo.analysis.evidence.summary import render_artifact_issue
card.listing(
"issues/blockers",
(
f"{issue.severity} {issue.kind}: {render_artifact_issue(issue)}"
for issue in contract.issues
),
)
if contract.attribute_admission is not None:
admission = contract.attribute_admission
admission_text = (
f"status={admission.status} attribution_shape={admission.attribution_shape}"
)
if admission.status == "blocked":
admission_text += f" blocker={admission.blocker}"
else:
admission_text += " multiple_axes=" + "|".join(admission.mode.multiple_axes)
admission_text += f" multiple_axes_default={admission.mode.multiple_axes_default}"
card.field(f"{_capability_public_entrypoint('attribute')} admission", admission_text)
if contract.cumulative_attribution is not None:
capability = contract.cumulative_attribution
for route_name, route in (
("business_axes", capability.business_axes),
("accumulation_time", capability.accumulation_time),
("mixed_axes", capability.mixed_axes),
):
route_text = (
f"supported path={route.path}"
if route.status == "supported"
else f"blocked blocker={route.blocker}"
)
card.field(
f"{_capability_public_entrypoint('attribute')} [{route_name}]",
route_text,
)
if contract.row_arithmetic is not None:
card.field("row_arithmetic", contract.row_arithmetic)
if contract.temporal_contract is not None:
card.field(
"temporal_contract",
str(contract.temporal_contract.model_dump(mode="json")),
)
affordances = tuple(
affordance for affordance in contract.affordances if _affordance_visible(affordance)
)
if not affordances:
card.field("continuations", "none")
else:
card.field("continuations", "mechanical, unranked")
for owner, grouped in _grouped_affordances(affordances):
card.listing(owner.display, (_render_affordance(item) for item in grouped))
precondition_lines = tuple(
line
for entrypoints, precondition in _grouped_preconditions(affordances)
for line in _render_precondition(entrypoints, precondition)
)
if precondition_lines:
card.listing("preconditions and repairs", precondition_lines)
card.listing(
"terminal exits",
(
f"{port.public_entrypoint}; help={_help_call(port.help_target)}; "
f"preserves={', '.join(port.preserves)}; "
f"does_not_preserve={', '.join(port.does_not_preserve)}"
for port in contract.boundary_ports
),
)
return card
def _render_affordance(affordance: ArtifactAffordance) -> str:
rendered_requirements: list[str] = []
for requirement in affordance.input_requirements:
details = [f"current_artifact={str(requirement.bindable_from_current_artifact).lower()}"]
if requirement.derivable_from_current_artifact:
details.append("derived_from_current_artifact=true")
if requirement.accepted_semantic_shapes:
details.append(f"semantic_shapes={','.join(requirement.accepted_semantic_shapes)}")
if requirement.acquisition is not None:
details.append(f"acquire={requirement.acquisition}")
details.append(
"help=" + ",".join(_help_call(target) for target in requirement.help_targets)
)
families = ",".join(requirement.accepted_families) or "constructed"
rendered_requirements.append(f"{requirement.parameter}={families} ({'; '.join(details)})")
bindings = "; ".join(rendered_requirements)
base = (
f"{affordance.public_entrypoint} -> {affordance.expected_output_family or 'none'}; "
f"inputs={bindings or 'none'}; help={_help_call(affordance.help_target)}"
)
if not affordance.call_options:
return base
options = " | ".join(f"{option.label}: {option.snippet}" for option in affordance.call_options)
return f"{base}; options={options}"
def _render_precondition(
entrypoints: tuple[str, ...],
precondition: ArtifactPrecondition,
) -> tuple[str, ...]:
reason = precondition.reason or "none"
label = f"{', '.join(entrypoints)} [{precondition.check}]"
lines = [f"{label}: status={precondition.status}; reason={reason}"]
repairs = (
(precondition.repair,) if precondition.repair is not None else precondition.repair_options
)
for index, repair in enumerate(repairs, start=1):
snippet = repair.snippet.replace("\n", "\\n") if repair.snippet is not None else "none"
option = f" option={index}" if len(repairs) > 1 else ""
lines.append(
f"{label}{option}: repair={repair.kind}; "
f"action={repair.action}; help={_repair_help_call(repair)}; snippet={snippet}"
)
return tuple(lines)
def _grouped_preconditions(
affordances: tuple[ArtifactAffordance, ...],
) -> tuple[tuple[tuple[str, ...], ArtifactPrecondition], ...]:
"""Project identical current-state gates once with every owning entrypoint."""
grouped: dict[str, tuple[list[str], ArtifactPrecondition]] = {}
order: list[str] = []
for affordance in affordances:
for precondition in affordance.preconditions:
if not _visible_precondition(precondition):
continue
key = precondition.model_dump_json()
if key not in grouped:
grouped[key] = ([], precondition)
order.append(key)
grouped[key][0].append(affordance.public_entrypoint)
return tuple((tuple(dict.fromkeys(grouped[key][0])), grouped[key][1]) for key in order)
def _output_family_str(desc: Any) -> str:
"""Return a string representation of a capability descriptor's output family."""
model_module = importlib.import_module("marivo.analysis._capabilities.model")
output_contract = getattr(desc, "output_contract", None)
if output_contract is not None:
return str(output_contract.render())
output = desc.output_family
if isinstance(output, model_module.SameAsInputFamily):
return f"same as {output.parameter}"
return str(output)
def _build_boundary_ports(registry: Any) -> list[ArtifactBoundaryPort]:
"""Build the single terminal boundary port from the registry."""
model_module = importlib.import_module("marivo.analysis._capabilities.model")
desc = registry.by_id("boundary.to_pandas")
assert isinstance(desc, model_module.BoundaryCapability)
return [
ArtifactBoundaryPort(
kind="terminal_exit",
capability_id="boundary.to_pandas",
public_entrypoint=desc.public_entrypoint,
help_target=LiveHelpTarget(
surface="analysis",
canonical_id=desc.help_target,
),
preserves=desc.preserves,
does_not_preserve=desc.does_not_preserve,
)
]
def _column_role(column_name: str) -> ArtifactColumnRole:
"""Infer a column role from name heuristics, defaulting to ``dimension``.
The ``"unknown"`` role is only reachable via direct ``ArtifactColumn``
construction or the field default; this function never returns it.
"""
normalized = column_name.lower()
if normalized in {"bucket_start", "bucket_end", "window_start", "window_end", "time"}:
return "time"
if normalized in {"value", "current_value", "baseline_value", "delta", "contribution"}:
return "value"
if normalized in {"measure", "metric"}:
return "measure"
return "dimension"
[docs]
@dataclass(repr=False)
class BaseFrame(RenderableResult):
"""Call marivo.help(BaseFrame) for its public consumption contract."""
_df: pd.DataFrame
meta: BaseFrameMeta
_auxiliary_frames: dict[str, pd.DataFrame] = field(default_factory=dict, repr=False)
_NEXT_INTENTS: tuple[str, ...] = ()
_AVAILABLE_ENTRIES: tuple[str, ...] = (
".contract().show()",
".findings(...)",
".to_pandas()",
)
def __post_init__(self) -> None:
self._assert_row_count_consistent()
def _assert_row_count_consistent(self) -> None:
materialized = int(self._df.shape[0])
if self.meta.row_count != materialized:
raise ValueError(
"frame row count mismatch: "
f"meta.row_count={self.meta.row_count}, materialized={materialized}"
)
@property
def ref(self) -> str:
return self.meta.ref
@property
def lineage(self) -> Lineage:
return self.meta.lineage
@property
def kind(self) -> str:
return self.meta.kind
@property
def quality_summary(self) -> QualitySummary | None:
return self.meta.quality_summary
def _evaluate_construction_quality(
self,
*,
artifact_id: str,
source_history: BaseFrame | None = None,
) -> Any:
"""Evaluate the Frame-owned pre-publication checks at the commit boundary."""
quality_module = importlib.import_module("marivo.analysis.frames._quality")
return quality_module.evaluate_frame_quality(
self,
artifact_id=artifact_id,
source_history=source_history,
)
@property
def evidence_status(self) -> Literal["complete", "partial", "unavailable"]:
return self.meta.evidence_status
@property
def evidence_digest(self) -> ArtifactDigest | None:
return self.meta.evidence_digest
@property
def finding_count(self) -> int:
"""Return the exact persisted Finding count for this Artifact.
Returns:
The non-negative count committed with this Artifact.
Example:
>>> artifact.finding_count
3
Constraints:
The count is persisted metadata, not a live ledger query.
"""
return self.meta.finding_count
[docs]
def findings(self, *, limit: int = 20, cursor: str | None = None) -> FindingPage:
"""Return one bounded page of Findings owned by this Artifact.
Args:
limit: Maximum Findings to retain, from 1 through 100.
cursor: Opaque continuation returned by the previous page.
Returns:
An immutable Artifact-scoped :class:`FindingPage`.
Example:
>>> page = artifact.findings(limit=20)
>>> page.items[0].show() if page.items else None
Constraints:
Reads the ledger identified by immutable Artifact metadata and never
attaches a mutable Session handle.
"""
from marivo.analysis.evidence.artifact_reads import findings
return findings(self, limit=limit, cursor=cursor)
[docs]
def finding(self, finding_id: str) -> Finding:
"""Return one exact Finding owned by this Artifact.
Args:
finding_id: Exact id returned by this Artifact's ``findings()`` page.
Returns:
The full immutable :class:`Finding`, including derivation.
Example:
>>> finding = artifact.finding("finding_01")
>>> finding.show()
Constraints:
A Finding owned by another Artifact is treated as not found.
"""
from marivo.analysis.evidence.artifact_reads import finding
return finding(self, finding_id)
@property
def state(self) -> ArtifactState:
return ArtifactState(
materialization="materialized",
content_hash=self.meta.content_hash,
)
def _build_schema(self) -> ArtifactSchema:
"""Build the schema descriptor embedded in the artifact contract."""
public_columns = self._public_column_names()
columns = [
ArtifactColumn(
name=name,
dtype=("float64" if _is_decimal_column(self._df.iloc[:, idx]) else str(dtype)),
nullable=bool(self._df.iloc[:, idx].isna().any()) if len(self._df) else True,
role=_column_role(str(self._df.columns[idx])),
)
for idx, (name, dtype) in enumerate(zip(public_columns, self._df.dtypes, strict=True))
]
raw_shape = getattr(self.meta, "semantic_kind", None)
return ArtifactSchema(
columns=columns,
semantic_shape=raw_shape if isinstance(raw_shape, str) else None,
)
def _semantic_input_bindings(self) -> tuple[_ArtifactSemanticBinding, ...]:
"""Return exact semantic inputs retained by this artifact family."""
return ()
def _build_semantic_inputs(self) -> tuple[ArtifactSemanticInput, ...]:
"""Build bounded acquisition guidance without reading artifact rows."""
inputs: list[ArtifactSemanticInput] = []
seen: set[tuple[str, SemanticKind, str, str | None]] = set()
for binding in self._semantic_input_bindings():
key = (
binding.role,
binding.semantic_kind,
binding.semantic_path,
binding.output_column,
)
if key in seen:
continue
seen.add(key)
inputs.append(_artifact_semantic_input(binding))
return tuple(inputs)
def _bounded_semantic_inputs(self) -> tuple[tuple[ArtifactSemanticInput, ...], int]:
semantic_inputs = self._build_semantic_inputs()
retained = semantic_inputs[:_ARTIFACT_SEMANTIC_INPUT_LIMIT]
return retained, len(semantic_inputs) - len(retained)
[docs]
def contract(self) -> ArtifactContract:
"""Return the mechanical consumption contract for the artifact.
Affordances are mechanical compatibility entries derived from the
capability registry's exact Artifact consumer edges,
not recommendations.
Returns:
ArtifactContract listing artifact_schema, issues,
affordances, and boundary_ports.
Example:
>>> contract = frame.contract()
>>> contract.artifact_schema.columns
Constraints:
Does not materialize a data copy.
"""
registry_module = importlib.import_module("marivo.analysis._capabilities.registry")
model_module = importlib.import_module("marivo.analysis._capabilities.model")
registry = registry_module.REGISTRY
operator_cls = model_module.OperatorCapability
read_cls = model_module.ReadCapability
validation_module = importlib.import_module("marivo.analysis._capabilities.validation")
family = type(self).__name__
consumer_targets = (
tuple(dict.fromkeys(edge.target for edge in registry.artifact_consumer_edges(family)))
if family in model_module.ARTIFACT_FAMILIES
else ()
)
affordances: list[ArtifactAffordance] = []
for target in consumer_targets:
cap_id = target.canonical_id
if cap_id is None or cap_id == "boundary.to_pandas":
continue
desc = registry.by_id(cap_id)
if isinstance(desc, read_cls):
# Receiver-owned reads have current-state-specific affordance
# construction in their concrete Artifact subclasses.
continue
assert isinstance(desc, operator_cls)
artifact_parameters = tuple(
parameter
for parameter, families in desc.accepted_inputs.items()
if family in {str(item) for item in families}
)
if artifact_parameters and not any(
validation_module.evaluate_artifact_admission(
desc.id,
parameter,
self,
).allowed
for parameter in artifact_parameters
):
continue
hidden_parameters: set[str] = set()
semantic_kind = getattr(self.meta, "semantic_kind", None)
if desc.id == "attribute" and family == "DeltaFrame" and semantic_kind != "funnel":
hidden_parameters.add("target")
if desc.id == "compare" and family == "EventFrame" and semantic_kind == "funnel":
hidden_parameters.add("alignment")
input_requirements: list[ArtifactInputRequirement] = []
for parameter in sorted({*desc.accepted_inputs, *desc.parameter_help}):
if parameter in hidden_parameters:
continue
families = desc.accepted_inputs.get(parameter, frozenset())
parameter_help = desc.parameter_help.get(parameter)
input_requirements.append(
ArtifactInputRequirement(
parameter=parameter,
accepted_families=tuple(sorted(str(item) for item in families)),
bindable_from_current_artifact=(family in {str(item) for item in families}),
derivable_from_current_artifact=(
bool(parameter_help and parameter_help.derivable_from_current_artifact)
),
accepted_semantic_shapes=(
tuple(
sorted(
desc.artifact_admission[parameter].semantic_shapes.get(
family,
(),
)
)
)
if parameter in desc.artifact_admission
else ()
),
acquisition=(
parameter_help.acquisition if parameter_help is not None else None
),
help_targets=(
parameter_help.help_targets if parameter_help is not None else ()
),
)
)
output_family = _output_family_str(desc)
affordance = ArtifactAffordance(
capability_id=desc.id,
public_entrypoint=desc.public_entrypoint,
help_target=LiveHelpTarget(
surface="analysis",
canonical_id=desc.help_target,
),
input_requirements=tuple(input_requirements),
expected_output_family=output_family,
)
if desc.id in {"MetricFrame.components", "DeltaFrame.components"}:
from marivo.analysis.frames._component_contract import (
_component_frame_contract_precondition,
)
affordance = affordance.model_copy(
update={"preconditions": (_component_frame_contract_precondition(self),)}
)
# Suppress affordances with failed preconditions that lack visible repair.
if _affordance_visible(affordance):
affordances.append(affordance)
artifact_schema = self._build_schema()
semantic_inputs, semantic_inputs_omitted = self._bounded_semantic_inputs()
return ArtifactContract(
kind=self.meta.kind,
ref=self.meta.ref,
is_canonical=True,
artifact_schema=artifact_schema,
semantic_inputs=semantic_inputs,
semantic_inputs_omitted=semantic_inputs_omitted,
issues=self.meta.issues,
affordances=tuple(affordances),
boundary_ports=tuple(_build_boundary_ports(registry)),
temporal_contract=getattr(self.meta, "temporal_contract", None),
)
def _dataframe_copy(self) -> pd.DataFrame:
"""Return an internal defensive copy without public export shaping."""
return self._df.copy()
def _auxiliary_tables(self) -> tuple[_FrameAuxiliaryTable, ...]:
"""Return private persisted payloads owned by this frame.
Base frames own no auxiliary data. Shape-specific frames may override
this hook, but private tables never participate in public row reads.
"""
if self._auxiliary_frames:
raise ValueError(f"{type(self).__name__} does not accept auxiliary tables")
return ()
def _bind_auxiliary_receipts(
self,
meta: BaseFrameMeta,
receipts: tuple[_FrameAuxiliaryReceipt, ...],
) -> BaseFrameMeta:
"""Bind written auxiliary hashes into shape-specific metadata."""
if receipts:
raise ValueError(f"{type(self).__name__} does not accept auxiliary receipts")
return meta
def _public_column_names(self) -> list[str]:
"""Return the column names shared by every public frame read path."""
return _display_column_names(self._df.columns)
def _public_dataframe_view(self) -> pd.DataFrame:
"""Return a shallow view with public column names for bounded selection."""
df = self._df.copy(deep=False)
df.columns = self._public_column_names()
return df
def _export_dataframe(self) -> pd.DataFrame:
"""Return the DataFrame shape exposed by the terminal pandas boundary.
Decimal-valued columns are coerced to ``float64`` so terminal pandas
arithmetic works out of the box; every other dtype is preserved.
"""
return _coerce_decimal_columns(self._public_dataframe_view().copy())
[docs]
def to_pandas(self) -> pd.DataFrame:
"""Return an isolated DataFrame for complete reads or terminal custom work.
Takes no parameters. The returned copy supports inspection, plotting,
and presentation, or methods outside the installed typed surface.
Keep supported analytical calculations in typed flow; discover them
with ``marivo.help("analysis.methods")`` before treating work as custom.
For example, ``rows = frame.to_pandas()`` reads complete result rows;
``rows.to_dict(orient="records")`` prepares them for presentation.
Keep the source Artifact identity alongside those rows. The copy does
not carry lineage, metadata, session ownership, or Evidence guarantees
and cannot re-enter typed analysis. The original frame remains usable
by typed methods and is unaffected by mutations of the copy.
Numeric columns surfaced as ``decimal.Decimal`` are returned as
``float64``; see :meth:`_export_dataframe`.
"""
return self._export_dataframe()
def __getitem__(self, key: Any) -> Any:
return self._public_dataframe_view()[key].copy()
@property
def shape(self) -> tuple[int, int]:
return self._df.shape
@property
def row_count(self) -> int:
"""Return materialized rows and fail closed on persisted metadata drift."""
self._assert_row_count_consistent()
return int(self._df.shape[0])
@property
def columns(self) -> list[str]:
return self._public_column_names()
def __len__(self) -> int:
return len(self._df)
def __iter__(self) -> Iterator[str]:
return iter(self.columns)
def __setitem__(self, key: Any, value: Any) -> None:
raise FrameMutationError(
message="frame is immutable; choose a typed method for analytical changes",
)
def __add__(self, other: Any) -> Any:
raise FrameMutationError(
message="frame arithmetic is blocked; choose a typed method for the analytical intent",
)
def __sub__(self, other: Any) -> Any:
raise FrameMutationError(
message="frame arithmetic is blocked; choose a typed method for the analytical intent",
)
def __mul__(self, other: Any) -> Any:
raise FrameMutationError(
message="frame arithmetic is blocked; choose a typed method for the analytical intent",
)
def __truediv__(self, other: Any) -> Any:
raise FrameMutationError(
message="frame arithmetic is blocked; choose a typed method for the analytical intent",
)
def _preview_rows_provider(self) -> Iterator[tuple[str, ...]]:
columns = self._public_column_names()
for row in self._df.itertuples(index=False, name=None):
yield tuple(str(_preview_cell(value)) for value in row[: len(columns)])
def _repr_identity(self) -> str:
return f"{type(self).__name__} ref={self.meta.ref} rows={self.meta.row_count}"
def _evidence_status_token(self, *, digest_unavailable: bool) -> str | None:
if digest_unavailable:
return f"status={self.meta.evidence_status} digest=unavailable"
if self.meta.evidence_digest is not None or self.meta.evidence_status in {
"partial",
"unavailable",
}:
return f"status={self.meta.evidence_status}"
return None
def _repr_html_(self) -> None:
return None
def _append_evidence_sections(
self,
card: Card,
*,
include_digest_items: bool = True,
include_quality_issues: bool = True,
) -> Card:
issues = tuple(
issue
for issue in self.meta.issues
if include_quality_issues or not isinstance(issue, DataQualityIssue)
)
if issues:
from marivo.analysis.evidence.summary import render_artifact_issue
card.listing(
"issues",
(
f"{issue.severity} {issue.kind}: {render_artifact_issue(issue)}"
for issue in issues
),
)
digest = self.meta.evidence_digest
digest_unavailable = any(
issue.kind == "evidence_digest_unavailable" for issue in self.meta.issues
)
status = self._evidence_status_token(digest_unavailable=digest_unavailable)
summary = [status] if status else []
if digest is not None:
from marivo.analysis.evidence.summary import render_digest_item, render_digest_selection
summary.append(
f"items={len(digest.items)}" if digest.items else "no evidence findings emitted"
)
selection = render_digest_selection(digest)
if selection is not None:
summary.append(f"selection={selection}")
if digest.omissions.omitted_items:
summary.append(f"omitted={digest.omissions.omitted_items}")
if digest_unavailable or (digest is not None and digest.omissions.omitted_items):
summary.append("recover=artifact.findings(limit=20)")
if summary:
card.field("evidence", " ".join(summary))
if digest is not None:
if include_digest_items and digest.items:
card.listing("evidence items", (render_digest_item(item) for item in digest.items))
if digest.boundaries:
card.listing(
"inference boundaries", (boundary.kind for boundary in digest.boundaries)
)
return card
def _base_card(self) -> Card:
"""Build the shared card header and fields before any preview table."""
card = self._header_card()
self._append_evidence_sections(card)
return card
def _header_card(self, status_prefix: str | None = None) -> Card:
"""Build identity, state, and purpose before family-specific context."""
card = Card(identity=self._repr_identity(), available=self._AVAILABLE_ENTRIES)
if status_prefix:
card.status(status_prefix)
if self.meta.analysis_purpose:
card.field("analysis_purpose", self.meta.analysis_purpose)
return card
def _card(self) -> Card:
return self._append_preview_table(self._base_card())
def _append_preview_table(self, card: Card, *, label: str = "preview") -> Card:
"""Append the shared bounded row preview after decision-critical context."""
return card.lazy_table(
columns=self._public_column_names(),
rows_provider=self._preview_rows_provider,
row_count=len(self._df),
label=label,
show_omission_counts=True,
bounded_row_limit=_DEFAULT_FRAME_PREVIEW_ROWS,
column_alignments=tuple(
(pd.api.types.is_numeric_dtype(dtype) and not pd.api.types.is_bool_dtype(dtype))
or _is_decimal_column(self._df.iloc[:_DEFAULT_FRAME_PREVIEW_ROWS, index])
for index, dtype in enumerate(
self._df.dtypes.iloc[: len(self._public_column_names())]
)
),
recovery=f"session.artifact('{self.meta.ref}').to_pandas()",
)