"""The guarded application-owned edit executor for the Modelo Edit Contract V1.
Behind :mod:`cadrumo.application.modelo`. Only the enrolled operation executor
may invoke this boundary (D6). Immediately before effect it revision-loads the
work and calculation catalogues and rechecks every baseline coordinate through
:func:`~._edit_services.reconfirm_modelo_edit_baseline`; any disagreement
refuses with ``stale_edit_baseline``, writes nothing, and settles the domain
effect as ``NONE``. It never silently rebases, merges, or promotes a green
preflight into authority.
V1 scope: the ``CALCULATE`` mutation family with ``SET_TYPED_VALUE`` and
``CLEAR_DECLARED_VALUE`` scalar intents, because those are the only intents
the shared calculation boundary
(:func:`~._calculation_actions.calculate_modelo_revision_from_bucket_aggregation_with_diagnostics`)
has an input shape for today -- a set value and an explicit clear, the latter
via its own ``cleared_casilla_ids`` identity axis so a cleared casilla stays
provably distinguishable from one never declared. Every other syntactically
admitted intent -- including every binding-override intent, which the
calculate boundary has no ``binding_overrides``-clearing input shape for yet
-- refuses with a typed, enumerated
:class:`~._edit_models.ModeloEditUnsupportedIntentReason` naming the specific
intent kind, not a generic bucket -- extending the shared calculation
boundary's own input surface further, or materialising repeatable rows inside
this executor, are out of this Step's scope (they would either widen the live
calculate path's blast radius or re-implement the write path it already
owns).
"""
from __future__ import annotations
from datetime import datetime
from decimal import Decimal
from typing import TYPE_CHECKING
from ...core.casilla_id import CasillaId
from ...core.hashing import content_hash_hex
from ...core.secure_object_write import SecureObjectWrite
from ...domain.modelos.calculation_revision import CalculationRevisionCatalogue
from ...domain.modelos.errors import ModeloError
from ...domain.modelos.row_models import ModeloDetailRow
from .calculation_action_ports import CalculationActionPorts
from .calculation_actions import calculate_modelo_revision_from_bucket_aggregation_with_diagnostics
from .calculation_revision_gate import require_calculation_revision_coordinates_current
from .edit_contract import ModeloEditMutationFamily, ModeloEditMutationResultReceiptV1
from .edit_models import (
ModeloDetailRowEditIntentV1,
ModeloEditAddressV1,
ModeloEditApplyRequestV1,
ModeloEditBaselineV1,
ModeloEditBindingIntentKind,
ModeloEditDetailRowIntentKind,
ModeloEditDomainRefusalV1,
ModeloEditExecutionNoEffectV1,
ModeloEditExecutionResultV1,
ModeloEditExecutionUpdatedV1,
ModeloEditRefusalCode,
ModeloEditRowIntentKind,
ModeloEditScalarIntentKind,
ModeloEditSubmissionV1,
ModeloEditUnsupportedIntentReason,
ModeloEditUnsupportedIntentRefusalV1,
)
from .edit_receipt_ports import ModeloEditReceiptRepositoryPort
from .edit_services import RESPONSIBLE_OWNER as _RESPONSIBLE_OWNER
from .edit_services import (
detail_row_natural_key,
reconfirm_modelo_edit_baseline,
validate_scalar_intent,
writable_scalar_entry,
)
if TYPE_CHECKING:
from ...domain.calculations.registry.authority import PinnedAuthorityOperation
_UNSUPPORTED_RECONSIDERATION = "resubmit without this intent once its follow-on Step lands, or split the submission"
_ROW_UNSUPPORTED_REASON: dict[ModeloEditRowIntentKind, ModeloEditUnsupportedIntentReason] = {
ModeloEditRowIntentKind.ADD_ROW: ModeloEditUnsupportedIntentReason.ADD_ROW_NOT_YET_WIRED,
ModeloEditRowIntentKind.UPDATE_ROW: ModeloEditUnsupportedIntentReason.UPDATE_ROW_NOT_YET_WIRED,
ModeloEditRowIntentKind.DELETE_ROW: ModeloEditUnsupportedIntentReason.DELETE_ROW_NOT_YET_WIRED,
ModeloEditRowIntentKind.MOVE_ROW: ModeloEditUnsupportedIntentReason.MOVE_ROW_NOT_YET_WIRED,
}
_BINDING_UNSUPPORTED_REASON: dict[ModeloEditBindingIntentKind, ModeloEditUnsupportedIntentReason] = {
ModeloEditBindingIntentKind.SET_OVERRIDE_VALUE: ModeloEditUnsupportedIntentReason.SET_OVERRIDE_VALUE_NOT_YET_WIRED,
ModeloEditBindingIntentKind.REMOVE_OVERRIDE: ModeloEditUnsupportedIntentReason.REMOVE_OVERRIDE_NOT_YET_WIRED,
}
_NUMERIC_DATA_TYPES = frozenset({"decimal", "money", "integer", "ratio", "year"})
def _unsupported_intent_refusal(
reason: ModeloEditUnsupportedIntentReason, *, address: ModeloEditAddressV1 | None = None
) -> ModeloEditExecutionNoEffectV1:
return ModeloEditExecutionNoEffectV1(
refusal=ModeloEditUnsupportedIntentRefusalV1(
address=address,
reason=reason,
responsible_owner=_RESPONSIBLE_OWNER,
reconsideration_condition=_UNSUPPORTED_RECONSIDERATION,
),
)
def _reachable_scalar_inputs(
submission: ModeloEditSubmissionV1,
) -> tuple[dict[str, Decimal], dict[str, str], tuple[CasillaId, ...]] | ModeloEditExecutionNoEffectV1:
"""Translate SET_TYPED_VALUE and CLEAR_DECLARED_VALUE scalar intents into the calculate boundary's own input shape.
A ``CLEAR_DECLARED_VALUE`` intent contributes no entry to either casilla
input dict; it contributes its address to the returned cleared-casilla
tuple instead, which the caller threads into the calculate boundary's own
``cleared_casilla_ids`` identity axis so the clear is provably
distinguishable from a casilla simply never supplied.
Every other intent kind this V1 executor cannot yet reach refuses here,
before any catalogue read, naming the exact unreachable intent kind
(never a single generic "unsupported" bucket).
"""
if submission.row_intents:
first_row = submission.row_intents[0]
return _unsupported_intent_refusal(_ROW_UNSUPPORTED_REASON[first_row.kind], address=first_row.address)
if submission.binding_intents:
first_binding = submission.binding_intents[0]
return _unsupported_intent_refusal(
_BINDING_UNSUPPORTED_REASON[first_binding.kind], address=first_binding.address
)
baseline = submission.baseline
casilla_inputs: dict[str, Decimal] = {}
text_casilla_inputs: dict[str, str] = {}
cleared_casilla_ids: list[CasillaId] = []
for intent in submission.scalar_intents:
if intent.kind is ModeloEditScalarIntentKind.CLEAR_DECLARED_VALUE:
cleared_casilla_ids.append(intent.address.casilla_id)
continue
entry = writable_scalar_entry(baseline, intent.address.casilla_id)
data_type = entry.data_type if entry is not None else "text"
if data_type in _NUMERIC_DATA_TYPES:
casilla_inputs[intent.address.casilla_id] = Decimal(str(intent.value))
else:
text_casilla_inputs[intent.address.casilla_id] = str(intent.value)
return casilla_inputs, text_casilla_inputs, tuple(cleared_casilla_ids)
def _detail_row_natural_key_refusal(address: ModeloEditAddressV1) -> ModeloEditExecutionNoEffectV1:
return ModeloEditExecutionNoEffectV1(
refusal=ModeloEditDomainRefusalV1(
code=ModeloEditRefusalCode.DISALLOWED_INTENT,
address=address,
responsible_owner=_RESPONSIBLE_OWNER,
reconsideration_condition="address only a detail row currently declared under this natural key",
)
)
def _detail_rows_by_kind(
current_detail_rows: tuple[ModeloDetailRow, ...],
) -> dict[str, list[ModeloDetailRow]]:
"""Group the current rows without changing their within-kind order."""
by_kind: dict[str, list[ModeloDetailRow]] = {}
for row in current_detail_rows:
by_kind.setdefault(row.row_type, []).append(row)
return by_kind
def _apply_detail_row_intent(
by_kind: dict[str, list[ModeloDetailRow]],
intent: ModeloDetailRowEditIntentV1,
) -> ModeloEditExecutionNoEffectV1 | None:
"""Apply one natural-key row intent, returning its typed refusal if invalid."""
kind = intent.address.detail_row_kind
rows = by_kind.setdefault(kind, [])
keys = [detail_row_natural_key(row) for row in rows]
if intent.kind is ModeloEditDetailRowIntentKind.ADD_ROW:
if intent.row is None:
return _detail_row_natural_key_refusal(intent.address)
rows.append(intent.row)
elif intent.kind is ModeloEditDetailRowIntentKind.UPDATE_ROW:
if intent.row is None:
return _detail_row_natural_key_refusal(intent.address)
if intent.address.natural_key not in keys:
return _detail_row_natural_key_refusal(intent.address)
rows[keys.index(intent.address.natural_key)] = intent.row
else:
if intent.address.natural_key not in keys:
return _detail_row_natural_key_refusal(intent.address)
rows.pop(keys.index(intent.address.natural_key))
return None
def _ordered_detail_rows(by_kind: dict[str, list[ModeloDetailRow]]) -> tuple[ModeloDetailRow, ...]:
"""Flatten grouped rows by row type while retaining each group's order."""
result: list[ModeloDetailRow] = []
for kind in sorted(by_kind):
result.extend(by_kind[kind])
return tuple(result)
def _reconstruct_detail_rows(
*,
current_detail_rows: tuple[ModeloDetailRow, ...],
detail_row_intents: tuple[ModeloDetailRowEditIntentV1, ...],
) -> tuple[ModeloDetailRow, ...] | ModeloEditExecutionNoEffectV1:
"""Rebuild the complete ``detail_rows`` tuple the memoryless calculate boundary requires.
Rows are grouped by ``row_type`` (a distinct fichero record per kind) and
addressed by their own natural key, never position or a minted identity
-- mirroring ``RetencionObservationRepository.replace_observations``'s
established whole-set-replacement convention. A row absent from the
result is simply not declared; there is no separate "explicitly deleted"
axis, because a row (unlike a scalar) has no ambiguous middle state
between "declared" and "absent".
No intent reorders an existing row: the calculation revision's content
address is explicitly order-blind
(``_calculation_revision._canonical_detail_rows`` sorts by
``(row_type, nif-like)`` so insertion order never affects the id), so a
pure reorder would compute the SAME id as the existing revision and be
silently absorbed by the guarded duplicate-result branch rather than
actually persist. See :class:`~._edit_models.ModeloEditDetailRowIntentKind`.
"""
by_kind = _detail_rows_by_kind(current_detail_rows)
for intent in detail_row_intents:
refusal = _apply_detail_row_intent(by_kind, intent)
if refusal is not None:
return refusal
return _ordered_detail_rows(by_kind)
def _prepare_scalar_edit_inputs(
submission: ModeloEditSubmissionV1,
) -> tuple[dict[str, Decimal], dict[str, str], tuple[CasillaId, ...]] | ModeloEditExecutionNoEffectV1:
"""Resolve and validate scalar intents before any catalogue read."""
reachable = _reachable_scalar_inputs(submission)
if isinstance(reachable, ModeloEditExecutionNoEffectV1):
return reachable
casilla_inputs, text_casilla_inputs, cleared_casilla_ids = reachable
baseline = submission.baseline
for intent in submission.scalar_intents:
refusal = validate_scalar_intent(baseline, intent.address, intent.kind)
if refusal is not None:
return ModeloEditExecutionNoEffectV1(refusal=refusal)
return casilla_inputs, text_casilla_inputs, cleared_casilla_ids
def _capture_edit_receipt(
calculation_revision_id: str,
bucket_event_id: str | None,
*,
request: ModeloEditApplyRequestV1,
submission: ModeloEditSubmissionV1,
receipt_repository: ModeloEditReceiptRepositoryPort,
now: datetime,
result_destination: str,
captured_receipt: list[ModeloEditMutationResultReceiptV1],
) -> tuple[SecureObjectWrite, ...]:
"""Build and capture the safe receipt co-committed with the calculation."""
baseline = submission.baseline
receipt_id = content_hash_hex(
{
"operation_id": request.operation_id,
"baseline_id": baseline.baseline_id,
"calculation_revision_id": calculation_revision_id,
"bucket_event_id": bucket_event_id or "",
},
)
receipt = ModeloEditMutationResultReceiptV1(
receipt_id=receipt_id,
operation_id=request.operation_id,
mutation_family=submission.mutation_family,
baseline_id=baseline.baseline_id,
work_unit_id=baseline.work_unit_id,
calculation_revision_id=calculation_revision_id,
bucket_event_id=bucket_event_id,
committed_at=now,
result_destination=result_destination,
)
captured_receipt.append(receipt)
return (receipt_repository.to_secure_object_write(receipt),)
def _execute_modelo_edit(
*,
request: ModeloEditApplyRequestV1,
submission: ModeloEditSubmissionV1,
casilla_inputs: dict[str, Decimal],
text_casilla_inputs: dict[str, str],
cleared_casilla_ids: tuple[CasillaId, ...],
detail_rows: tuple[ModeloDetailRow, ...],
ports: CalculationActionPorts,
receipt_repository: ModeloEditReceiptRepositoryPort,
now: datetime,
result_destination: str,
) -> ModeloEditExecutionUpdatedV1:
"""Run the canonical calculation writer and retain its co-committed receipt."""
captured_receipt: list[ModeloEditMutationResultReceiptV1] = []
def _co_commit_receipt(calculation_revision_id: str, bucket_event_id: str | None) -> tuple[SecureObjectWrite, ...]:
return _capture_edit_receipt(
calculation_revision_id,
bucket_event_id,
request=request,
submission=submission,
receipt_repository=receipt_repository,
now=now,
result_destination=result_destination,
captured_receipt=captured_receipt,
)
calculate_modelo_revision_from_bucket_aggregation_with_diagnostics(
submission.baseline.work_unit_id,
actor=_RESPONSIBLE_OWNER,
casilla_inputs=casilla_inputs or None,
text_casilla_inputs=text_casilla_inputs or None,
cleared_casilla_ids=cleared_casilla_ids,
detail_rows=detail_rows,
ports=ports,
clock=now,
additional_secure_object_writes_for_revision=_co_commit_receipt,
)
if not captured_receipt:
raise ModeloError(
"the calculation boundary resolved no revision id for the applied edit",
)
return ModeloEditExecutionUpdatedV1(receipt=captured_receipt[0])
def _reconstruct_current_edit_detail_rows(
*,
baseline: ModeloEditBaselineV1,
calculation_catalogue: CalculationRevisionCatalogue,
detail_row_intents: tuple[ModeloDetailRowEditIntentV1, ...],
operation: PinnedAuthorityOperation,
) -> tuple[ModeloDetailRow, ...] | ModeloEditExecutionNoEffectV1:
"""Revalidate the current revision coordinate before reconstructing its rows."""
current_revision = (
calculation_catalogue.get(baseline.current_calculation_revision_id)
if baseline.current_calculation_revision_id is not None
else None
)
if current_revision is not None:
require_calculation_revision_coordinates_current(current_revision, operation=operation)
return _reconstruct_detail_rows(
current_detail_rows=current_revision.detail_rows if current_revision is not None else (),
detail_row_intents=detail_row_intents,
)
[docs]
def apply_modelo_edit(
request: ModeloEditApplyRequestV1,
*,
ports: CalculationActionPorts,
receipt_repository: ModeloEditReceiptRepositoryPort,
now: datetime,
result_destination: str,
) -> ModeloEditExecutionResultV1:
"""Recheck the baseline at the guarded commit point and execute a CALCULATE edit.
Delegates the actual formula evaluation and persistence to the existing
calculation boundary and single-writer primitive; this function adds only
the commit-point recheck, the intent-reachability gate, and the
atomically co-committed result receipt.
The required calculation ports arrive as one bucket-bound bundle from the
outer composition root. This executor owns no calculation repository
construction.
"""
submission = request.submission
baseline = submission.baseline
if submission.mutation_family is not ModeloEditMutationFamily.CALCULATE:
return _unsupported_intent_refusal(ModeloEditUnsupportedIntentReason.RECALCULATE_NOT_YET_WIRED)
reachable = _prepare_scalar_edit_inputs(submission)
if isinstance(reachable, ModeloEditExecutionNoEffectV1):
return reachable
casilla_inputs, text_casilla_inputs, cleared_casilla_ids = reachable
# Immediately before effect: revision-load the catalogues and recheck
# every baseline coordinate. No re-read happens between this check and
# the guarded commit below other than the calculation boundary's own
# internal, independently CAS-guarded reads.
work_catalogue = ports.work_unit_repository.load()
calculation_catalogue = ports.calculation_repository.load()
stale = reconfirm_modelo_edit_baseline(
baseline, work_catalogue=work_catalogue, calculation_catalogue=calculation_catalogue
)
if stale is not None:
return ModeloEditExecutionNoEffectV1(refusal=stale)
reconstructed_detail_rows = _reconstruct_current_edit_detail_rows(
baseline=baseline,
calculation_catalogue=calculation_catalogue,
detail_row_intents=submission.detail_row_intents,
operation=ports.operation,
)
if isinstance(reconstructed_detail_rows, ModeloEditExecutionNoEffectV1):
return reconstructed_detail_rows
return _execute_modelo_edit(
request=request,
submission=submission,
casilla_inputs=casilla_inputs,
text_casilla_inputs=text_casilla_inputs,
cleared_casilla_ids=cleared_casilla_ids,
detail_rows=reconstructed_detail_rows,
ports=ports,
receipt_repository=receipt_repository,
now=now,
result_destination=result_destination,
)
__all__ = ["apply_modelo_edit"]