cadrumo.application.calculations.observations_repository module¶

Application contracts for past-filing casilla observations.

Stores RegistryModeloObservation records — (modelo, filing_year, period, casilla_values) — as encrypted audit envelopes in the SecureObjectRepository. Past-filing value rows are bound to CALCULATION_OBSERVATIONS_NAMESPACE; IVA wallet decisions are split between the latest-state IVA_WALLET_RECONCILIATION_DECISIONS_NAMESPACE and immutable IVA_WALLET_RECONCILIATION_DECISION_EVENTS_NAMESPACE namespaces. The records are the substrate read by PreviousFilingSourceResolver and RelationPrefillSourceResolver so annual modelos can roll up prior quarterlies, IVA prorrata can compute its four-year backward mean, IS BIN carryforward can replay prior-year bases imponibles negativas, and IVA regularización inversiones can apply its 5/10 year straight-line schedule.

Producers are out of scope for this module: the modelo filing flow will write here when an operator successfully files via the app, and the live-AEAT capture path will write here when justificantes are parsed. This module exposes only the typed read/write surface.

Sensitivity is SensitivityClass AUDIT — these records reconstruct exactly what was filed and so are identity-bearing tax substrate. They are stored encrypted at rest through an Envelope-wrapped repository.

The store is value-centric. Clean-state proof still has to join these rows with filing records, verification reports, and justificante evidence through evaluate_cross_period_clean_state().

require_decision_registry_coordinates_current(decision, *, operation)[source]¶

Refuse a persisted IVA decision whose target or source revision diverges.

Return type:

None

Parameters:

decision (IvaCompensationReconciliationDecision)
operation (PinnedAuthorityOperation)

class ObservationSourceKind(*values)[source]¶

Bases: StrEnum

Origin of a persisted calculation observation.

This classifies the observation provenance before a filing’s separate ExternalEvidenceKind is materialised. Only the three AEAT origins are official filing evidence; local app and operator rows may support calculation prefill but cannot establish filing-grade cross-period readiness.

APP_FILING¶

OPERATOR_MANUAL¶

AEAT_SEDE_JUSTIFICANTE¶

AEAT_SEDE_LIVE_CAPTURE¶

AEAT_CSV_REGISTER¶

property is_official_aeat: bool¶: Whether this provenance was observed from an AEAT source.

APP_FILING_SOURCE_KIND: Final[ObservationSourceKind]¶: Non-official provenance stamped on observations filed through the app.

is_official_aeat_observation_source(source_kind)[source]¶

Return whether an observation provenance is official AEAT evidence.

Unknown and aggregate values (for example "mixed" group evidence) fail closed as non-official rather than silently gaining filing authority.

Return type:: bool
Parameters:: source_kind (ObservationSourceKind | str)

class ResultDispositionProjection(**data)[source]¶

Bases: BaseModel

Validated Modelo 303 disposition evidence owned by an observation envelope.

The filed declaration_type is not a casilla. Keeping it here means a RegistryModeloObservation remains a calculation-only record while the envelope retains the typed disposition and how it was established.

Parameters:: data (Any)

disposition: ResultDisposition¶

provenance_kind: Literal['source_header', 'app_filing']¶

provenance_locator: str¶

class PriorDomiciliationElectionProjection(**data)[source]¶

Bases: BaseModel

Safe provenance for a Modelo 303 rectificativa’s prior-debit election.

A cancellation/modification is actionable only when the application has already joined the rectificativa to an AEAT-attested baseline filing and to that filed declaration’s submitted-file U header. The projection keeps the semantic election and redacted join coordinates, never an account or rendered header value.

Parameters:: data (Any)

election: PriorDomiciliationElection¶

baseline_filing_record_id: FilingRecordId | None¶

baseline_evidence_reference_id: str | None¶

baseline_result_disposition: ResultDisposition | None¶

baseline_source_header_locator: str | None¶

class ObservationOverride(**data)[source]¶

Bases: BaseModel

Audit of one operator figure that replaced the effective observation of a coordinate.

replaced_source_kind and replaced_values describe the envelope that was effective when the operator recorded the figure; both are empty when the coordinate held nothing. The replaced envelope itself is not destroyed: an official layer stays stored beside the pending-local one.

Parameters:: data (Any)

actor: ModeloActorLabel¶

reason: OperatorReason¶

recorded_at: UtcInstant¶

replaced_source_kind: ObservationSourceKind | None¶

replaced_values: dict[CasillaId, str]¶

class ObservationEnvelopePayload(**data)[source]¶

Bases: BaseModel

Canonical public persistence payload for filed observations.

This strict, frozen model is returned by CalculationObservationRepository reads and iteration. It keeps the RegistryModeloObservation calculation evidence separate from the application capture provenance: captured_at, constrained source_kind, optional member_nif, required stamped_revision_id, and source-specific source_metadata. The model does not encrypt that metadata; the secure repository envelope does.

captured_at is the canonical UtcInstant. A bare datetime field admitted a naive value, so a capture instant with no zone reached persistence and every later comparison against a UTC-aware instant was answering a different question than it appeared to.

Every persisted observation carries its source registry revision stamp so carry reads can reconfirm the value against the law-determined revision. A missing or structurally invalid stamp refuses at load; a valid but divergent stamp is refused later by the carry gate.

Parameters:: data (Any)

observation: RegistryModeloObservation¶

captured_at: UtcInstant¶

source_kind: ObservationSourceKind¶

member_nif: str | None¶

stamped_revision_id: RevisionId¶

source_metadata: dict[str, str]¶

source_headers: tuple[ObservedHeaderFact, ...]¶

result_disposition: ResultDispositionProjection | None¶

prior_domiciliation_election: PriorDomiciliationElectionProjection | None¶

property registry_snapshot_ref: RegistrySnapshotRef¶: Return the one canonical coordinate represented by this envelope.

m303_compensation_basis: M303CompensationBasisValue | None¶

override: ObservationOverride | None¶

property is_pending_local: bool¶: Return whether this envelope belongs to the pending-local layer.

class ObservationLayers(**data)[source]¶

Bases: BaseModel

Both stored observation layers of one (modelo, filing_year, period, member) coordinate.

official holds evidence observed from AEAT. pending_local holds a local filing or an operator figure that AEAT has not confirmed. Readers that want one answer use effective; the pending layer keeps its non-official source_kind, so filing-grade gates still refuse it.

Parameters:: data (Any)

modelo: str¶

filing_year: int¶

period: str¶

member_nif: str | None¶

official: ObservationEnvelopePayload | None¶

pending_local: ObservationEnvelopePayload | None¶

property effective: ObservationEnvelopePayload | None¶: Return the pending-local envelope when present, otherwise the official one.

require_observation_envelope_coordinates_current(payload, *, operation)[source]¶

Return a persisted observation only when its producing coordinate re-confirms.

Return type:

ObservationEnvelopePayload

Parameters:

payload (ObservationEnvelopePayload)
operation (PinnedAuthorityOperation)

class IvaWalletDecisionEnvelopePayload(**data)[source]¶

Bases: BaseModel

Serialisable wrapper for an IVA wallet reconciliation decision.

Parameters:: data (Any)

decision: IvaCompensationReconciliationDecision¶

decision_payload_digest(decision)[source]¶

Return the stable digest used to identify a reconciliation decision.

Return type:: str
Parameters:: decision (IvaCompensationReconciliationDecision)

require_observation_period(period)[source]¶

Refuse a non-period value before it enters a persisted observation key.

Return type:: Period
Parameters:: period (Period)

observation_key_for_token(modelo, filing_year, period_token)[source]¶

Stable repository key for a modelo/year/raw registry-period token triple.

Censo modelos can declare non-date registry tokens such as alta or modificacion that are not valid Period codes. The encrypted observation store still keys them by the same logical triple.

Return type:

str

Parameters:

modelo (str)
filing_year (int)
period_token (str)

observation_key(modelo, period)[source]¶

Stable repository key for a (modelo, Period) pair.

Validated through safe_repository_id() so each component is constrained to the SecureObjectRepository id contract before composition.

Return type:

str

Parameters:

modelo (str)
period (Period)

member_observation_key_for_token(modelo, filing_year, period_token, member_nif)[source]¶

Storage key for an observation keyed by a raw registry period token.

Return type:

str

Parameters:

modelo (str)
filing_year (int)
period_token (str)
member_nif (str | None)

member_observation_key(modelo, period, member_nif)[source]¶

Storage key for an observation, widened by a grupo member NIF when present.

When member_nif is None the key is the single-filer observation_key unchanged, so every existing consumer (the default previous_filing path, the multi-year resolver) keys identically. When set, a member segment is appended so two members’ filings for the same (modelo, filing_year, period) persist as distinct rows — the cross-member fan-in the 353<-322 per_grupo_member aggregation enumerates and sums.

That segment is the sha256 of the member’s tax_id_identity_token(), the same normalise-then-digest step every other identifier-bearing object key in the registry takes. The normalisation is the load-bearing half: appending the declared value verbatim made two spellings of ONE member address two rows, so a member whose identifier arrived lower-cased in one capture and space-padded in the next was counted twice by the very fan-in this widening exists to serve. The digest is the hygiene half – it keeps a real identifier out of the natural-key surface, which is addressing metadata rather than encrypted payload.

Return type:

str

Parameters:

modelo (str)
period (Period)
member_nif (str | None)

iva_wallet_decision_key(taxpayer_nif, target_period)[source]¶

Opaque latest-decision key for one taxpayer and Modelo 303 target period.

Secure-object payloads are encrypted, but object keys are storage metadata. Hash the taxpayer/period tuple so the repository does not expose NIF/NIE values in cleartext database rows.

Return type:

str

Parameters:

taxpayer_nif (str)
target_period (Period)

iva_wallet_decision_event_key(decision)[source]¶

Opaque immutable event key for one persisted reconciliation decision.

Return type:: str
Parameters:: decision (IvaCompensationReconciliationDecision)

validate_observation_casilla_ids(observation, *, operation, stamped_revision_id=None)[source]¶

Validate all observed and operand casillas against the selected revision.

The revision is chosen by law from the observation’s coordinates. Only when those coordinates alone are ambiguous – a filing year split between editions at a dated boundary, which an observation carries no date to resolve – does stamped_revision_id decide, and only among the law’s own candidates.

Return type:

str

Parameters:

observation (RegistryModeloObservation)
operation (PinnedAuthorityOperation)
stamped_revision_id (str | None)

class CalculationObservationStorageProtocol(*args, **kwargs)[source]¶

Bases: Protocol

Application-facing storage capability for atomic observation writes.

apply_batch(writes)[source]¶

Commit prepared observation/history writes atomically.

Return type:: None
Parameters:: writes (tuple[SecureObjectWrite, ...])

property engine: object¶: Expose only backend identity needed for same-store validation.

class CalculationObservationRepositoryProtocol(*args, **kwargs)[source]¶

Bases: Protocol

Application-owned read/write capability for filed observations.

load_observation(modelo, period)[source]¶

Load the effective observation of one single-filer coordinate.

Return type:

ObservationEnvelopePayload | None

Parameters:

modelo (str)
period (Period)

iter_modelo(modelo)[source]¶

Iterate the effective observation of every coordinate of one modelo.

Return type:: Iterator[ObservationEnvelopePayload]
Parameters:: modelo (str)

iter_records()[source]¶

Iterate the effective observation of every stored coordinate.

Return type:: Iterator[ObservationEnvelopePayload]

prepare_observation_envelope(observation, *, source_kind, stamped_revision_id, captured_at=None, member_nif=None, source_metadata=None, source_headers=(), result_disposition=None, prior_domiciliation_election=None, override=None)[source]¶

Validate and normalize one observation before persistence.

Return type:

ObservationEnvelopePayload

Parameters:

observation (RegistryModeloObservation)
source_kind (ObservationSourceKind | str)
stamped_revision_id (RevisionId)
captured_at (datetime | None)
member_nif (str | None)
source_metadata (Mapping[str, str] | None)
source_headers (tuple[ObservedHeaderFact, ...])
result_disposition (ResultDispositionProjection | None)
prior_domiciliation_election (PriorDomiciliationElectionProjection | None)
override (ObservationOverride | None)

load_observation_layers(modelo, period, *, member_nif=None)[source]¶

Return both stored layers of one coordinate; absent layers are None.

Return type:

ObservationLayers

Parameters:

modelo (str)
period (Period)
member_nif (str | None)

save(payload)[source]¶

Persist one validated observation into the layer its source kind selects.

Return type:: None
Parameters:: payload (ObservationEnvelopePayload)

promote_pending_local(modelo, period, *, member_nif=None, source_kind, source_metadata, captured_at)[source]¶

Prepare the writes that make the pending-local layer the official one.

Used when AEAT confirms a pending local filing: the pending-local envelope’s casilla values become the official layer under the official source_kind and source_metadata, and the pending-local layer is removed. Removal is expressed as upserts so the writes join the caller’s unit of work. Returns () when the coordinate has no pending-local layer.

Return type:

tuple[SecureObjectWrite, ...]

Parameters:

modelo (str)
period (Period)
member_nif (str | None)
source_kind (ObservationSourceKind)
source_metadata (Mapping[str, str])
captured_at (datetime)

clear_pending_local(modelo, period, *, member_nif=None, replacement_official=None)[source]¶

Prepare the writes that remove the pending-local layer of one coordinate.

The official layer is left untouched unless replacement_official is given, in which case it replaces the official layer in the same write. Removal is expressed as upserts so the writes join the caller’s unit of work. Returns () when the coordinate has no pending-local layer.

Return type:

tuple[SecureObjectWrite, ...]

Parameters: