"""IVA compensation wallet reconciliation orchestration.
The live AEAT wallet is external state. Local Modelo 303 recurrence is
internal reconstruction. This module is the application boundary that turns
those evidence sources, plus an explicit taxpayer override when present, into
the effective binding decision consumed by Modelo 303 calculation.
The pure decision logic
(:func:`~domain.iva_compensation.reconciliation.reconcile_iva_compensation_wallet`
and its wallet/recurrence predicates) lives in
:mod:`~domain.iva_compensation.reconciliation`; it consumes structural
ports such as
:class:`~domain.iva_compensation.reconciliation.IvaCompensationWalletObservationProtocol`
and
:class:`~domain.iva_compensation.reconciliation.LocalIvaCompensationRecurrenceProtocol`
so the domain never imports the Sede adapter. This module orchestrates
:class:`~.observations_repository.CalculationObservationRepositoryProtocol` reads,
:class:`~.observations_repository.IvaWalletDecisionRepositoryProtocol` persistence, and
source-mesh resolution around that pure decision.
Binding resolution reads its active revision through a
:class:`~domain.calculations.registry.schema.RegistrySnapshot` supplied via the
source mesh context.
See Also:
:func:`~._binding_prefill.extract_modelo_303_local_iva_compensation_recurrence`
Reconstructs the local Modelo 303 recurrence compared with wallet
evidence.
:class:`~application.aggregation.source_mesh.CalculationSourceResolution`
The source-mesh envelope produced by
:class:`~application.calculations.iva_wallet_reconciliation.IvaWalletDecisionSourceResolver`.
"""
from __future__ import annotations
from datetime import date, datetime
from decimal import Decimal
from typing import TYPE_CHECKING, ClassVar
from pydantic import BaseModel, ConfigDict
from ...core.aggregation import BindingSourceKind, CalculationSourceLineageRole
from ...core.hashing import sha256_hex
from ...core.modelo import Modelo
from ...core.period import Period
from ...domain.calculations.registry.authority import PinnedAuthorityOperation
from ...domain.calculations.registry.facts.resolution import ResolvedScalarFact, ScalarFactQuery
from ...domain.calculations.registry.schema import RegistrySnapshot
from ...domain.calculations.registry.schema_base import DateAxis
from ...domain.iva_compensation.errors import IvaCompensationReconciliationInputError
from ...domain.iva_compensation.reconciliation import (
DEFAULT_MAX_WALLET_AGE_DAYS,
IvaCompensationOverride,
IvaCompensationReconciliationDecision,
IvaCompensationWalletObservationProtocol,
local_recurrence_authority_source,
reconcile_iva_compensation_wallet,
validate_wallet_matches_snapshot,
)
from ..aggregation.source_mesh import (
CalculationSourceContext,
CalculationSourceProvenance,
CalculationSourceResolution,
)
from .binding_prefill import BindingPrefillReport as _BindingPrefillReport
if TYPE_CHECKING:
from .binding_prefill import BindingPrefillReport, LocalIvaCompensationRecurrence
from .observations_repository import CalculationObservationRepositoryProtocol, IvaWalletDecisionRepositoryProtocol
[docs]
class IvaCompensationReconciliationReport(BaseModel):
"""Application-level reconciliation result for one Modelo 303 target.
Carries the domain
:class:`~domain.iva_compensation.reconciliation.IvaCompensationReconciliationDecision`
plus the :class:`~._binding_prefill.BindingPrefillReport` used to reconstruct
the local recurrence side of the comparison.
"""
model_config = ConfigDict(strict=True, frozen=True, extra="forbid", arbitrary_types_allowed=True)
decision: IvaCompensationReconciliationDecision
prefill_report: BindingPrefillReport
IvaCompensationReconciliationReport.model_rebuild(
_types_namespace={"BindingPrefillReport": _BindingPrefillReport},
)
del _BindingPrefillReport
[docs]
class IvaWalletDecisionSourceResolver:
"""Source-mesh adapter for persisted Modelo 303 IVA wallet decisions.
Owns ``iva_wallet_decision`` and materialises the
``modelo-303-compensacion-pendiente-anteriores`` binding from a persisted
:class:`~domain.iva_compensation.reconciliation.IvaCompensationReconciliationDecision`.
The returned :class:`~application.aggregation.source_mesh.CalculationSourceResolution`
carries the selected amount plus provenance for every authority source that
participated in the wallet/filed-history/local-recurrence decision.
"""
resolver_id: ClassVar[str] = "iva_wallet_decision"
owned_sources: ClassVar[tuple[BindingSourceKind, ...]] = (BindingSourceKind.IVA_WALLET_DECISION,)
binding_id = "modelo-303-compensacion-pendiente-anteriores"
def __init__(self, decision: IvaCompensationReconciliationDecision | None) -> None:
"""Initialize the resolver with the persisted reconciliation decision, if any."""
self._decision = decision
[docs]
def resolve(self, context: CalculationSourceContext) -> CalculationSourceResolution:
"""Resolve the Modelo 303 IVA wallet decision binding for ``context``.
Returns:
An empty resolution when the modelo is not Modelo 303 or no
decision was supplied, otherwise the resolved
:class:`~application.aggregation.source_mesh.CalculationSourceResolution`
carrying the selected amount and its provenance.
Raises:
IvaCompensationReconciliationInputError: When the decision does
not target this filing year and period, blocks the
calculation, or carries no selected amount.
"""
if context.modelo != Modelo("303").value or self._decision is None:
return CalculationSourceResolution(
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
)
decision = self._decision
if decision.target_year != context.filing_year or decision.target_period != context.period:
raise IvaCompensationReconciliationInputError(
translated_message="application.calculations.iva_wallet.errors.decision_target_mismatch",
context={
"decision_target_year": decision.target_year,
"decision_target_period": str(decision.target_period),
"context_filing_year": context.filing_year,
"context_period": str(context.period),
},
)
if decision.blocked:
raise IvaCompensationReconciliationInputError(
translated_message="application.calculations.iva_wallet.errors.decision_blocks_calculation",
context={
"divergence": str(decision.divergence),
"target_year": decision.target_year,
"target_period": str(decision.target_period),
},
)
if decision.selected_amount is None:
raise IvaCompensationReconciliationInputError(
translated_message="application.calculations.iva_wallet.errors.decision_no_selected_amount",
context={
"target_year": decision.target_year,
"target_period": str(decision.target_period),
},
)
fingerprint = f"sha256:{sha256_hex(decision.model_dump_json().encode('utf-8'))}"
from .observations_repository import iva_wallet_decision_event_key
primary_ref = iva_wallet_decision_event_key(decision)
return CalculationSourceResolution(
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
binding_values={self.binding_id: Decimal(decision.selected_amount)},
provenance=(
CalculationSourceProvenance(
resolver_id=self.resolver_id,
resolved_binding_source=BindingSourceKind.IVA_WALLET_DECISION,
contributor_source_kind=BindingSourceKind.IVA_WALLET_DECISION.value,
contributor_binding_source=BindingSourceKind.IVA_WALLET_DECISION,
lineage_role=CalculationSourceLineageRole.PRIMARY,
source_ref=primary_ref,
parent_source_ref=None,
# No terminal origin: ``iva_wallet_decision`` is a mesh-only
# source with no provider registration, so no declared
# expectation exists for a resolved class to be audited
# against.
terminal_origin=None,
fingerprint=fingerprint,
),
*tuple(
CalculationSourceProvenance(
resolver_id=self.resolver_id,
resolved_binding_source=BindingSourceKind.IVA_WALLET_DECISION,
contributor_source_kind=source.source_kind,
contributor_binding_source=_binding_source_or_none(source.source_kind),
lineage_role=CalculationSourceLineageRole.CONTRIBUTOR,
source_ref=source.source_locator,
parent_source_ref=primary_ref,
)
for source in decision.authority_sources
),
),
)
def _binding_source_or_none(source_kind: str) -> BindingSourceKind | None:
"""Project an authority-source token only when it belongs to the binding taxonomy."""
try:
return BindingSourceKind(source_kind)
except ValueError:
return None
def _resolve_first_period_compensation_amount(
*,
filing_year: int,
operation: PinnedAuthorityOperation,
) -> Decimal:
"""Resolve the governed first-period compensation amount."""
resolved = operation.resolve_governed_fact(
ScalarFactQuery(
fact_id="liva-art-99:first-period-compensation-zero",
date_axis=DateAxis.FILING_PERIOD,
effective_date=date(filing_year, 12, 31),
),
)
if not isinstance(resolved, ResolvedScalarFact):
raise TypeError("first-period compensation declaration must resolve as a scalar fact")
value = resolved.payload.value
if isinstance(value, bool) or not isinstance(value, (Decimal, int)):
raise TypeError("first-period compensation declaration must resolve as a numeric amount")
return Decimal(str(value))
def _resolve_reconciliation_repositories(
*,
repository: CalculationObservationRepositoryProtocol,
decision_repository: IvaWalletDecisionRepositoryProtocol,
persist: bool,
) -> tuple[CalculationObservationRepositoryProtocol, IvaWalletDecisionRepositoryProtocol]:
"""Validate the required repository pair and refuse a split secure backend."""
repo = repository
decision_repo = decision_repository
if persist and decision_repo.secure_object_repository.engine is not repo.secure_object_repository.engine:
raise IvaCompensationReconciliationInputError(
translated_message="application.calculations.iva_wallet.errors.decision_repository_backend_split",
context={"persist": persist, "decision_repository_supplied": True},
)
return repo, decision_repo
[docs]
def reconcile_modelo_303_iva_compensation(
snapshot: RegistrySnapshot,
*,
taxpayer_nif: str,
wallet: IvaCompensationWalletObservationProtocol | None,
repository: CalculationObservationRepositoryProtocol,
decision_repository: IvaWalletDecisionRepositoryProtocol,
override: IvaCompensationOverride | None = None,
decided_at: datetime | None = None,
max_wallet_age_days: int = DEFAULT_MAX_WALLET_AGE_DAYS,
treat_absent_recurrence_as_first_period: bool = False,
local_evidence_found_but_unusable: bool = False,
local_recurrence: LocalIvaCompensationRecurrence | None,
prefill_report: BindingPrefillReport,
persist: bool = True,
operation: PinnedAuthorityOperation,
) -> IvaCompensationReconciliationReport:
"""Resolve, compare, and optionally persist the Modelo 303 IVA wallet decision.
The function validates wallet evidence against the target
:class:`~domain.calculations.registry.schema.RegistrySnapshot`, reconstructs local recurrence through
:func:`~._binding_prefill.extract_modelo_303_local_iva_compensation_recurrence`,
delegates authority selection to
:func:`~domain.iva_compensation.reconciliation.reconcile_iva_compensation_wallet`,
and persists the resulting
:class:`~domain.iva_compensation.reconciliation.IvaCompensationReconciliationDecision`
through :class:`~.observations_repository.IvaWalletDecisionRepositoryProtocol` when
``persist`` is true.
Args:
snapshot: The :class:`~domain.calculations.registry.schema.RegistrySnapshot`
identifying the Modelo 303 target revision.
taxpayer_nif: Taxpayer identifier expected to match live wallet evidence.
wallet: Live AEAT wallet observation to reconcile, when available.
repository: Required
:class:`~application.calculations.observations_repository.CalculationObservationRepositoryProtocol`
used to read prior local recurrence.
decision_repository: Required
:class:`~application.calculations.observations_repository.IvaWalletDecisionRepositoryProtocol`
used to persist the resulting wallet authority. When persistence is
enabled, it must use the same encrypted storage backend as ``repository``.
override: Optional
:class:`~domain.iva_compensation.reconciliation.IvaCompensationOverride`
evidence when the operator has resolved a divergence.
decided_at: Decision timestamp override for deterministic replay and tests.
max_wallet_age_days: Maximum accepted age for live wallet evidence.
treat_absent_recurrence_as_first_period: When ``True`` and there is no
live wallet and no prior local recurrence, treat the target as the
taxpayer's FIRST IVA period: ``iva.compensacion-pendiente-periodos-anteriores``
is zero per LIVA art. 99.5 (no prior compensation balance can exist), yielding the non-blocking
``first_period_zero`` decision instead of the ``missing`` block. The
caller asserts first-period status (e.g. the calculate path verifies
no prior 303 compensation history exists). It NEVER fabricates a
non-zero balance: a present recurrence still flows through normally.
local_evidence_found_but_unusable: Whether the caller FOUND a prior
local record and could not interpret it as prior-compensation
evidence. Only the caller knows this: an absent amount looks the
same here whether nothing was stored or something was and could not
be read. It changes no outcome, only what the no-authority decision
says, so an operator is not told nothing exists while their own
prior record sits in the store.
local_recurrence: Required caller-resolved canonical recurrence, or
explicit ``None`` when the caller found no admissible recurrence.
prefill_report: Required report from the same recurrence resolution.
Reconciliation never performs a second repository fallback.
persist: Whether to store the resulting decision for later calculation replay.
operation: Caller-owned generation-pinned authority operation used to
resolve first-period governed facts.
The local side is not recomputed here. It is read through the same
previous-filing binding resolver used by the calculation chain.
Returns an
:class:`~application.calculations.iva_wallet_reconciliation.IvaCompensationReconciliationReport`.
See Also:
:class:`~application.calculations.iva_wallet_reconciliation.IvaWalletDecisionSourceResolver`
Converts a persisted non-blocking decision into the Modelo 303
``iva_wallet_decision`` source value.
"""
if str(getattr(snapshot.modelo, "id", snapshot.modelo)) != Modelo("303").value:
raise IvaCompensationReconciliationInputError(
translated_message="application.calculations.iva_wallet.errors.modelo_303_only",
context={"modelo": str(getattr(snapshot.modelo, "id", snapshot.modelo))},
)
snapshot_period = Period.from_year_and_code(snapshot.filing_year, snapshot.period)
if wallet is not None:
validate_wallet_matches_snapshot(
wallet,
taxpayer_nif=taxpayer_nif,
target_year=snapshot.filing_year,
target_period=snapshot_period,
)
_repo, decision_repo = _resolve_reconciliation_repositories(
repository=repository,
decision_repository=decision_repository,
persist=persist,
)
recurrence = local_recurrence
local_recurrence_amount = recurrence.amount if recurrence is not None else None
# First-period treatment: resolve the caller-asserted first-period amount
# through the selected registry revision, then pass it with the first-period flag so the decision
# is the non-blocking ``first_period_zero`` rather than the ``missing``
# block. A present recurrence still flows through normally.
#
# This never overrides a recurrence it can SEE, but seeing one is the
# caller's job: a ``None`` here says only that the caller produced no
# recurrence, never why. The claim that used to stand in this comment --
# that only the genuinely-absent case is mapped to zero -- was the defect
# rather than the guarantee, because a caller that had found a stored
# observation and could not read it arrived with the same ``None`` as a
# caller that had found nothing at all, and an unreadable envelope became a
# proven zero on the compensación. The caller must therefore pass the flag
# false whenever it saw evidence it could not use.
is_first_iva_period = treat_absent_recurrence_as_first_period and wallet is None and local_recurrence_amount is None
if is_first_iva_period:
local_recurrence_amount = _resolve_first_period_compensation_amount(
filing_year=int(snapshot.filing_year),
operation=operation,
)
decision = reconcile_iva_compensation_wallet(
taxpayer_nif=taxpayer_nif,
target_year=snapshot.filing_year,
target_period=snapshot_period,
target_registry_snapshot_ref=snapshot.snapshot_ref,
wallet=wallet,
local_recurrence_amount=local_recurrence_amount,
local_recurrence_source=local_recurrence_authority_source(recurrence),
override=override,
decided_at=decided_at,
max_wallet_age_days=max_wallet_age_days,
is_first_iva_period=is_first_iva_period,
local_evidence_found_but_unusable=local_evidence_found_but_unusable,
)
if persist:
decision_repo.save_decision(decision)
return IvaCompensationReconciliationReport(
decision=decision,
prefill_report=prefill_report,
)
__all__ = [
"IvaCompensationReconciliationReport",
"IvaWalletDecisionSourceResolver",
"reconcile_iva_compensation_wallet",
"reconcile_modelo_303_iva_compensation",
]