cadrumo.application.aggregation.source_resolution_operations module

Canonical operations over resolved calculation-source envelopes.

The source-resolution contract lives in source_mesh; this module owns the operations that combine resolver results, emit repeated diagnostics and provenance, and make unresolved or degraded sources explicit. Keeping those operations together leaves the envelope module focused on its validated data contract while preserving one implementation for every merge collision rule.

source_diagnostics_for(items, *, reason, source_kind, resolver_id, message, source_ref=<function _no_source_text>, remedy=<function _no_source_text>)[source]

Project repeated source facts through the diagnostic contract.

Return type:

tuple[CalculationSourceDiagnostic, ...]

Parameters:

  • items (Iterable)

  • reason (Literal['duplicate_binding_owner', 'duplicate_bound_casilla_owner', 'duplicate_relation_owner', 'source_issue', 'unresolved_binding', 'storage_degraded', 'source_domain_not_ready', 'unhandled_binding_source', 'deferred_binding_source', 'unresolved_derived_binding', 'orphaned_override', 'terminal_origin_mismatch', 'unrouted_observation', 'unrouted_declarable_quantity', 'structurally_unroutable_base_category', 'ungrounded_income_substrate', 'unusable_sales_invoice_evidence', 'devengo_date_proxy_attribution', 'oss_no_live_source', 'missing_transaction_evidence', 'administrador_retencion_rate_mismatch', 'inferred_retencion_rate_unmatched', 'inferred_retencion_sectoral_rate_unconfirmed', 'm349_clave_inferred_from_category', 'unconverted_foreign_currency', 'unclassified_declarant_role_fact', 'invoice_category_counterparty_mismatch', 'invoice_reverse_charge_cuota_not_derivable', 'invoice_recargo_departs_from_published_rate', 'official_box_unpopulated', 'prior_payment_not_deducted', 'prior_payment_minoracion_not_captured', 'settlement_not_computed', 'prorrata_especial_obligatoria', 'prorrata_especial_check_unavailable', 'dt12_regime_window_closed', 'dt12_regime_window_unverified', 'dt12_parcial_rescate_guidance', 'rate_boxes_underaccount_total', 'aggregation_activity_undeclared', 'operator_override_diverges_from_computed'])

  • source_kind (str)

  • resolver_id (str | None)

  • message (Callable[[T], str])

  • source_ref (Callable[[T], str | None])

  • remedy (Callable[[T], str | None])

source_issue_diagnostics(issues, *, source_kind, resolver_id, suppressed_reasons=frozenset({}))[source]

Project typed aggregation issues, optionally excluding known non-advisories.

Return type:

tuple[CalculationSourceDiagnostic, ...]

Parameters:
source_provenance_for(items, project)[source]

Project one provenance record for each source fact.

Return type:

tuple[CalculationSourceProvenance, ...]

Parameters:
flatten_source_provenance_for(items, project)[source]

Flatten source facts that contribute more than one provenance record.

Return type:

tuple[CalculationSourceProvenance, ...]

Parameters:
sorted_source_ids(items, project)[source]

Return stable identifiers for a source collection.

Return type:

tuple[str, ...]

Parameters:
merge_source_resolutions(resolutions, *, resolver_id=CompositeSourceResolverId.EXCLUSIVE_MESH)[source]

Merge resolver outputs and reject ambiguous ownership.

Return type:

CalculationSourceResolution

Parameters:
merge_source_resolutions_by_precedence(tiers, *, resolver_id=CompositeSourceResolverId.PRECEDENCE_MESH)[source]

Overlay tiers into one source resolution, with later tiers winning values.

Return type:

CalculationSourceResolution

Parameters:
collect_unhandled_source_diagnostics(revision, *, handled_sources, manual_sources=frozenset({'manual_input'}))[source]

Return diagnostics for revision bindings with no enrolled resolver.

Core types: ModeloRevision.

Return type:

tuple[CalculationSourceDiagnostic, ...]

Parameters:
storage_degradation_resolution(*, resolver_id, owned_sources, source_kinds, error)[source]

Return an empty resolution carrying secure-storage degradation diagnostics.

Return type:

CalculationSourceResolution

Parameters: