"""Filing-record actions for modelo calculation revisions.
:func:`~cadrumo.application.modelo.filing_actions.file_modelo_revision` promotes a verified
:class:`CalculationRevision` into a current
:class:`ModeloRecord` after the
:class:`WorkflowEngine` preflight gate passes. Filing
transitions and audit entries are persisted through the application-owned
bucket-event capability shared by the modelo revision services.
The action records the operator's local/internal filing state only. It never
submits to AEAT, never marks AEAT acceptance, and never fabricates official
external evidence. A successful transition sets the target revision to
``PRESENTADO``, creates a ``VIGENTE``
:class:`ModeloRecord` with ``aeat_accepted=False``, and
delegates cross-period carry projection to
:func:`~cadrumo.application.modelo.revision_persistence.persist_filed_revision`,
which stamps locally-filed observations as non-official ``app_filing`` evidence.
See Also:
:func:`~cadrumo.application.modelo.external_import_actions.import_external_filing_evidence`:
Separate AEAT-attested import path that creates
:class:`ExternalEvidence` baselines; this local
filing action deliberately does not.
:func:`~cadrumo.application.modelo.revision_persistence.persist_filed_revision`:
Persists the filing catalogue, revision state, work-unit pointers,
bucket events, participation index rows, and optional carry observation.
:func:`~cadrumo.application.modelo.filed_revision_observation.persist_filed_revision_observation`:
Projects filed casillas into non-official cross-period observations.
:func:`~cadrumo.application.modelo.result_disposition_resolution.resolve_modelo_result_disposition`:
Resolves the shared Modelo 303 refund/carry disposition before the file
transition persists.
``cadrumo.application.modelo.verification_actions._require_cross_period_clean_state``:
Rechecks cross-period dependencies before local filing state is written.
"""
from __future__ import annotations
from collections.abc import Iterable
from datetime import date, datetime
from pathlib import Path
from typing import TYPE_CHECKING
from ...core.config import Settings
from ...core.identity.hex_ids import CalculationRevisionId
from ...core.payment_election import PaymentElection
from ...core.prior_domiciliation_election import PriorDomiciliationElection
from ...core.refund_election import RefundElection
from ...core.result_disposition import ResultDisposition
from ...core.time.clock import now as _utc_now
from ...domain.calculations.registry.applicability import derive_taxpayer_files_economic_activity
from ...domain.calculations.registry.applicability_modelo202 import derive_modelo_202_modality
from ...domain.calculations.registry.authority import PinnedAuthorityOperation
from ...domain.deadlines.models import TaxpayerProfile
from ...domain.modelos.calculation_revision import CalculationRevision, CalculationRevisionState
from ...domain.modelos.codes import ModeloCode
from ...domain.modelos.errors import ModeloError
from ...domain.modelos.filing_record import ModeloRecord, ModeloRecordCatalogue, ModeloRecordStatus
from ...domain.modelos.verification_report import VerificationReport
from ...domain.modelos.work_unit import WorkUnit
from ..calculations.cross_period_models import CrossPeriodExpectedMemberSet
from ..calculations.m303_regimen_simplificado_annual_summary import (
validate_m303_regimen_simplificado_annual_summary_target_revision,
)
from ..calculations.verification_report_gate import require_verification_report_coordinates_current
from ..workflow.engine import WorkflowEngine
from ._ledger_evidence_gate import raise_if_deductible_iva_evidence_missing
from ._required_binding_gate import (
require_persisted_revision_required_bindings_resolved as _require_persisted_required_bindings_resolved,
)
from .action_errors import (
CalculationRevisionNotFoundError,
CalculationRevisionStateError,
ModeloPreconditionErrorMixin,
ModeloRecordNotFoundError,
VerificationReportNotFoundError,
WorkUnitNotFoundError,
)
from .calculation_revision_gate import require_calculation_revision_coordinates_current
from .filing_action_ports import FilingActionPorts
from .iva_wallet_gate import (
require_persisted_iva_compensation_decision_matches_revision as _require_iva_compensation_revision_match,
)
from .m303_regimen_simplificado_scope import (
m303_regimen_simplificado_annual_summary_applies_to_profile,
taxpayer_profile_for_work,
)
from .preconditions import build_modelo_work_file_unverified_revision_failure
from .prior_domiciliation import resolveprior_domiciliation_election
from .result_disposition_resolution import resolve_modelo_result_disposition
from .revision_persistence import persist_filed_revision, require_filing_instance_evidence_for_work_unit
from .verification_cross_period import cross_period_expected_member_sets_from_profile, require_cross_period_clean_state
from .work_lifecycle import RevisionParentOperation, require_revision_parent_active
from .workflow_gate import build_revision_workflow_engine as _build_revision_workflow_engine
from .workflow_gate import run_revision_workflow_gate as _run_revision_workflow_gate
if TYPE_CHECKING:
from ..auth.certificate_secret_backend import CertificateSecretBackendFactory
from ..auth.operator_scope_ports import OperatorScopePorts
from .work_profile import ModeloWorkProfile
[docs]
class ModeloFilingEvidenceMissingError(ModeloPreconditionErrorMixin, ModeloError):
"""Raised when internal filing would seal deductible IVA without evidence."""
def _existing_vigente_filing_record(
catalogue: ModeloRecordCatalogue,
calculation_revision_id: CalculationRevisionId,
) -> ModeloRecord | None:
"""Return the current (VIGENTE) filing record for a filed revision, if any.
Used by the idempotent re-file no-op: a revision already in ``PRESENTADO``
state is the current filed answer, so its VIGENTE
:class:`ModeloRecord` is returned unchanged. A
``PRESENTADO`` revision must have exactly one VIGENTE record; ``None``
signals an inconsistent state the caller refuses rather than masking.
"""
for record in catalogue.records.values():
if record.calculation_revision_id == calculation_revision_id and record.status is ModeloRecordStatus.VIGENTE:
return record
return None
[docs]
def file_modelo_revision(
calculation_revision_id: CalculationRevisionId,
*,
certificate_secret_backend_factory: CertificateSecretBackendFactory,
operator_scope_ports: OperatorScopePorts,
ports: FilingActionPorts,
actor: str,
workflow_profile: TaxpayerProfile,
operation: PinnedAuthorityOperation,
notes: str | None = None,
refund_election: RefundElection = RefundElection.COMPENSAR,
payment_election: PaymentElection = PaymentElection.INGRESO,
prior_domiciliation_election: object = PriorDomiciliationElection.KEEP,
cross_period_expected_member_sets: Iterable[CrossPeriodExpectedMemberSet] = (),
workflow_engine: WorkflowEngine | None = None,
workflow_runs_dir: Path | None = None,
settings: Settings | None = None,
clock: datetime | None = None,
profile: ModeloWorkProfile | None = None,
) -> ModeloRecord:
"""Mark a verified-complete revision as the current internal filed answer.
This is the application service behind ``aeat app modelo work file``. It is
a local state transition, not an AEAT presentation: the resulting
:class:`ModeloRecord` has ``aeat_accepted=False`` and no
external evidence.
Preconditions and state changes:
1. Verify the revision is in ``VERIFICADO_COMPLETO`` state.
2. Recheck profile readiness, persisted Modelo 303 IVA-wallet decision
compatibility, and cross-period clean state.
3. Run the :class:`WorkflowEngine` gate for the
revision's modelo and period.
4. Resolve the Modelo 303 result disposition from
:class:`RefundElection`, :class:`PaymentElection`, and
:class:`TaxpayerProfile`.
5. If a prior current filing exists, mark its
:class:`ModeloRecord` as ``SUPERSEDIDO`` and its
prior :class:`CalculationRevision` as
``PRESENTADO_SUPERSEDIDO``.
6. Create the new filing record with status ``VIGENTE`` and transition the
target calculation revision from ``VERIFICADO_COMPLETO`` to
``PRESENTADO``.
7. Advance the work unit's ``filed_calculation_revision_id`` and
``current_filing_record_id`` pointers, emit ``MODELO_FILED``/
``MODELO_FILED_SUPERSEDED`` bucket events, and persist any local
``app_filing`` carry observation.
Args:
calculation_revision_id: The id of the verified-complete revision
to file.
certificate_secret_backend_factory: Factory for the encrypted
certificate-secret capability used by workflow authentication.
operator_scope_ports: Operator-scope capabilities used by workflow
authentication.
actor: Operator identifier recorded in the filing record and audit
trail.
workflow_profile: The :class:`TaxpayerProfile`
used to evaluate :class:`WorkflowEngine`
gate conditions and cross-period clean-state applicability.
notes: Optional operator-supplied filing notes.
refund_election: The operator's per-filing Modelo 303 negative-result
disposition election. Defaults to ``COMPENSAR`` (carry the credit
forward). ``DEVOLVER`` requests the credit back as a refund and is
honoured only when the period is a lawful refund period (the year's
last filing period for a non-REDEME taxpayer; every period for REDEME).
An out-of-window ``DEVOLVER`` is refused.
payment_election: The operator's positive-result settlement election.
``INGRESO`` is the default, ``DOMICILIACION`` is available only for
Modelo 303, and ``CUENTA_CORRIENTE`` remains explicitly refused
until its AEAT capability is grounded.
prior_domiciliation_election: Whether to preserve the prior direct debit
or request its cancellation/modification. The latter is accepted
only for an M303 rectificativa with the official baseline-U proof.
ports: Complete application-owned filing capability bundle. It supplies
the work-unit, calculation, filing, verification, observation,
IVA-wallet, bucket-event, and workflow-run authorities for the
active profile bucket.
operation: Caller-owned generation-pinned indexed authority operation.
cross_period_expected_member_sets: Optional expected grupo member
rosters used by the cross-period clean-state proof.
workflow_engine: Optional workflow engine override for the preflight
gate.
workflow_runs_dir: Optional workflow runs directory override.
settings: Optional settings override.
clock: Optional UTC timestamp override.
profile: The work profile the calling command already loaded for the
revision's bucket. When omitted, this entry loads it once; every
profile read below uses that record.
Returns:
The newly created local :class:`ModeloRecord` in
``VIGENTE`` status.
Raises:
CalculationRevisionNotFoundError: When the revision id is
absent.
CalculationRevisionStateError: When the revision is not in
``VERIFICADO_COMPLETO`` state.
WorkUnitNotFoundError: When the revision's parent work
unit cannot be loaded.
See Also:
:func:`~cadrumo.application.modelo.revision_persistence.persist_filed_revision`:
Performs the repository writes once all gates pass.
:func:`~cadrumo.application.modelo.external_import_actions.import_external_filing_evidence`:
Creates official-evidence baselines for imported filings; use that
path when a :class:`ExternalEvidence` reference
must be carried.
:func:`~cadrumo.application.modelo.filed_revision_observation.persist_filed_revision_observation`:
Saves the non-official ``app_filing`` observation used by later
``previous_filing`` calculations.
:func:`~cadrumo.application.modelo.export.export_modelo_revision`:
Sibling local finish line that writes the fichero-BOE artefact
without requiring this internal file marker.
"""
wu_repo = ports.work_unit_repository
cr_repo = ports.calculation_repository
fr_repo = ports.filing_repository
obs_repo = ports.observation_repository
bv_repo = ports.bucket_event_repository
if not isinstance(prior_domiciliation_election, PriorDomiciliationElection):
from .action_errors import ModeloPriorDomiciliationElectionRefusedError
raise ModeloPriorDomiciliationElectionRefusedError(
translated_message="errors.refused.refused_modeloprior_domiciliation_election",
context={"received_type": type(prior_domiciliation_election).__name__},
)
run_repo = ports.workflow_run_repository
revisions = cr_repo.load()
target = revisions.get(calculation_revision_id)
if target is None:
raise CalculationRevisionNotFoundError(
translated_message="application.modelo.errors.calculation_revision_not_found",
context={"calculation_revision_id": calculation_revision_id},
)
require_calculation_revision_coordinates_current(target, operation=operation)
work_units = wu_repo.load()
work_unit = work_units.get(target.work_unit_id)
if work_unit is None:
raise WorkUnitNotFoundError(
translated_message="application.modelo.errors.work_unit_not_found",
context={
"calculation_revision_id": calculation_revision_id,
"work_unit_id": target.work_unit_id,
},
)
require_filing_instance_evidence_for_work_unit(work_unit=work_unit, revision=target)
require_revision_parent_active(
work_unit=work_unit,
calculation_revision_id=calculation_revision_id,
operation=RevisionParentOperation.FILE,
)
if profile is None:
from .profile_readiness_gate import load_modelo_work_profile
profile = load_modelo_work_profile(
bucket_id=work_unit.bucket_id,
profile_decode_context=operation.profile_decode_context(),
)
validate_m303_regimen_simplificado_annual_summary_target_revision(
target_work_unit=work_unit,
target_revision=target,
work_unit_repository=wu_repo,
calculation_repository=cr_repo,
filing_repository=fr_repo,
regimen_simplificado_applies=m303_regimen_simplificado_annual_summary_applies_to_profile(
taxpayer_profile_for_work(profile),
),
operation=operation,
)
if target.state is CalculationRevisionState.PRESENTADO:
# Idempotent re-file: this revision is already the current filed answer.
# A retry of a completed single-subject file returns the existing VIGENTE
# filing record as a clean no-op - no new filing record, no duplicate
# MODELO_FILED lifecycle event, and (filing is a local transition, never
# an AEAT submission per sensitive-financial-data-secure-storage-only) no write/submit path is
# touched. Mirrors the verify-report content-pinned idempotency. The CLI
# surfaces the no-op as an info Notice. A PRESENTADO revision with no
# VIGENTE record is an inconsistent state, so it falls through to the
# hard refusal below rather than fabricating a record.
existing = _existing_vigente_filing_record(fr_repo.load(), calculation_revision_id)
if existing is not None:
# An idempotent retry remains subject to the same fail-closed typed
# election boundary: an unproven ``X`` request cannot hide behind
# a prior local filing no-op.
resolveprior_domiciliation_election(
election=prior_domiciliation_election,
work_unit=work_unit,
revision=target,
filing_repository=fr_repo,
observation_repository=obs_repo,
operation=operation,
)
return existing
if target.state is not CalculationRevisionState.VERIFICADO_COMPLETO:
raise CalculationRevisionStateError(
translated_message="errors.error.error_modelo_calculation_revision_state",
context={"calculation_revision_id": calculation_revision_id, "state": target.state.value},
precondition_failure=build_modelo_work_file_unverified_revision_failure(
calculation_revision_id=calculation_revision_id,
state=target.state.value,
work_unit=work_unit,
),
)
_require_filing_preconditions(
work_unit=work_unit,
target=target,
workflow_profile=workflow_profile,
ports=ports,
cross_period_expected_member_sets=cross_period_expected_member_sets,
operation=operation,
profile=profile,
)
now = clock or _utc_now()
gate_engine = workflow_engine or _build_revision_workflow_engine(
certificate_secret_backend_factory=certificate_secret_backend_factory,
operator_scope_ports=operator_scope_ports,
revision=target,
work_unit=work_unit,
profile=workflow_profile,
actor=actor.strip(),
clock=now,
settings=settings,
draft_review_ports=ports.draft_review_ports,
workflow_gate_ports=ports.workflow_gate_ports,
operation=operation,
)
_run_revision_workflow_gate(
engine=gate_engine,
profile=workflow_profile,
work_unit=work_unit,
today=now.date(),
runs_dir=workflow_runs_dir,
run_repository=run_repo,
)
result_disposition = _filed_revision_result_disposition(
work_unit=work_unit,
target=target,
workflow_profile=workflow_profile,
refund_election=refund_election,
payment_election=payment_election,
)
prior_domiciliation_provenance = resolveprior_domiciliation_election(
election=prior_domiciliation_election,
work_unit=work_unit,
revision=target,
filing_repository=fr_repo,
observation_repository=obs_repo,
operation=operation,
)
return persist_filed_revision(
target=target,
work_unit=work_unit,
work_units=work_units,
notes=notes,
actor=actor,
now=now,
calculation_repository=cr_repo,
filing_repository=fr_repo,
work_unit_repository=wu_repo,
bucket_event_repository=bv_repo,
calculation_observation_repository=obs_repo,
participation_index_repository=ports.participation_index_repository,
prorrata_register_repository=ports.prorrata_register_repository,
operation=operation,
iva_compensation_history_repository=ports.iva_compensation_history_repository,
result_disposition=result_disposition,
prior_domiciliation_election=prior_domiciliation_provenance,
taxpayer_nif=workflow_profile.tax_id,
justificante_repository=ports.justificante_repository,
)
def _filed_revision_result_disposition(
*,
work_unit: WorkUnit,
target: CalculationRevision,
workflow_profile: TaxpayerProfile,
refund_election: RefundElection,
payment_election: PaymentElection,
) -> ResultDisposition:
"""Resolve once at the filing boundary for both export and carry evidence."""
return resolve_modelo_result_disposition(
work_unit=work_unit,
revision=target,
workflow_profile=workflow_profile,
period=work_unit.period,
refund_election=refund_election,
payment_election=payment_election,
)
def _require_filing_preconditions(
*,
work_unit: WorkUnit,
target: CalculationRevision,
workflow_profile: TaxpayerProfile,
ports: FilingActionPorts,
cross_period_expected_member_sets: Iterable[CrossPeriodExpectedMemberSet],
operation: PinnedAuthorityOperation,
profile: ModeloWorkProfile | None,
) -> None:
from .profile_readiness_gate import require_profile_ready_for_work_unit
raise_if_deductible_iva_evidence_missing(
target,
error_type=ModeloFilingEvidenceMissingError,
)
require_profile_ready_for_work_unit(
work_unit,
profile_decode_context=operation.profile_decode_context(),
operation=operation,
profile=profile,
)
_require_persisted_required_bindings_resolved(
work_unit=work_unit, revision=target, action="file", operation=operation
)
iva_compensation_decision = _require_iva_compensation_revision_match(
work_unit,
target,
repository=ports.iva_compensation_decision_repository,
subject_leaf_key="modelo.work.file",
)
require_cross_period_clean_state(
work_unit,
observation_repository=ports.observation_repository,
filing_repository=ports.filing_repository,
calculation_repository=ports.calculation_repository,
verification_repository=ports.verification_repository,
justificante_repository=ports.justificante_repository,
iva_compensation_decision=iva_compensation_decision,
expected_member_sets=cross_period_expected_member_sets_from_profile(
workflow_profile,
cross_period_expected_member_sets,
),
taxpayer_tax_id=workflow_profile.tax_id,
activity_start_date=workflow_profile.activity_start_date,
modelo_202_modality=derive_modelo_202_modality(
workflow_profile,
effective_date=date(work_unit.filing_year, 12, 31),
).modality,
taxpayer_files_economic_activity=derive_taxpayer_files_economic_activity(workflow_profile),
workflow_profile=workflow_profile,
target_revision=target,
subject_leaf_key="modelo.work.file",
)
[docs]
def list_filing_records(
*,
ports: FilingActionPorts,
bucket_id: str | None = None,
modelo: str | ModeloCode | None = None,
include_superseded: bool = False,
) -> tuple[ModeloRecord, ...]:
"""List :class:`ModeloRecord` rows, optionally filtered to a bucket and modelo.
Superseded records are excluded unless ``include_superseded``
is true. Results are sorted by ``(bucket_id, filing_year,
modelo, period, filed_at)``.
"""
catalogue = ports.filing_repository.load()
modelo_code = ModeloCode(str(modelo)) if modelo is not None else None
records = tuple(
record
for record in catalogue.records.values()
if (bucket_id is None or record.bucket_id == bucket_id)
and (modelo_code is None or record.modelo == modelo_code)
and (include_superseded or record.status is ModeloRecordStatus.VIGENTE)
)
return tuple(
sorted(
records,
key=lambda r: (r.bucket_id, r.filing_year, str(r.modelo), r.period.registry_token, r.filed_at),
),
)
[docs]
def get_filing_record(
filing_record_id: str,
*,
ports: FilingActionPorts,
) -> ModeloRecord:
"""Return the :class:`ModeloRecord` for the given id, or raise."""
catalogue = ports.filing_repository.load()
record = catalogue.get(filing_record_id)
if record is None:
raise ModeloRecordNotFoundError(
translated_message="application.modelo.errors.filing_record_not_found",
context={"filing_record_id": filing_record_id},
)
return record
[docs]
def list_verification_reports(
*,
ports: FilingActionPorts,
calculation_revision_id: CalculationRevisionId | None = None,
operation: PinnedAuthorityOperation,
) -> tuple[VerificationReport, ...]:
"""List :class:`VerificationReport` records.
Optionally filtered to one
:class:`CalculationRevision`. The
:class:`VerificationReportCatalogueRepositoryProtocol`
supplies the persisted report catalogue. Results are sorted by
``(calculation_revision_id, run_at)``.
"""
catalogue = require_verification_report_coordinates_current(
ports.verification_repository.load(),
operation=operation,
)
reports = tuple(
r
for r in catalogue.reports.values()
if calculation_revision_id is None or r.calculation_revision_id == calculation_revision_id
)
return tuple(sorted(reports, key=lambda r: (r.calculation_revision_id, r.run_at)))
[docs]
def get_verification_report(
verification_report_id: str,
*,
ports: FilingActionPorts,
operation: PinnedAuthorityOperation,
) -> VerificationReport:
"""Return one :class:`VerificationReport` by id, or raise.
The optional
:class:`VerificationReportCatalogueRepositoryProtocol`
supplies the persisted report catalogue for tests or alternate storage
boundaries.
"""
catalogue = require_verification_report_coordinates_current(
ports.verification_repository.load(),
operation=operation,
)
report = catalogue.get(verification_report_id)
if report is None:
raise VerificationReportNotFoundError(
translated_message="application.modelo.errors.verification_report_not_found",
context={"verification_report_id": verification_report_id},
)
return report