"""Canonical application-layer source resolution contracts.
The source mesh is the calculation-facing envelope for values derived from
bucket-local ledgers, invoices, prior filings, profile facts, borrador data,
relation prefill, and other registry-declared sources. A
:class:`CalculationSourceContext` binds the active bucket, modelo,
:class:`Period`, and selected :class:`ModeloRevision`; each
:class:`ModeloSourceResolver` claims one or more :class:`BindingSourceKind`
members and returns a :class:`CalculationSourceResolution`.
``CalculationSourceResolution`` is the single resolved-source carrier consumed
by modelo calculation. It carries decimal, enum, date, row-indexed binding,
relation, bound-casilla, detail-row, transaction-id, diagnostic, and provenance
channels. Exclusive merges use :func:`merge_source_resolutions`; precedence overlays use
:func:`merge_source_resolutions_by_precedence`; and
:func:`collect_unhandled_source_diagnostics` is the no-silent-blank safety net
for declared binding sources without an enrolled resolver.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from datetime import date, datetime
from decimal import Decimal
from enum import StrEnum
from types import MappingProxyType
from typing import Literal, NamedTuple, Protocol, Self, runtime_checkable
from pydantic import BaseModel, Field, field_serializer, field_validator, model_validator
from ...core import STRICT_FROZEN_CONFIG as _STRICT_FROZEN
from ...core import BindingSourceKind, Period
from ...core.decimal import coerce_decimal
from ...core.errors import CoreValidationError
from ...core.i18n import tr
from ...core.identity import BucketId
from ...core.logging import get_logger
from ...domain.calculations.registry import (
BindingId,
CasillaId,
LegalRefId,
ModeloId,
ModeloRevision,
RelationId,
SourceRefId,
)
from ...domain.modelos import ModeloDetailRow
from ._errors import AggregationValidationError, t
RowBindingKey = tuple[BindingId, int]
RowBindingValue = str | Decimal
[docs]
class SourceMeshError(CoreValidationError):
"""Raised when a ``CalculationSourceMesh`` field validator rejects an invariant.
Replaces bare :exc:`ValueError` at the ``owned_sources`` uniqueness / blank
guards and the ``source_transaction_ids`` uniqueness / blank guards so
callers receive a typed, registry-bound, localized error. Inherits from
:class:`~core.errors.CoreValidationError` (which inherits from
:exc:`ValueError`) so pydantic field validators surface it through
``ValidationError`` without special handling.
"""
def __init__(self, message_key: str) -> None:
super().__init__(message_key, translated_message=message_key)
_log = get_logger(__name__)
CalculationSourceDiagnosticReason = Literal[
"duplicate_binding_owner",
"duplicate_bound_casilla_owner",
"duplicate_relation_owner",
"source_issue",
"unresolved_binding",
"storage_degraded",
"source_domain_not_ready",
"unhandled_binding_source",
"unrouted_observation",
"oss_no_live_source",
"missing_transaction_evidence",
"administrador_retencion_rate_mismatch",
"official_box_unpopulated",
"prior_payment_not_deducted",
"prior_payment_minoracion_not_captured",
"settlement_not_computed",
"prorrata_especial_obligatoria",
"prorrata_especial_check_unavailable",
"dt12_regime_window_closed",
"dt12_regime_window_unverified",
"dt12_parcial_rescate_guidance",
]
def _binding_source_for_token(value: object) -> BindingSourceKind | None:
if isinstance(value, BindingSourceKind):
return value
if not isinstance(value, str):
return None
token = value.strip()
if not token:
return None
try:
return BindingSourceKind(token)
except ValueError:
return None
def _infer_binding_source(payload: object) -> object:
"""Hydrate ``binding_source`` when the free ``source_kind`` token is canonical."""
if not isinstance(payload, Mapping):
return payload
data = dict(payload)
source = _binding_source_for_token(data.get("source_kind"))
if isinstance(data.get("source_kind"), BindingSourceKind):
data["source_kind"] = data["source_kind"].value
explicit = data.get("binding_source")
if explicit is None:
if source is not None:
data["binding_source"] = source
return data
explicit_source = _binding_source_for_token(explicit)
if explicit_source is not None:
data["binding_source"] = explicit_source
if source is not None and explicit_source is not None and source is not explicit_source:
raise SourceMeshError("aggregation.source_mesh.errors.binding_source_mismatch")
return data
[docs]
class DeferredSourceTarget(NamedTuple):
"""The governed promotion target of a deferred binding source kind.
A deferred kind has no live mesh resolver yet, so it produces a standing
calculate-path advisory rather than a silent blank. Re-ratification per the
deferrals ADR replaces the former free-prose deferral comments with this
structured annotation so the deferral set is governed, not merely
enumerated: every deferred kind names the decision that owns it and the
condition that promotes it, and a kind whose trigger has fired but which
remains deferred is a mechanically-detectable finding at the swarm-audit
cadence.
Members:
owning_adr: The decision-record stem that ratifies this deferral and its
promotion target.
trigger: The condition under which the kind should be promoted to a live
mesh binding (a dependency for the IVA kinds; a per-modelo review
for the informativa detail-row kinds, which carry no promotion date).
promotion_depends_on: For a kind gated on another source kind landing,
the source kind it waits on. ``None`` for kinds whose trigger is a
human review rather than a mechanical source-kind dependency. When a
named dependency has itself been promoted out of the deferred set,
this kind's trigger has fired.
"""
owning_adr: str
trigger: str
promotion_depends_on: BindingSourceKind | None = None
# Source kinds that are explicitly deferred — no mesh resolver is built yet, but
# they are known to the system and must produce a standing advisory on
# source_diagnostics rather than a silent blank. The S26 boundary gate (in
# _calculation_actions) accepts them without flagging them as unknown-novel
# sources, and the S08 safety net emits the advisory while keeping them off the
# manual_sources allowlist. Each carries a typed promotion target (owning ADR +
# trigger) per the deferrals re-ratification; ``DEFERRED_SOURCE_KINDS`` is
# derived from the mapping so the membership set and its governance cannot drift.
DEFERRED_SOURCE_KIND_TARGETS: Mapping[BindingSourceKind, DeferredSourceTarget] = MappingProxyType(
{
# Informativa detail-row kinds (Sheets-pull-only, no resolver design):
# re-ratified with no promotion date; the review trigger is the modelo's
# next hardening campaign or an operator filing need, whichever comes
# first, and promotion requires its own grounded design ADR.
BindingSourceKind.RELATED_PARTY_OPERATION: DeferredSourceTarget(
owning_adr="2026-07-02-arch-remediation-source-kind-deferrals-adr",
trigger=(
"No promotion date. Review at M232's next hardening campaign or an operator filing need; "
"promotion needs its own grounded ADR (row taxonomy, evidence shape, detail-record fold)."
),
),
BindingSourceKind.REFUND_OPERATION: DeferredSourceTarget(
owning_adr="2026-07-02-arch-remediation-source-kind-deferrals-adr",
trigger=(
"No promotion date. Review at M360's next hardening campaign or an operator filing need; "
"promotion needs its own grounded ADR (row taxonomy, evidence shape, detail-record fold)."
),
),
BindingSourceKind.DONATIVO_DONOR: DeferredSourceTarget(
owning_adr="2026-07-02-arch-remediation-source-kind-deferrals-adr",
trigger=(
"No promotion date. Review at M182's next hardening campaign or an operator filing need; "
"promotion needs its own grounded ADR (row taxonomy, evidence shape, detail-record fold)."
),
),
},
)
# Derived so the membership set and its governance annotations cannot diverge:
# every deferred kind is a key in DEFERRED_SOURCE_KIND_TARGETS.
DEFERRED_SOURCE_KINDS: frozenset[BindingSourceKind] = frozenset(DEFERRED_SOURCE_KIND_TARGETS)
# Source kinds reserved-undeclared: a member that exists in the closed taxonomy
# but carries no registry binding and no resolver yet (counterpart / invoice-shaped
# headroom). They are neither enrolled nor deferred-with-advisory; the disposition
# registry records them RESERVED so the parity gate accounts for every member.
RESERVED_SOURCE_KINDS: frozenset[BindingSourceKind] = frozenset(
{
BindingSourceKind.PURCHASE_INVOICE_EVIDENCE,
BindingSourceKind.LEDGER_TRANSACTION,
},
)
[docs]
class CallerOverrideDisposition(StrEnum):
"""Whether the calculate path permits a caller override of a source's value.
The override disposition axis of the caller-override precedence ladder
(aggregation-taxonomy ADR ruling D2).
Members:
LOCK: Deterministic bucket-owned resolvers (the ledger aggregations and
the invoice families). A caller override is REJECTED so the persisted
revision faithfully reflects the sources it aggregates.
CARRY: Carry-style sources (previous_filing, relation_prefill, the
IVA-compensation annual partition, and prorrata regularizacion).
A caller override of an
automatically-carried prior value is legitimate and must reach the
engine, so these are EXCLUDED from the post-merge caller-override
guard.
"""
LOCK = "lock"
CARRY = "carry"
[docs]
class CallerOverridePrecedenceTier(NamedTuple):
"""One ordered tier of the calculate-path caller-override precedence ladder.
Carries the tier name, the source kinds it owns, and the override disposition
the guard applies to them. The ordered ladder is the single declaration the
caller-override guard sets are derived from — :data:`CALLER_OVERRIDE_PRECEDENCE_LADDER`
replaces the hand-listed lock / carry frozensets, and a conformance test binds
the policy's derived sets to it so the two cannot silently diverge.
"""
name: str
source_kinds: frozenset[BindingSourceKind]
disposition: CallerOverrideDisposition
#: The calculate-path caller-override precedence ladder as ordered tier data
#: (aggregation-taxonomy ADR ruling D2), lowest-precedence tier first. The
#: guard's lock and carry source sets are the unions of the LOCK- and
#: CARRY-disposition tiers (see :func:`precedence_ladder_sources`). This encodes
#: the override DISPOSITION axis only; the merge OVERLAY order (profile < mesh
#: backend < borrador < caller, later tier wins) is enforced separately by
#: :func:`merge_source_resolutions`.
CALLER_OVERRIDE_PRECEDENCE_LADDER: tuple[CallerOverridePrecedenceTier, ...] = (
CallerOverridePrecedenceTier(
name="deterministic_lock",
source_kinds=frozenset(
{
BindingSourceKind.LEDGER_IVA_AGGREGATION,
BindingSourceKind.LEDGER_RENTA_EXPENSE_AGGREGATION,
BindingSourceKind.LEDGER_RENTA_INCOME_AGGREGATION,
BindingSourceKind.LEDGER_RENTA_GASTO_AGGREGATION,
BindingSourceKind.LEDGER_IMPATRIADO_INCOME_AGGREGATION,
BindingSourceKind.LEDGER_OSS_AGGREGATION,
BindingSourceKind.COLLECTIBLE_INVOICE,
BindingSourceKind.PAYABLE_INVOICE,
},
),
disposition=CallerOverrideDisposition.LOCK,
),
CallerOverridePrecedenceTier(
name="carry_forward",
source_kinds=frozenset(
{
BindingSourceKind.PREVIOUS_FILING,
BindingSourceKind.RELATION_PREFILL,
BindingSourceKind.IVA_COMPENSATION_ANNUAL_PARTITION,
BindingSourceKind.PRORRATA_REGULARIZACION,
},
),
disposition=CallerOverrideDisposition.CARRY,
),
)
[docs]
def precedence_ladder_sources(disposition: CallerOverrideDisposition) -> frozenset[BindingSourceKind]:
"""Union of the source kinds carried by every ladder tier of ``disposition``.
The single derivation the caller-override policy sets read, so a source kind's
lock-vs-carry disposition is declared once in
:data:`CALLER_OVERRIDE_PRECEDENCE_LADDER` rather than hand-listed per set.
Returns:
The :class:`BindingSourceKind` members carried by every ladder tier
whose disposition matches *disposition*.
"""
return frozenset(
kind
for tier in CALLER_OVERRIDE_PRECEDENCE_LADDER
if tier.disposition is disposition
for kind in tier.source_kinds
)
[docs]
class BindingSourceDisposition(StrEnum):
"""Where a binding source kind resolves on the live calculate mesh.
The single closed answer to "where does source X resolve" for every
:class:`~core.BindingSourceKind` member, replacing the four scattered
enrollment structures (the ``merge_source_resolutions`` resolver tuple, the
pre-mesh-handled set, ``DEFERRED_SOURCE_KINDS``, and the per-modelo service
provider enum).
"""
ENROLLED = "enrolled" # routed by an active resolver / pre-mesh tier on the live calculate path
DEFERRED = "deferred" # known but no resolver yet; emits a standing advisory, never a silent blank
RESERVED = "reserved" # in the taxonomy but no binding and no resolver yet (counterpart/invoice headroom)
[docs]
def build_binding_source_dispositions(
enrolled_sources: frozenset[BindingSourceKind],
) -> Mapping[BindingSourceKind, BindingSourceDisposition]:
"""Classify every :class:`BindingSourceKind` member by its live mesh :class:`BindingSourceDisposition`.
``enrolled_sources`` is the LIVE enrolled set read at execution time -- the
union of every active resolver's ``owned_sources`` plus the pre-mesh tiers and
``manual_input`` -- so no disposition is hard-coded; a newly-enrolled source
(e.g. withholding, or profile / borrador now folded into the mesh) is reflected
automatically. ``DEFERRED_SOURCE_KINDS`` and ``RESERVED_SOURCE_KINDS`` supply the
other two states. Raises if a member is in two states at once, or in none
(an unaccounted source kind -- the "neither set contains the other" defect).
"""
dispositions: dict[BindingSourceKind, BindingSourceDisposition] = {}
for member in BindingSourceKind:
states = (
(member in enrolled_sources, BindingSourceDisposition.ENROLLED),
(member in DEFERRED_SOURCE_KINDS, BindingSourceDisposition.DEFERRED),
(member in RESERVED_SOURCE_KINDS, BindingSourceDisposition.RESERVED),
)
matched = [disposition for present, disposition in states if present]
if len(matched) != 1:
raise AggregationValidationError(
t("aggregation.source_mesh.errors.ambiguous_source_disposition"),
context={
"source_kind": member.value,
"matched_dispositions": [disposition.value for disposition in matched],
},
)
dispositions[member] = matched[0]
return MappingProxyType(dispositions)
[docs]
class CalculationSourceContext(BaseModel):
"""Context supplied to a calculation source resolver.
The ``period`` field is the typed :class:`~core.Period` value
carrying both the filing year and the bare registry period code. Consumers
that need the raw token for a downstream ``str``-typed API should use
``context.period.registry_token``; those that need only the year can use
``context.period.year`` (which mirrors ``context.filing_year``).
"""
model_config = _STRICT_FROZEN
bucket_id: BucketId
modelo: str = Field(min_length=1, max_length=16)
filing_year: int = Field(ge=2000, le=2099)
period: Period
revision: ModeloRevision
calculated_at: datetime | None = None
[docs]
class CalculationSourceDiagnostic(BaseModel):
"""Diagnostic emitted while resolving source-backed calculation values."""
model_config = _STRICT_FROZEN
reason: CalculationSourceDiagnosticReason
source_kind: str = Field(min_length=1, max_length=64)
binding_source: BindingSourceKind | None = None
"""Canonical binding source when ``source_kind`` names one; ``None`` for advisory categories."""
message: str = Field(min_length=1, max_length=512)
resolver_id: str | None = Field(default=None, min_length=1, max_length=128)
binding_id: BindingId | None = None
relation_id: RelationId | None = None
casilla_id: CasillaId | None = None
out_of_window_count: int | None = Field(default=None, ge=1)
out_of_window_min_filing_date: date | None = None
out_of_window_max_filing_date: date | None = None
@model_validator(mode="before")
@classmethod
def _set_binding_source(cls, value: object) -> object:
return _infer_binding_source(value)
@model_validator(mode="after")
def _validate_out_of_window_summary(self) -> Self:
summary_fields = (
self.out_of_window_count,
self.out_of_window_min_filing_date,
self.out_of_window_max_filing_date,
)
if all(value is None for value in summary_fields):
return self
if any(value is None for value in summary_fields):
raise SourceMeshError("aggregation.source_mesh.errors.out_of_window_summary_incomplete")
if self.out_of_window_max_filing_date < self.out_of_window_min_filing_date:
raise SourceMeshError("aggregation.source_mesh.errors.out_of_window_summary_date_span_invalid")
return self
[docs]
def out_of_window_summary_message(
*,
count: int,
min_filing_date: date,
max_filing_date: date,
) -> str:
"""Return the standard source-diagnostic message for summarized period exclusions."""
return (
f"{count} ledger transaction(s) have filing dates outside the requested period "
f"({min_filing_date.isoformat()}..{max_filing_date.isoformat()}); "
"excluded by period before classification"
)
[docs]
def out_of_window_summary_source_diagnostic(
*,
source_kind: str,
resolver_id: str,
count: int,
min_filing_date: date,
max_filing_date: date,
) -> CalculationSourceDiagnostic:
"""Build one structured source diagnostic for summarized ``OUTSIDE_PERIOD`` rows."""
return CalculationSourceDiagnostic(
reason="source_issue",
source_kind=source_kind,
resolver_id=resolver_id,
message=out_of_window_summary_message(
count=count,
min_filing_date=min_filing_date,
max_filing_date=max_filing_date,
),
out_of_window_count=count,
out_of_window_min_filing_date=min_filing_date,
out_of_window_max_filing_date=max_filing_date,
)
[docs]
class CalculationSourceProvenance(BaseModel):
"""Stable source object provenance produced by a resolver."""
model_config = _STRICT_FROZEN
source_kind: str = Field(min_length=1, max_length=64)
binding_source: BindingSourceKind | None = None
"""Canonical binding source when ``source_kind`` names one; ``None`` for non-binding provenance."""
source_ref: str = Field(min_length=1, max_length=256)
fingerprint: str | None = Field(default=None, min_length=1, max_length=256)
relation_id: RelationId | None = None
source_modelo: ModeloId | None = None
source_filing_year: int | None = Field(default=None, ge=2000, le=2099)
source_periods: tuple[str, ...] = ()
source_casilla_ids: tuple[CasillaId, ...] = ()
legal_refs: tuple[LegalRefId, ...] = ()
source_refs: tuple[SourceRefId, ...] = ()
@model_validator(mode="before")
@classmethod
def _set_binding_source(cls, value: object) -> object:
return _infer_binding_source(value)
@model_validator(mode="after")
def _relation_provenance_is_complete(self) -> CalculationSourceProvenance:
if self.relation_id is None:
return self
if (
self.source_modelo is None
or self.source_filing_year is None
or not self.source_periods
or not self.source_casilla_ids
or not self.legal_refs
or not self.source_refs
):
raise SourceMeshError("aggregation.source_mesh.errors.relation_provenance_incomplete")
return self
[docs]
class BorradorSourceProvenance(BaseModel):
"""Typed borrador-snapshot provenance carried on a source resolution.
The AEAT borrador snapshot is the one source whose downstream consumer
(``persist_calculation_revision``) needs more than the generic
:class:`CalculationSourceProvenance` row: it persists the originating
``borrador_snapshot_id`` and the sorted ``bindings_sourced_from_borrador``
trace onto the :class:`CalculationRevision`. Carrying that as ONE typed
sub-model keeps the generic :class:`CalculationSourceResolution` envelope
from accreting per-source named fields while preserving the trace as typed
data the call site reads directly -- never by parsing the
``borrador:{id}:binding:{bid}`` provenance ``source_ref`` strings.
"""
model_config = _STRICT_FROZEN
snapshot_id: str = Field(min_length=1, max_length=128)
bindings_sourced: tuple[BindingId, ...] = Field(default_factory=tuple)
[docs]
class CalculationSourceResolution(BaseModel):
"""Resolved values and provenance returned by one source resolver."""
model_config = _STRICT_FROZEN
resolver_id: str = Field(min_length=1, max_length=128)
owned_sources: tuple[BindingSourceKind, ...] = Field(default_factory=tuple)
binding_values: Mapping[BindingId, Decimal] = Field(default_factory=dict)
enum_binding_values: Mapping[BindingId, str] = Field(default_factory=dict)
date_binding_values: Mapping[BindingId, date] = Field(default_factory=dict)
row_binding_values: Mapping[RowBindingKey, RowBindingValue] = Field(default_factory=dict)
relation_values: Mapping[RelationId, Decimal] = Field(default_factory=dict)
unresolved_relation_ids: tuple[RelationId, ...] = Field(default_factory=tuple)
unresolved_binding_ids: tuple[BindingId, ...] = Field(default_factory=tuple)
bound_inputs_by_casilla_id: Mapping[CasillaId, Decimal] = Field(default_factory=dict)
detail_rows: tuple[ModeloDetailRow, ...] = Field(default_factory=tuple)
source_transaction_ids: Sequence[str] = Field(default_factory=tuple)
# Typed borrador provenance. Carried only by the borrador resolution
# (``Modelo100BorradorSourceResolver``); ``merge_source_resolutions``
# preserves it onto the merged result so the calculate call site reads the
# snapshot id and sourced-binding set as TYPED data and hands them to
# ``persist_calculation_revision``. ``None`` for every other resolver.
borrador_provenance: BorradorSourceProvenance | None = None
diagnostics: tuple[CalculationSourceDiagnostic, ...] = Field(default_factory=tuple)
provenance: tuple[CalculationSourceProvenance, ...] = Field(default_factory=tuple)
@field_validator("owned_sources", mode="before")
@classmethod
def _coerce_owned_sources(cls, value: object) -> object:
"""Hydrate known bare source-token strings to their :class:`BindingSourceKind` member.
The model carries :data:`~core.STRICT_FROZEN_CONFIG` (``strict=True``),
which disables string→enum coercion. Resolvers declare their owned source as a
canonical token and may pass either the member or its bare string value; this
before-validator maps each KNOWN bare string to its member (the
``BindingAggregation._coerce_op`` precedent in :mod:`core.aggregation`) so
the field stays strictly typed while a known token still validates. A blank
string raises :class:`SourceMeshError`; any other non-member value is left
untouched for the strict field to reject with its standard enum error, so a
genuine typo is still caught — without minting a new diagnostic locale key.
"""
if not isinstance(value, (tuple, list)):
return value
coerced: list[object] = []
for item in value:
if isinstance(item, BindingSourceKind):
coerced.append(item)
continue
if isinstance(item, str):
stripped = item.strip()
if not stripped:
raise SourceMeshError("aggregation.source_mesh.errors.owned_sources_blank")
try:
coerced.append(BindingSourceKind(stripped))
except ValueError:
# Unknown token: leave it for the strict typed field to reject.
coerced.append(item)
continue
coerced.append(item)
return tuple(coerced)
@field_validator("owned_sources")
@classmethod
def _owned_sources_are_unique(cls, value: tuple[BindingSourceKind, ...]) -> tuple[BindingSourceKind, ...]:
# After the before-coercer, every item is a canonical BindingSourceKind member
# (no blank/whitespace possible). Guard uniqueness and sort by the stable string
# value so the carrier is deterministic, preserving members (never downgrading
# them to bare str).
if len(value) != len(set(value)):
raise SourceMeshError("aggregation.source_mesh.errors.owned_sources_duplicate")
return tuple(sorted(value, key=lambda source: source.value))
@field_validator("binding_values")
@classmethod
def _freeze_binding_values(cls, value: Mapping[BindingId, Decimal]) -> Mapping[BindingId, Decimal]:
return MappingProxyType(dict(sorted(value.items())))
@field_validator("enum_binding_values")
@classmethod
def _freeze_enum_binding_values(cls, value: Mapping[BindingId, str]) -> Mapping[BindingId, str]:
return MappingProxyType(dict(sorted(value.items())))
@field_validator("date_binding_values")
@classmethod
def _freeze_date_binding_values(cls, value: Mapping[BindingId, date]) -> Mapping[BindingId, date]:
return MappingProxyType(dict(sorted(value.items())))
@field_validator("row_binding_values", mode="before")
@classmethod
def _coerce_row_binding_values(cls, value: object) -> object:
if isinstance(value, Mapping) or not isinstance(value, (list, tuple)):
return value
normalized: dict[tuple[object, object], object] = {}
for item in value:
if not isinstance(item, Mapping):
return value
row_value = item.get("value")
if item.get("value_kind") == "decimal":
row_value = coerce_decimal(row_value)
if row_value is None:
raise SourceMeshError("aggregation.source_mesh.errors.row_binding_value_invalid")
normalized[(item.get("binding_id"), item.get("row_index"))] = row_value
return normalized
@field_validator("row_binding_values")
@classmethod
def _freeze_row_binding_values(
cls,
value: Mapping[RowBindingKey, RowBindingValue],
) -> Mapping[RowBindingKey, RowBindingValue]:
normalized: dict[RowBindingKey, RowBindingValue] = {}
for (binding_id, row_index), row_value in value.items():
if row_index < 1:
raise SourceMeshError("aggregation.source_mesh.errors.row_binding_index_invalid")
normalized[(binding_id, row_index)] = row_value
return MappingProxyType(dict(sorted(normalized.items(), key=lambda item: (item[0][0], item[0][1]))))
@field_validator("relation_values")
@classmethod
def _freeze_relation_values(cls, value: Mapping[RelationId, Decimal]) -> Mapping[RelationId, Decimal]:
return MappingProxyType(dict(sorted(value.items())))
@field_validator("unresolved_relation_ids")
@classmethod
def _freeze_unresolved_relation_ids(cls, value: tuple[RelationId, ...]) -> tuple[RelationId, ...]:
normalized = tuple(item.strip() for item in value)
if any(not item for item in normalized):
raise SourceMeshError("aggregation.source_mesh.errors.unresolved_relation_ids_blank")
if len(normalized) != len(set(normalized)):
raise SourceMeshError("aggregation.source_mesh.errors.unresolved_relation_ids_duplicate")
return tuple(sorted(normalized))
@field_validator("unresolved_binding_ids")
@classmethod
def _freeze_unresolved_binding_ids(cls, value: tuple[BindingId, ...]) -> tuple[BindingId, ...]:
normalized = tuple(item.strip() for item in value)
if any(not item for item in normalized):
raise SourceMeshError("aggregation.source_mesh.errors.unresolved_binding_ids_blank")
if len(normalized) != len(set(normalized)):
raise SourceMeshError("aggregation.source_mesh.errors.unresolved_binding_ids_duplicate")
return tuple(sorted(normalized))
@field_validator("bound_inputs_by_casilla_id")
@classmethod
def _freeze_bound_inputs_by_casilla_id(cls, value: Mapping[CasillaId, Decimal]) -> Mapping[CasillaId, Decimal]:
return MappingProxyType(dict(sorted(value.items())))
@field_validator("source_transaction_ids")
@classmethod
def _freeze_source_transaction_ids(cls, value: Sequence[str]) -> tuple[str, ...]:
normalized = tuple(item.strip() for item in value)
if any(not item for item in normalized):
raise SourceMeshError("aggregation.source_mesh.errors.source_transaction_ids_blank")
if len(normalized) != len(set(normalized)):
raise SourceMeshError("aggregation.source_mesh.errors.source_transaction_ids_duplicate")
return tuple(sorted(normalized))
@field_serializer("binding_values")
def _serialize_binding_values(self, value: Mapping[BindingId, Decimal]) -> dict[BindingId, Decimal]:
return dict(value)
@field_serializer("enum_binding_values")
def _serialize_enum_binding_values(self, value: Mapping[BindingId, str]) -> dict[BindingId, str]:
return dict(value)
@field_serializer("date_binding_values")
def _serialize_date_binding_values(self, value: Mapping[BindingId, date]) -> dict[BindingId, date]:
return dict(value)
@field_serializer("row_binding_values")
def _serialize_row_binding_values(
self,
value: Mapping[RowBindingKey, RowBindingValue],
) -> tuple[dict[str, object], ...]:
return tuple(
{
"binding_id": binding_id,
"row_index": row_index,
"value": row_value,
"value_kind": "decimal" if isinstance(row_value, Decimal) else "text",
}
for (binding_id, row_index), row_value in value.items()
)
@field_serializer("relation_values")
def _serialize_relation_values(self, value: Mapping[RelationId, Decimal]) -> dict[RelationId, Decimal]:
return dict(value)
@field_serializer("unresolved_relation_ids")
def _serialize_unresolved_relation_ids(self, value: tuple[RelationId, ...]) -> tuple[RelationId, ...]:
return tuple(value)
@field_serializer("unresolved_binding_ids")
def _serialize_unresolved_binding_ids(self, value: tuple[BindingId, ...]) -> tuple[BindingId, ...]:
return tuple(value)
@field_serializer("bound_inputs_by_casilla_id")
def _serialize_bound_inputs_by_casilla_id(self, value: Mapping[CasillaId, Decimal]) -> dict[CasillaId, Decimal]:
return dict(value)
@field_serializer("source_transaction_ids")
def _serialize_source_transaction_ids(self, value: Sequence[str]) -> tuple[str, ...]:
return tuple(value)
[docs]
@runtime_checkable
class ModeloSourceResolver(Protocol):
"""Application port implemented by one calculation source adapter."""
@property
def resolver_id(self) -> str:
"""Stable resolver identifier for diagnostics and provenance."""
...
@property
def owned_sources(self) -> tuple[BindingSourceKind, ...]:
"""Registry :class:`BindingSourceKind` this resolver owns."""
...
[docs]
def resolve(self, context: CalculationSourceContext) -> CalculationSourceResolution:
"""Resolve source-backed calculation values for ``context``.
Returns a :class:`CalculationSourceResolution` carrying resolved
binding values, provenance, and any source diagnostics.
"""
...
[docs]
def merge_source_resolutions(
resolutions: Sequence[CalculationSourceResolution],
*,
resolver_id: str = "source_mesh",
) -> CalculationSourceResolution:
"""Merge resolver outputs and reject ambiguous ownership.
Returns a :class:`CalculationSourceResolution`.
"""
binding_values: dict[BindingId, Decimal] = {}
enum_binding_values: dict[BindingId, str] = {}
date_binding_values: dict[BindingId, date] = {}
row_binding_values: dict[RowBindingKey, RowBindingValue] = {}
relation_values: dict[RelationId, Decimal] = {}
unresolved_relation_ids: set[RelationId] = set()
unresolved_binding_ids: set[BindingId] = set()
bound_inputs_by_casilla_id: dict[CasillaId, Decimal] = {}
detail_rows: list[ModeloDetailRow] = []
source_transaction_ids: set[str] = set()
diagnostics: list[CalculationSourceDiagnostic] = []
provenance: list[CalculationSourceProvenance] = []
owned_sources: set[BindingSourceKind] = set()
binding_owners: dict[BindingId, str] = {}
row_binding_owners: dict[RowBindingKey, str] = {}
relation_owners: dict[RelationId, str] = {}
casilla_owners: dict[CasillaId, str] = {}
# The borrador resolution is the sole contributor of the typed borrador
# provenance; preserve it onto the merged result. Exactly one resolution
# carries a non-None borrador_provenance (the borrador resolver) so a plain
# last-writer-wins carry is unambiguous.
borrador_provenance: BorradorSourceProvenance | None = None
for resolution in resolutions:
owned_sources.update(resolution.owned_sources)
diagnostics.extend(resolution.diagnostics)
provenance.extend(resolution.provenance)
detail_rows.extend(resolution.detail_rows)
source_transaction_ids.update(resolution.source_transaction_ids)
unresolved_relation_ids.update(resolution.unresolved_relation_ids)
unresolved_binding_ids.update(resolution.unresolved_binding_ids)
if resolution.borrador_provenance is not None:
borrador_provenance = resolution.borrador_provenance
for binding_id, value in resolution.binding_values.items():
_claim_binding(binding_owners, binding_id, resolution.resolver_id)
binding_values[binding_id] = value
unresolved_binding_ids.discard(binding_id)
for binding_id, value in resolution.enum_binding_values.items():
_claim_binding(binding_owners, binding_id, resolution.resolver_id)
enum_binding_values[binding_id] = value
unresolved_binding_ids.discard(binding_id)
for binding_id, value in resolution.date_binding_values.items():
_claim_binding(binding_owners, binding_id, resolution.resolver_id)
date_binding_values[binding_id] = value
unresolved_binding_ids.discard(binding_id)
for row_binding_key, value in resolution.row_binding_values.items():
_claim_row_binding(row_binding_owners, row_binding_key, resolution.resolver_id)
row_binding_values[row_binding_key] = value
unresolved_binding_ids.discard(row_binding_key[0])
for relation_id, value in resolution.relation_values.items():
_claim_relation(relation_owners, relation_id, resolution.resolver_id)
relation_values[relation_id] = value
unresolved_relation_ids.discard(relation_id)
for casilla_id, value in resolution.bound_inputs_by_casilla_id.items():
_claim_bound_casilla(casilla_owners, casilla_id, resolution.resolver_id)
bound_inputs_by_casilla_id[casilla_id] = value
return CalculationSourceResolution(
resolver_id=resolver_id,
owned_sources=tuple(sorted(owned_sources)),
binding_values=binding_values,
enum_binding_values=enum_binding_values,
date_binding_values=date_binding_values,
row_binding_values=row_binding_values,
relation_values=relation_values,
unresolved_relation_ids=tuple(sorted(unresolved_relation_ids.difference(relation_values))),
unresolved_binding_ids=tuple(
sorted(
unresolved_binding_ids.difference(
binding_values,
enum_binding_values,
date_binding_values,
{binding_id for binding_id, _row_index in row_binding_values},
),
),
),
bound_inputs_by_casilla_id=bound_inputs_by_casilla_id,
detail_rows=tuple(detail_rows),
source_transaction_ids=tuple(sorted(source_transaction_ids)),
borrador_provenance=borrador_provenance,
diagnostics=tuple(diagnostics),
provenance=tuple(provenance),
)
[docs]
def merge_source_resolutions_by_precedence(
tiers: Sequence[CalculationSourceResolution],
*,
resolver_id: str = "source_mesh_precedence",
) -> CalculationSourceResolution:
"""Overlay tiers into one :class:`CalculationSourceResolution`.
Later tiers win on collision.
Unlike :func:`merge_source_resolutions` (which is EXCLUSIVE: a binding claimed
by two resolvers in one tier is a hard ``AggregationValidationError``), this
merge is a precedence OVERLAY: the binding / enum / date channels dict-merge in
tier order so a higher-precedence tier silently overrides a lower one. It is the
explicit form of the historical ``{**profile, **backend, **borrador, **caller}``
ladder: each tier is itself an intra-tier-exclusive
:func:`merge_source_resolutions` output, and the tiers are layered lowest -> highest.
The non-channel fields (relations, bound-casilla inputs, source transaction ids,
unresolved relations, diagnostics, provenance, owned_sources) accumulate across
tiers; ``borrador_provenance`` is carried from whichever tier supplies it
(exactly one does).
"""
binding_values: dict[BindingId, Decimal] = {}
enum_binding_values: dict[BindingId, str] = {}
date_binding_values: dict[BindingId, date] = {}
row_binding_values: dict[RowBindingKey, RowBindingValue] = {}
relation_values: dict[RelationId, Decimal] = {}
unresolved_relation_ids: set[RelationId] = set()
unresolved_binding_ids: set[BindingId] = set()
bound_inputs_by_casilla_id: dict[CasillaId, Decimal] = {}
detail_rows: list[ModeloDetailRow] = []
source_transaction_ids: set[str] = set()
diagnostics: list[CalculationSourceDiagnostic] = []
provenance: list[CalculationSourceProvenance] = []
owned_sources: set[BindingSourceKind] = set()
borrador_provenance: BorradorSourceProvenance | None = None
for tier in tiers:
owned_sources.update(tier.owned_sources)
diagnostics.extend(tier.diagnostics)
provenance.extend(tier.provenance)
detail_rows.extend(tier.detail_rows)
source_transaction_ids.update(tier.source_transaction_ids)
unresolved_relation_ids.update(tier.unresolved_relation_ids)
unresolved_binding_ids.update(tier.unresolved_binding_ids)
if tier.borrador_provenance is not None:
borrador_provenance = tier.borrador_provenance
# Precedence overlay: later tier wins (dict update), no exclusive claim.
binding_values.update(tier.binding_values)
enum_binding_values.update(tier.enum_binding_values)
date_binding_values.update(tier.date_binding_values)
row_binding_values.update(tier.row_binding_values)
for binding_id, _row_index in tier.row_binding_values:
unresolved_binding_ids.discard(binding_id)
bound_inputs_by_casilla_id.update(tier.bound_inputs_by_casilla_id)
for relation_id, value in tier.relation_values.items():
relation_values[relation_id] = value
unresolved_relation_ids.discard(relation_id)
return CalculationSourceResolution(
resolver_id=resolver_id,
owned_sources=tuple(sorted(owned_sources)),
binding_values=binding_values,
enum_binding_values=enum_binding_values,
date_binding_values=date_binding_values,
row_binding_values=row_binding_values,
relation_values=relation_values,
unresolved_relation_ids=tuple(sorted(unresolved_relation_ids.difference(relation_values))),
unresolved_binding_ids=tuple(
sorted(
unresolved_binding_ids.difference(
binding_values,
enum_binding_values,
date_binding_values,
{binding_id for binding_id, _row_index in row_binding_values},
),
),
),
bound_inputs_by_casilla_id=bound_inputs_by_casilla_id,
detail_rows=tuple(detail_rows),
source_transaction_ids=tuple(sorted(source_transaction_ids)),
borrador_provenance=borrador_provenance,
diagnostics=tuple(diagnostics),
provenance=tuple(provenance),
)
[docs]
def collect_unhandled_source_diagnostics(
revision: ModeloRevision,
*,
handled_sources: frozenset[str],
manual_sources: frozenset[str] = frozenset({"manual_input"}),
) -> tuple[CalculationSourceDiagnostic, ...]:
"""Return :class:`CalculationSourceDiagnostic` entries for revision bindings with no enrolled resolver.
Args:
revision: The :class:`ModeloRevision` whose bindings are inspected for missing resolvers.
handled_sources: Source kind strings already claimed by enrolled resolvers.
manual_sources: Source kind strings treated as intentionally unresolved.
"""
diagnostics: list[CalculationSourceDiagnostic] = []
for binding in revision.bindings:
source = str(binding.source)
if source in handled_sources or source in manual_sources:
continue
diagnostics.append(
CalculationSourceDiagnostic(
reason="unhandled_binding_source",
source_kind=source,
binding_id=binding.id,
message=f"binding {binding.id!r} declares source {source!r} with no enrolled resolver",
),
)
return tuple(diagnostics)
[docs]
def storage_degradation_resolution(
*,
resolver_id: str,
owned_sources: tuple[BindingSourceKind, ...],
source_kinds: Sequence[str],
error: BaseException,
) -> CalculationSourceResolution:
"""Return an empty :class:`CalculationSourceResolution` carrying secure-storage degradation diagnostics."""
normalized_sources = tuple(sorted({source.strip() for source in source_kinds if source.strip()}))
_log.debug(
"source mesh resolver storage degradation resolver_id=%s source_kinds=%s error_type=%s",
resolver_id,
",".join(normalized_sources),
type(error).__name__,
exc_info=(type(error), error, error.__traceback__),
)
return CalculationSourceResolution(
resolver_id=resolver_id,
owned_sources=owned_sources,
diagnostics=tuple(
CalculationSourceDiagnostic(
reason="storage_degraded",
source_kind=source_kind,
resolver_id=resolver_id,
message=tr("errors.integrity.integrity_storage_secure_object_unreadable"),
)
for source_kind in normalized_sources
),
)
def _claim_binding(owners: dict[BindingId, str], binding_id: BindingId, resolver_id: str) -> None:
existing = owners.get(binding_id)
if existing is None:
owners[binding_id] = resolver_id
return
raise AggregationValidationError(
t("aggregation.source_mesh.errors.duplicate_binding_owner"),
context={"binding_id": binding_id, "first_resolver": existing, "second_resolver": resolver_id},
)
def _claim_row_binding(owners: dict[RowBindingKey, str], row_binding_key: RowBindingKey, resolver_id: str) -> None:
existing = owners.get(row_binding_key)
if existing is None:
owners[row_binding_key] = resolver_id
return
binding_id, row_index = row_binding_key
raise AggregationValidationError(
t("aggregation.source_mesh.errors.duplicate_row_binding_owner"),
context={
"binding_id": binding_id,
"row_index": row_index,
"first_resolver": existing,
"second_resolver": resolver_id,
},
)
def _claim_bound_casilla(owners: dict[CasillaId, str], casilla_id: CasillaId, resolver_id: str) -> None:
existing = owners.get(casilla_id)
if existing is None:
owners[casilla_id] = resolver_id
return
raise AggregationValidationError(
t("aggregation.source_mesh.errors.duplicate_bound_casilla_owner"),
context={"casilla_id": casilla_id, "first_resolver": existing, "second_resolver": resolver_id},
)
def _claim_relation(owners: dict[RelationId, str], relation_id: RelationId, resolver_id: str) -> None:
existing = owners.get(relation_id)
if existing is None:
owners[relation_id] = resolver_id
return
raise AggregationValidationError(
t("aggregation.source_mesh.errors.duplicate_relation_owner"),
context={"relation_id": relation_id, "first_resolver": existing, "second_resolver": resolver_id},
)
__all__ = [
"CALLER_OVERRIDE_PRECEDENCE_LADDER",
"DEFERRED_SOURCE_KINDS",
"DEFERRED_SOURCE_KIND_TARGETS",
"RESERVED_SOURCE_KINDS",
"BindingSourceDisposition",
"BorradorSourceProvenance",
"CalculationSourceContext",
"CalculationSourceDiagnostic",
"CalculationSourceDiagnosticReason",
"CalculationSourceProvenance",
"CalculationSourceResolution",
"CallerOverrideDisposition",
"CallerOverridePrecedenceTier",
"DeferredSourceTarget",
"ModeloSourceResolver",
"RowBindingKey",
"RowBindingValue",
"build_binding_source_dispositions",
"collect_unhandled_source_diagnostics",
"merge_source_resolutions",
"merge_source_resolutions_by_precedence",
"out_of_window_summary_message",
"out_of_window_summary_source_diagnostic",
"precedence_ladder_sources",
"storage_degradation_resolution",
]