aeat.domain.calculations.registry._live_parity module

Modelo-agnostic live parity oracle backend.

This module sits one level above _remote_state_guard and ties the existing fail-closed remote-state policy to a uniform contract for read-only verification of registry-rendered payloads against AEAT live surfaces.

Two-fold hardening underpins the design:

  1. Local hardening — already in place via the registry’s static conformance tests (record-design positions, casilla widths, byte roundtrips, formula closure).

  2. Live conformance — drive a synthetic, registry-rendered payload through an AEAT-published verification surface that must not modify remote state (open simulators, file validators like TGVI online, VIES IVA-ID checkers, pre-filing validators, AEAT integration test services). Every planned operation is pre-flighted against the cross-reference’s RemoteStateGuardPolicy before any HTTP or browser action runs; any policy-violating step is rejected before it leaves the process.

Each ModeloDefinition’s registry TOML declares which oracle a cross-reference is bound to via oracle_id; this module owns the runtime contract and the shared catalogue. Concrete oracle adapters live in sibling modules so the abstraction stays free of network code.

class OracleEnvironment(*values)[source]

Bases: StrEnum

Runtime environment classification for oracle catalogue entries.

PRODUCTION — the oracle is safe to call against the live AEAT surface. TEST_ENVIRONMENT — the oracle targets a sandboxed / integration-test surface only and must not be invoked from production callers. BOTH — the oracle is safe under either classification (e.g., public read-only surfaces that carry no production-state side-effect).

PRODUCTION
TEST_ENVIRONMENT
BOTH
class ParityFieldComparison(**data)[source]

Bases: _ParityModel

One field-level comparison between an expected and observed value.

Parameters:

  • name (str)

  • expected (str)

  • observed (str)

  • verdict (Literal['match', 'mismatch', 'unverifiable'])

name: str
expected: str
observed: str
verdict: Literal['match', 'mismatch', 'unverifiable']
class ParityResult(**data)[source]

Bases: _ParityModel

Outcome of running a synthetic payload through a live parity oracle.

The oracle layer never returns “filing succeeded” or any other side-effect confirmation; the only signal callers consume is whether AEAT’s response confirms the registry-rendered payload conforms (match), diverges (mismatch), is structurally unanswerable by the surface (unverifiable), or was refused before it left the process by the remote-state guard (blocked).

Parameters:

  • oracle_id (OracleId)

  • cross_reference_id (CrossReferenceId)

  • verdict (Literal['match', 'mismatch', 'unverifiable', 'blocked'])

  • narrative (str)

  • fields (tuple[ParityFieldComparison, ...])

  • raw_evidence_locator (str | None)

oracle_id: OracleId
cross_reference_id: CrossReferenceId
verdict: ParityVerdict
narrative: str
fields: tuple[ParityFieldComparison, ...]
raw_evidence_locator: str | None
class LiveParityOracle(*args, **kwargs)[source]

Bases: Protocol

Read-only AEAT verification surface contract.

Every concrete oracle must satisfy two invariants:

  • planned_operations enumerates every HTTP request, browser action, or local computation it will perform, in the order they will run. The oracle must not perform any unlisted operation. Callers iterate the planned list through assert_remote_operation_allowed() before any side-effecting code is reached.

  • verify_payload returns a ParityResult; it never raises on AEAT-side mismatch (mismatch is data, not an exception) and never returns "match" if any planned operation was skipped or rewritten.

property oracle_id: OracleId

Stable identifier this oracle registers under in the catalogue.

A modelo (an AEAT tax form) binds one of its live cross-references to an oracle by naming this id in registry TOML; the runtime resolves the binding by looking the same id up in the LiveParityCatalogue. The value must be non-empty and unique across the process-wide catalogue.

Returns:

The oracle’s typed catalogue key.

property surface_kind: Literal['file_validator', 'open_simulator', 'iva_id_check', 'pre_filing_validator', 'integration_test_service']

Kind of AEAT verification surface this oracle drives.

One of the OracleSurfaceKind literals (file_validator, open_simulator, iva_id_check, pre_filing_validator, integration_test_service). The boot-time binding audit cross-checks this value against the cross-reference’s own surface using the _COMPATIBLE_SURFACE_PAIRS allow-list, so a mismatch is reported rather than silently called.

Returns:

The surface classification as an OracleSurfaceKind literal.

planned_operations(payload, *, expected)[source]

Enumerate every remote step this oracle will perform, in order.

Returns the full, ordered set of HTTP requests, browser actions, or local computations the oracle intends to run for payload (the registry-rendered bytes to verify) and expected (the expected response values, keyed by label or casilla — a casilla being a numbered box on the form). The oracle must not perform any operation absent from this tuple; callers pre-flight each entry through the remote-state guard before any side-effecting code runs.

Parameters:

  • payload (bytes) – The synthetic, registry-rendered bytes to verify.

  • expected (Mapping[str, object]) – Expected response values the oracle will compare against.

Return type:

tuple[RemoteOperation, ...]

Returns:

The planned steps as a tuple of RemoteOperation.

verify_payload(policy, payload, *, expected)[source]

Run the payload through the live surface and report parity.

Pre-flights every planned operation against policy (the fail-closed remote-state guard for this cross-reference), then drives the surface and compares its response to expected. Never raises on an AEAT-side divergence — a mismatch is data, surfaced as the verdict — and never reports "match" if any planned operation was skipped or rewritten. A step the policy forbids yields a "blocked" verdict instead of a remote call.

Parameters:

  • policy (RemoteStateGuardPolicy) – The RemoteStateGuardPolicy gating remote operations.

  • payload (bytes) – The synthetic, registry-rendered bytes to verify.

  • expected (Mapping[str, object]) – Expected response values to compare against.

Return type:

ParityResult

Returns:

A ParityResult carrying the verdict and per-field comparisons.

class LiveParityCatalogue[source]

Bases: object

Registry of live parity oracles keyed by oracle_id.

Every modelo that wants live conformance verification declares an oracle_id in its registry cross-reference; the runtime looks the oracle up here. Catalogue registration is process-wide so adapters can self-register at import time.

Every registration declares an explicit environment classification so that adapters targeting AEAT pre-production / test-NIF surfaces cannot leak into production code paths. The lookup() call requires an environment context; oracles registered as "production" only are invisible to test-environment lookups and vice versa. "both" is reserved for adapters whose surface is provably safe under either classification (e.g., pure read-only public services that never touch AEAT NIF state under any environment).

register(oracle, *, environment)[source]

Register an oracle under an explicit environment classification.

Return type:

None

Parameters:
lookup(oracle_id, *, environment=OracleEnvironment.PRODUCTION)[source]

Return the registered oracle for the requested environment.

Raises when the oracle is unknown, or when its declared environment does not include the requested context. Production lookups never return test-environment-only oracles; test-environment lookups never return production-only oracles.

Return type:

LiveParityOracle

Returns:

The LiveParityOracle registered under oracle_id.

Parameters:
environment_of(oracle_id)[source]

Return the declared environment of a registered oracle.

Return type:

OracleEnvironment

Returns:

The OracleEnvironment declared for oracle_id.

Parameters:

oracle_id (OracleId)

is_registered(oracle_id)[source]

Report whether an oracle is registered under oracle_id.

A membership check that ignores environment classification: it returns True for any registered oracle regardless of whether it is production-only, test-environment-only, or both. Use lookup when the environment-visibility rules must be enforced.

Parameters:

oracle_id (TypeAliasType) – The catalogue key to test.

Return type:

bool

Returns:

True if an oracle is registered under oracle_id.

ids(*, environment=None)[source]

Return oracle ids, optionally filtered to those visible under environment.

Return type:

tuple[TypeAliasType, ...]

Parameters:

environment (OracleEnvironment | None)

build_planned_operations(oracle, payload, *, expected)[source]

Return the oracle’s planned operations as an immutable tuple.

A thin wrapper that lets callers obtain the plan without invoking the verification flow, useful for static-analysis tests that assert no oracle declares a forbidden operation under any policy.

Return type:

tuple[RemoteOperation, ...]

Returns:

Tuple of RemoteOperation steps planned by the oracle.

Parameters:
pre_flight_oracle_operations(oracle, policy, payload, *, expected)[source]

Pre-flight every planned RemoteOperation through the remote-state guard.

Returns the validated operation tuple if every operation is allowed. Raises RegistryValidationError on the first refused operation; the oracle must not be invoked when this raises, since the planned set contains a step the policy forbids.

Return type:

tuple[RemoteOperation, ...]

Parameters:
evaluate_planned_operations(oracle, policy, payload, *, expected)[source]

Evaluate planned RemoteOperation items against the policy without raising.

Returns either a blocked-verdict ParityResult (when any planned operation is rejected) or the validated RemoteOperation tuple itself. Callers that prefer an exception-free interface use this; tests and dry-runs use it to inspect blocked verdicts.

Return type:

ParityResult | tuple[RemoteOperation, ...]

Parameters:
assert_oracle_operations_allowed(oracle, policy, operations)[source]

Raise unless every operation in operations is allowed by policy.

Concrete oracle adapters call this at the entry of verify_payload so that the guard is the only gate before any side-effecting code, even when the oracle reuses an externally constructed operation list.

Return type:

None

Parameters:
class CrossReferenceApplicability(**data)[source]

Bases: _ParityModel

Profile-applicability outcome for one live cross-reference decision.

The model is the typed signal callers consume to decide whether to invoke a cross-reference at all. applicable=True with no predicates declared is the explicit unconditionally-applicable case.

Parameters:

  • cross_reference_id (CrossReferenceId)

  • applicable (bool)

  • matched_explanations (tuple[str, ...])

  • unmet_predicate_fields (tuple[str, ...])

cross_reference_id: CrossReferenceId
applicable: bool
matched_explanations: tuple[str, ...]
unmet_predicate_fields: tuple[str, ...]
evaluate_cross_reference_applicability(decision, profile_facts)[source]

Evaluate a cross-reference’s applicability against a profile.

Returns a typed CrossReferenceApplicability rather than a bare bool so callers (resolver, audit, live tests) consume a single shape. The function is profile-state evaluation only; it performs no network or catalogue lookup.

A decision with no applicability_predicates is unconditionally applicable. When predicates are declared, mode governs combination: all requires every predicate to match; any requires at least one match.

Return type:

CrossReferenceApplicability

Parameters:
resolve_cross_reference_oracle(*, cross_reference_id, oracle_id, catalogue, environment=OracleEnvironment.PRODUCTION, decision=None, profile_facts=None)[source]

Resolve a cross-reference’s bound oracle through the catalogue.

Re-frames every catalogue lookup error so the message names the cross-reference id alongside the oracle id. Callers consume the result inside the calculation engine; the resolver does not perform any network operation by itself.

The resolver requires the cross-reference to declare a binding. Cross- references with no oracle are not resolved here; their absence is a distinct case from “binding present but unresolvable” and the caller handles it before delegating.

Optional applicability gate: when both decision and profile_facts are supplied, evaluate_cross_reference_applicability runs first and the resolver raises a typed RegistryValidationError naming the unmet predicate fields if the binding is not applicable to the profile. Callers that don’t thread profile facts (for example, audit code) use catalogue-only resolution by omitting both arguments.

Return type:

LiveParityOracle

Returns:

The LiveParityOracle bound to the cross-reference.

Parameters:
audit_oracle_bindings(modelo, catalogue, *, environment=OracleEnvironment.PRODUCTION)[source]

Inspect every cross-reference binding in a modelo against the catalogue.

Returns a tuple of human-readable failure strings, one per cross- reference whose bound oracle id either is not registered in the catalogue or is registered under an incompatible environment. Cross- references with no binding are skipped silently.

A declared binding must resolve through the supplied catalogue. An empty catalogue is valid only when no cross-reference declares an oracle binding.

The function never raises and never performs any network operation. Failure aggregation is the caller’s job.

Parameters:
Return type:

tuple[str, ...]

class CrossReferenceApplicabilityDeclaracion(**data)[source]

Bases: _ParityModel

A registry-declared applicability shape for one cross-reference.

The model is a structural read of the registry data — the audit surface emits this so CI / dashboards can see which bindings are profile-gated without re-evaluating any predicate. Decoupled from CrossReferenceApplicability (the run-time evaluation result).

Parameters:

  • modelo_id (str)

  • revision_id (RevisionId)

  • cross_reference_id (CrossReferenceId)

  • applicability_condition_mode (Literal['all', 'any'])

  • predicate_fields (tuple[str, ...])

modelo_id: str
revision_id: RevisionId
cross_reference_id: CrossReferenceId
applicability_condition_mode: Literal['all', 'any']
predicate_fields: tuple[str, ...]
collect_applicability_declarations(modelos)[source]

Return CrossReferenceApplicabilityDeclaracion items for every cross-reference with predicates.

Pure registry-data introspection: never reads profile facts, never invokes the evaluator. Cross-references with no predicates are omitted (the unconditionally-applicable default). Order is (modelo_id, revision_id, cross_reference_id) for deterministic audit output.

Parameters:

modelos (Iterable[ModeloDefinition]) – Iterable of ModeloDefinition entries to scan.

Return type:

tuple[CrossReferenceApplicabilityDeclaracion, ...]

collect_orphan_oracle_ids(modelos, catalogue)[source]

Return catalogue oracle ids that no cross-reference binds.

A registered-but-unused oracle indicates one of:

  • the oracle was registered for a future binding still in flight,

  • a cross-reference’s oracle_id was renamed without updating the catalogue,

  • the binding was retired but the catalogue registration stayed.

The audit surfaces the set so CI / dashboards can flag drift. Order is the catalogue’s lexicographic order for deterministic output.

Parameters:
Return type:

tuple[TypeAliasType, ...]

class ReplayPayload(**data)[source]

Bases: _ParityModel

Typed envelope for a decoded replay JSON payload.

Every replay driver shares the same top-level JSON contract: an observed mapping of captured surface strings to string values (kept only as audit evidence, not as a comparison key surface) and an optional raw_evidence_locator that links back to the raw HTTP response artifact for audit trails.

Replay fixtures on disk are captured response artefacts and carry additional documented metadata that pre-dates the tightened schema:

  • scenario_id — fixture-author label that identifies the operator scenario the payload was captured against;

  • profile_overrides — per-fixture profile overrides used to drive the registry comparison;

  • expected — captured human-readable labels paired with their expected values, retained only for audit readability;

  • expected_by_casilla_id — registry-casilla-id-keyed expected values, used by the oracle’s matcher;

  • observed_by_casilla_id — registry-casilla-id-keyed observed values, used by the oracle’s matcher.

model_config inherits strict=True, frozen=True, extra="forbid" from _ParityModel. The documented fields above are typed explicitly; any other unknown key still raises at validation.

Parameters:
observed: Mapping[str, str]
raw_evidence_locator: str | None
scenario_id: str | None
profile_overrides: Mapping[str, str]
expected: Mapping[str, str]
expected_by_casilla_id: Mapping[CasillaId, str]
observed_by_casilla_id: Mapping[CasillaId, str]
decode_replay_json_payload(raw, *, surface_label)[source]

Decode a UTF-8 JSON replay payload into a typed ReplayPayload.

Shared by replay drivers: enforces UTF-8 encoding, valid JSON, a top-level object (dict) shape, and the ReplayPayload schema. surface_label is interpolated into the error messages so callers can identify their oracle in failures (e.g. "AEAT NIF-IVA replay").

Return type:

ReplayPayload

Parameters:
class BaseCheckerOracle(*, driver=None)[source]

Bases: Generic

Shared orchestrator for checker-style oracles.

Encapsulates the common verify_payload template used by per-key verdict checkers (NIF-IVA, GROI, and analogous future surfaces): guard pre-flight, driver presence branch, driver-error → unverifiable translation, per-key field comparison, overall verdict, and result packing. Subclasses provide the surface-specific bits: the planned operations builder, expected-value normaliser, observed-value lookup, per-field comparison, and human narrative label.

Per-domain typed observation models (NIF/IVA vs GROI) stay in the concrete adapter modules; this base composes them generically.

Parameters:

driver (_CheckerDriver[CheckerObservation] | None)

surface_label: str
abstract property oracle_id: OracleId

Stable catalogue identifier for this checker oracle.

Abstract: each concrete per-key checker (NIF-IVA, GROI, and analogous surfaces) supplies the id its modelo cross-reference binds to in registry TOML. Must be non-empty and unique within the catalogue.

Returns:

The oracle’s typed catalogue key.

abstract property surface_kind: Literal['file_validator', 'open_simulator', 'iva_id_check', 'pre_filing_validator', 'integration_test_service']

Kind of AEAT verification surface this checker oracle drives.

Abstract: the concrete adapter returns the OracleSurfaceKind literal matching its surface, which the binding audit cross-checks against the cross-reference’s declared surface via the _COMPATIBLE_SURFACE_PAIRS allow-list.

Returns:

The surface classification as an OracleSurfaceKind literal.

abstractmethod planned_operations(payload, *, expected)[source]

Enumerate the remote steps this checker oracle will perform.

Abstract: the concrete adapter builds the ordered tuple of operations for payload (the registry-rendered bytes) and expected (the expected per-key values). The shared verify_payload template pre-flights this set through the remote-state guard before any step runs, so the returned tuple must list every operation the oracle intends to perform.

Parameters:

  • payload (bytes) – The synthetic, registry-rendered bytes to verify.

  • expected (Mapping[str, object]) – Expected response values the oracle will compare against.

Return type:

tuple[RemoteOperation, ...]

Returns:

The planned steps as a tuple of RemoteOperation.

verify_payload(policy, payload, *, expected)[source]

Run the shared checker template and report parity.

The concrete subclass supplies the surface-specific pieces; this template sequences them: pre-flight every planned operation against policy (returning a "blocked" verdict on refusal), translate a missing or erroring driver into an "unverifiable" verdict, compare expected against observed values per key, and pack the overall verdict into a ParityResult. The verdict is "match" only when at least one field compared and every field matched. Never raises on an AEAT-side divergence.

Parameters:

  • policy (RemoteStateGuardPolicy) – The RemoteStateGuardPolicy gating remote operations.

  • payload (bytes) – The synthetic, registry-rendered bytes to verify.

  • expected (Mapping[str, object]) – Expected per-key values to compare against.

Return type:

ParityResult

Returns:

A ParityResult carrying the verdict and per-field comparisons.

audit_registry_oracle_bindings(modelos, catalogue, *, environment=OracleEnvironment.PRODUCTION)[source]

Aggregate audit_oracle_bindings over an iterable of modelos.

Application bootstrap calls this once per startup to surface every binding-vs-catalogue mismatch in a single report alongside the registry-validator’s own failures. The function preserves the order of the input iterable so the report is deterministic.

Parameters:
Return type:

tuple[str, ...]