"""The sole Modelo Workspace V1 assembly and dispatch entry point.
This module owns the one ordering-critical sequence every Workspace read must
follow: capture WORK exactly once, derive the REGISTRY coordinate only from
that captured :class:`~.work_selection.ModeloWorkResolution`, then evaluate
the requested and stored revision axes independently against the SAME
REGISTRY capture through the sole pure assertion,
:func:`~.work_addressing.assert_work_target_revision`. Neither axis may ever
select the revision the other is judged by, and REGISTRY is never captured a
second time to answer a question the first capture already carries the
coordinates for.
Currently landed: the WORK-then-REGISTRY capture-and-assertion core
(:func:`resolve_modelo_workspace_revision_axes`), tested in isolation. The
full request/admission dispatch and the STATIC_INSPECTION and GRADED_SNAPSHOT
projection assemblies are NOT YET BUILT here; ``ModeloWorkspaceResolvedTargetV1``
also requires a ``review_status`` this module cannot yet source for the
STATIC_INSPECTION admission (see the exec record for the open question). Build
those once that is resolved -- do not infer the missing semantics.
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from ...core.authority_grade import RegistryAuthorityGrade
from ...core.errors.hierarchy import CadrumoError
from ...core.external_constants import OutputLanguage
from ...core.hashing import content_hash_hex
from ...core.period import Period
from ...core.schema_family_disposition import RegistrySchemaFamilyDisposition
from ...domain.calculations.registry.ids import BindingId
from ...domain.calculations.registry.modelo_localization import (
ModeloLocalizationFieldKind,
casilla_occurrence_locale_key,
modelo_localization_source,
revision_locale_key,
)
from ...domain.calculations.registry.relation_prefill_bindings import RelationPrefillProvider
from ...domain.calculations.registry.schema import (
BindingDefinition,
FormulaDefinition,
SchemaFamilyDispositionDeclaration,
)
from ...domain.calculations.registry.schema_formula import FormulaExpression, ParameterDefinition
from ...domain.calculations.registry.static_inspection import RegistryRevisionInspection
from ...domain.modelos.codes import ModeloCode
from ...domain.modelos.work_unit_repository import WorkUnitCatalogueRepositoryProtocol
from .work_addressing import (
ModeloExactWorkUnitTarget,
ModeloVisibleFilingTarget,
diverging_work_target_revision_axes,
)
from .work_selection import (
ModeloWorkResolution,
ModeloWorkSelectionMode,
ModeloWorkSelectorRequest,
)
from .workspace_models import (
ModeloWorkspaceBaselineV1,
ModeloWorkspaceBindingReferenceV1,
ModeloWorkspaceBoundedFacetV1,
ModeloWorkspaceCapabilityDisposition,
ModeloWorkspaceCapabilityName,
ModeloWorkspaceCapabilityV1,
ModeloWorkspaceCasillaReferenceV1,
ModeloWorkspaceContributorIdentityV1,
ModeloWorkspaceCursorV1,
ModeloWorkspaceEvidenceHorizonV1,
ModeloWorkspaceExactWorkUnitTargetV1,
ModeloWorkspaceFacetName,
ModeloWorkspaceFamilyDispositionV1,
ModeloWorkspaceFormulaBindingOperandReferenceV1,
ModeloWorkspaceFormulaCasillaOperandReferenceV1,
ModeloWorkspaceFormulaDateBindingOperandReferenceV1,
ModeloWorkspaceFormulaDispatchOperandReferenceV1,
ModeloWorkspaceFormulaLiteralOperandReferenceV1,
ModeloWorkspaceFormulaOperandReferenceV1,
ModeloWorkspaceFormulaParameterOperandReferenceV1,
ModeloWorkspaceFormulaReferenceV1,
ModeloWorkspaceLocaleDisposition,
ModeloWorkspaceLocaleSummaryV1,
ModeloWorkspaceLocalizedTextV1,
ModeloWorkspaceParameterReferenceV1,
ModeloWorkspaceProjectionV1,
ModeloWorkspaceRelationSourceEndpointReferenceV1,
ModeloWorkspaceRelationTargetEndpointReferenceV1,
ModeloWorkspaceResolvedTargetV1,
ModeloWorkspaceRevisionAssertionDisposition,
ModeloWorkspaceRevisionAssertionSource,
ModeloWorkspaceRevisionAssertionV1,
ModeloWorkspaceSchemaClassification,
ModeloWorkspaceSchemaIdentityV1,
ModeloWorkspaceSchemaRecordV1,
ModeloWorkspaceStaticInspectionResultV1,
ModeloWorkspaceStaticInspectionScopeV1,
ModeloWorkspaceTargetV1,
ModeloWorkspaceTechnicalLabelV1,
ModeloWorkspaceWorkReviewFacetV1,
)
from .workspace_producers import (
MODELO_WORKSPACE_BOUNDED_REVIEW_PRODUCER_CONTRACT_V1,
MODELO_WORKSPACE_CALCULATION_PRODUCER_CONTRACT_V1,
MODELO_WORKSPACE_FIELD_MANIFEST_PRODUCER_CONTRACT_V1,
MODELO_WORKSPACE_LOCALE_CATALOGUE_PRODUCER_CONTRACT_V1,
MODELO_WORKSPACE_READINESS_PRODUCER_CONTRACT_V1,
MODELO_WORKSPACE_REGISTRY_PRODUCER_CONTRACT_V1,
MODELO_WORKSPACE_WORK_PRODUCER_CONTRACT_V1,
ModeloWorkspaceContributingProjectionV1,
ModeloWorkspaceEpochV1,
ModeloWorkspaceFieldManifestPortV1,
ModeloWorkspaceLocaleCataloguePortV1,
ModeloWorkspaceProducerContractV1,
ModeloWorkspaceProducerStampV1,
ModeloWorkspaceRegistryPortV1,
ModeloWorkspaceRegistryProjectionV1,
ModeloWorkspaceWorkPortV1,
RegistryAuthorityCapturePort,
)
[docs]
class ModeloWorkspaceAbsentRegistryProjectionError(CadrumoError):
"""Raised when a registry capture returns without its promised projection."""
[docs]
class ModeloWorkspaceUnresolvedWorkError(CadrumoError):
"""Raised when a workspace read reaches an unresolved work selection."""
STATIC_INSPECTION_WORK_REVIEW_FACET = ModeloWorkspaceWorkReviewFacetV1(
disposition=ModeloWorkspaceCapabilityDisposition.UNMEASURED,
review=None,
)
[docs]
def modelo_work_selector_request_for_target(
target: ModeloWorkspaceTargetV1,
*,
bucket_id: str,
) -> ModeloWorkSelectorRequest:
"""Project one Workspace target arm into the WORK capture's selector request.
The Workspace-level target dataclasses already carry ``.to_work_address()``
for the *legacy* single-read call sites, but Workspace's own WORK port
captures over :class:`ModeloWorkSelectorRequest`, not
:class:`~.work_addressing.ModeloWorkAddress` -- so the mapping is built
directly from the target's own operands rather than round-tripping through
the address shape.
"""
if isinstance(target, ModeloWorkspaceExactWorkUnitTargetV1):
exact: ModeloExactWorkUnitTarget = target.target
return ModeloWorkSelectorRequest(
work_unit_id=exact.work_unit_id,
bucket_id=exact.bucket_id or bucket_id,
)
visible: ModeloVisibleFilingTarget = target.target
return ModeloWorkSelectorRequest(
modelo=ModeloCode(visible.modelo),
filing_year=visible.filing_year,
period=visible.period,
revision_id=visible.registry_revision_id,
bucket_id=visible.bucket_id or bucket_id,
)
[docs]
@dataclass(frozen=True, slots=True)
class ModeloWorkspaceRevisionAxes:
"""The law-determined revision plus both independently checked assertions.
Carries the exact two :class:`ModeloWorkspaceRevisionAssertionV1` rows
``ModeloWorkspaceResolvedTargetV1`` requires, already in their final typed
shape, so the caller assembling that record only has to plug them in.
"""
law_selected_revision_id: str
requested_revision_assertion: ModeloWorkspaceRevisionAssertionV1
stored_revision_assertion: ModeloWorkspaceRevisionAssertionV1
def _revision_assertion(
*,
source: ModeloWorkspaceRevisionAssertionSource,
asserted_revision_id: str | None,
law_revision_id: str,
mismatched_sources: set[ModeloWorkspaceRevisionAssertionSource],
) -> ModeloWorkspaceRevisionAssertionV1:
if asserted_revision_id is None:
disposition = ModeloWorkspaceRevisionAssertionDisposition.NOT_PRESENT
elif source in mismatched_sources:
disposition = ModeloWorkspaceRevisionAssertionDisposition.MISMATCHED
else:
disposition = ModeloWorkspaceRevisionAssertionDisposition.MATCHED
return ModeloWorkspaceRevisionAssertionV1(
source=source,
disposition=disposition,
asserted_revision_id=asserted_revision_id,
)
[docs]
def resolve_modelo_workspace_revision_axes(
resolution: ModeloWorkResolution,
*,
registry_projection: ModeloWorkspaceRegistryProjectionV1,
) -> ModeloWorkspaceRevisionAxes:
"""Judge the requested and stored revision axes against one REGISTRY capture.
``resolution`` MUST be the WORK capture already taken for this same
target; ``registry_projection`` MUST be the REGISTRY capture taken from
the coordinates that resolution carries (``resolution.modelo``,
``resolution.filing_year``, ``resolution.period``) and from no other
source. This function never captures REGISTRY itself -- it only evaluates
the two axes against what the caller already captured.
This function never raises on a mismatch. Each independently evaluated
assertion retains its typed ``MISMATCHED`` disposition in the resolved
target, so callers receive the divergence without losing either axis.
"""
requested_revision_id = resolution.requested_revision_id
stored_revision_id = resolution.work_unit.revision_id if resolution.work_unit is not None else None
law_revision_id = registry_projection.revision_id
# The comparison itself is the shared authority's, not a local copy: this
# surface differs from it only in DISPOSITION -- recording the divergence as
# typed data rather than raising -- and two open-coded copies of one
# normalisation drift silently.
diverging = diverging_work_target_revision_axes(
law_revision_id=law_revision_id,
requested_revision_id=requested_revision_id,
stored_revision_id=stored_revision_id,
)
mismatched: set[ModeloWorkspaceRevisionAssertionSource] = {
source
for axis, source in (
("requested", ModeloWorkspaceRevisionAssertionSource.REQUESTED),
("stored", ModeloWorkspaceRevisionAssertionSource.STORED),
)
if axis in diverging
}
return ModeloWorkspaceRevisionAxes(
law_selected_revision_id=law_revision_id,
requested_revision_assertion=_revision_assertion(
source=ModeloWorkspaceRevisionAssertionSource.REQUESTED,
asserted_revision_id=requested_revision_id,
law_revision_id=law_revision_id,
mismatched_sources=mismatched,
),
stored_revision_assertion=_revision_assertion(
source=ModeloWorkspaceRevisionAssertionSource.STORED,
asserted_revision_id=stored_revision_id,
law_revision_id=law_revision_id,
mismatched_sources=mismatched,
),
)
[docs]
def capture_modelo_workspace_target_captures(
target: ModeloWorkspaceTargetV1,
*,
bucket_id: str,
catalogue_repository: WorkUnitCatalogueRepositoryProtocol,
authority: RegistryAuthorityCapturePort,
grade: RegistryAuthorityGrade | None = None,
) -> tuple[
ModeloWorkspaceContributingProjectionV1[ModeloWorkResolution],
ModeloWorkspaceContributingProjectionV1[ModeloWorkspaceRegistryProjectionV1],
ModeloWorkspaceRevisionAxes,
]:
"""Capture WORK exactly once, then REGISTRY exactly once from its coordinates.
Returns the full stamped-and-epoched captures, not just their bare
projections, so a baseline assembler can fold the WORK/REGISTRY
contributor stamps and epochs into its consistency digest without a
second capture of either.
This is the ordering-critical sequence itself: WORK is resolved first, its
``(modelo, filing_year, period)`` is read back to build the REGISTRY port
-- never the target's own operands, which may name an exact work unit with
no natural coordinates of their own -- and REGISTRY is captured exactly
once from that single WORK-derived coordinate. ``grade=None`` (the
default) requests STATIC_INSPECTION admission
(``RegistryRevisionInspection``); passing a :class:`RegistryAuthorityGrade`
requests GRADED_SNAPSHOT admission (``RegistrySnapshot``) through the
exact same port and the exact same WORK-then-REGISTRY ordering -- the two
admissions differ only in which authority object the one REGISTRY read
returns, never in how many reads happen or in what order.
"""
request = modelo_work_selector_request_for_target(target, bucket_id=bucket_id)
work_port = ModeloWorkspaceWorkPortV1(
request=request,
catalogue_repository=catalogue_repository,
mode=ModeloWorkSelectionMode.VISIBLE_OR_EXACT,
)
work_capture = work_port.capture_projection_with_epoch()
resolution = work_capture.projection
resolved_modelo, resolved_filing_year, resolved_period = _resolved_obligation(resolution)
registry_port = ModeloWorkspaceRegistryPortV1(
authority=authority,
modelo_id=resolved_modelo,
filing_year=resolved_filing_year,
period=resolved_period.registry_token,
grade=grade,
)
registry_capture = registry_port.capture_projection_with_epoch()
registry_projection = registry_capture.projection
axes = resolve_modelo_workspace_revision_axes(resolution, registry_projection=registry_projection)
return work_capture, registry_capture, axes
def _resolve_locale_summary_and_value(
key: str,
*,
output_language: OutputLanguage,
) -> tuple[ModeloWorkspaceLocaleSummaryV1, str | None]:
"""Resolve one canonical locale coordinate plus its text value, for any key.
Shared by the revision-level summary (:func:`capture_modelo_workspace_locale_summary`)
and any per-record label resolution (schema_facet). Spanish is the source
language for every catalogue entry (``aeat-locales-cli``), so a requested
language whose own key is absent falls back to Spanish rather than to an
arbitrary third language; Spanish absent as well is the suppressed floor,
never a missing key propagated as an exception. The returned value is
``None`` only when even the Spanish source is absent -- callers needing a
non-empty display string treat that as a distinct refusal, never a blank.
"""
requested = ModeloWorkspaceLocaleCataloguePortV1(
translation_key=key,
locale=output_language.value,
).capture_projection_with_epoch()
if requested.projection.value is not None:
return (
ModeloWorkspaceLocaleSummaryV1(
requested_language=output_language,
resolved_language=output_language,
disposition=ModeloWorkspaceLocaleDisposition.EXACT,
catalogue_digest=requested.projection.catalogue_digest,
),
requested.projection.value,
)
if output_language is OutputLanguage.ES:
return (
ModeloWorkspaceLocaleSummaryV1(
requested_language=output_language,
resolved_language=OutputLanguage.ES,
disposition=ModeloWorkspaceLocaleDisposition.SUPPRESSED,
catalogue_digest=requested.projection.catalogue_digest,
),
None,
)
spanish = ModeloWorkspaceLocaleCataloguePortV1(
translation_key=key,
locale=OutputLanguage.ES.value,
).capture_projection_with_epoch()
disposition = (
ModeloWorkspaceLocaleDisposition.SPANISH_FALLBACK
if spanish.projection.value is not None
else ModeloWorkspaceLocaleDisposition.SUPPRESSED
)
return (
ModeloWorkspaceLocaleSummaryV1(
requested_language=output_language,
resolved_language=OutputLanguage.ES,
disposition=disposition,
catalogue_digest=spanish.projection.catalogue_digest,
),
spanish.projection.value,
)
[docs]
def capture_modelo_workspace_locale_summary(
resolved_target: ModeloWorkspaceResolvedTargetV1,
*,
output_language: OutputLanguage,
) -> ModeloWorkspaceLocaleSummaryV1:
"""Resolve the canonical locale coordinate for one resolved Workspace target.
Tests the resolved target's own revision-level display key
(:func:`revision_locale_key`) through the sole LOCALE_CATALOGUE port. This
is the natural per-read canonical key -- one Workspace read names exactly
one ``(modelo, revision)`` pair, and that pair's own display label is a
key every Workspace read already needs regardless of which facet a caller
goes on to request.
"""
key = revision_locale_key(resolved_target.modelo, resolved_target.law_selected_revision_id)
summary, _value = _resolve_locale_summary_and_value(key, output_language=output_language)
return summary
# Canonical capability and refusal facade: the
# capability-to-producer mapping is fixed by which of the contributors
# static inspection structurally never reads ("Static inspection captures
# exactly registry, work, locale_catalogue, and field_manifest; it does not
# read bounded_review, calculation, or readiness"), not by matching
# enum spellings. Every one of those three excluded contributors is UNMEASURED
# for this admission per the ADR's own rule -- "absence of a producer... is
# unmeasured, never available" -- which the ADR amendment clarifies covers an
# admission-structural exclusion, not only a graded producer that ran and
# declined to answer. NOT_APPLICABLE was the wrong disposition for this case;
# it is reserved for a producer that DID run and declared the fact
# inapplicable to the specific target.
#
# SCHEMA_INSPECTION is AVAILABLE (static inspection has its
# own field-manifest root, generate_modelo_workspace_field_manifest_for_inspection):
# field_manifest is a real contributor for this admission, so schema_inspection
# is the one capability static inspection answers AVAILABLE for.
_STATIC_INSPECTION_CAPABILITY_DISPOSITIONS: tuple[
tuple[ModeloWorkspaceCapabilityName, ModeloWorkspaceProducerContractV1, ModeloWorkspaceCapabilityDisposition],
...,
] = (
(
ModeloWorkspaceCapabilityName.SCHEMA_INSPECTION,
MODELO_WORKSPACE_FIELD_MANIFEST_PRODUCER_CONTRACT_V1,
ModeloWorkspaceCapabilityDisposition.AVAILABLE,
),
(
ModeloWorkspaceCapabilityName.CALCULATION_MATERIALIZATION,
MODELO_WORKSPACE_CALCULATION_PRODUCER_CONTRACT_V1,
ModeloWorkspaceCapabilityDisposition.UNMEASURED,
),
(
ModeloWorkspaceCapabilityName.VERIFICATION_READINESS,
MODELO_WORKSPACE_BOUNDED_REVIEW_PRODUCER_CONTRACT_V1,
ModeloWorkspaceCapabilityDisposition.UNMEASURED,
),
(
ModeloWorkspaceCapabilityName.FILING_DRAFT_READINESS,
MODELO_WORKSPACE_READINESS_PRODUCER_CONTRACT_V1,
ModeloWorkspaceCapabilityDisposition.UNMEASURED,
),
)
[docs]
def static_inspection_modelo_workspace_capabilities(
resolved_target: ModeloWorkspaceResolvedTargetV1,
) -> tuple[ModeloWorkspaceCapabilityV1, ...]:
"""Return the complete STATIC_INSPECTION capability denominator.
Every row cites the capability's own canonical producer contributor per
the canonical capability mapping; see the module-level comment above this function.
``schema_inspection`` is ``AVAILABLE`` -- field_manifest is a real
STATIC_INSPECTION contributor. The other three are ``UNMEASURED``:
their producers are contributors this admission structurally never reads.
GRADED_SNAPSHOT's dispositions are a distinct, not-yet-answered question
and MUST NOT be derived from this table.
"""
return tuple(
ModeloWorkspaceCapabilityV1(
capability=capability,
disposition=disposition,
target=resolved_target,
selected_revision_id=resolved_target.law_selected_revision_id,
producer_owner=contract.contributor.owner,
producer=contract.contributor.producer,
)
for capability, contract, disposition in _STATIC_INSPECTION_CAPABILITY_DISPOSITIONS
)
[docs]
def fold_slots(bindings: tuple[BindingDefinition, ...]) -> tuple[tuple[BindingId, RelationPrefillProvider], ...]:
"""Return every relation-prefill slot among a revision's bindings.
The fold is declared by the binding itself now, so the slot identity and
the fold identity are the same id; both workspace endpoint rows are built
from this one pairing rather than from a second family that could name a
binding this revision does not declare.
"""
return tuple(
(binding.id, binding.provider) for binding in bindings if isinstance(binding.provider, RelationPrefillProvider)
)
[docs]
def relation_source_endpoints_for_casilla(
bindings: tuple[BindingDefinition, ...],
casilla_id: str,
*,
source_modelo: str,
) -> tuple[ModeloWorkspaceRelationSourceEndpointReferenceV1, ...]:
"""Return the fold-source-endpoint rows whose declared source casilla matches.
A source casilla id is only meaningful inside its own modelo, so a fold
reading another modelo never matches a same-spelled casilla of this one.
"""
return tuple(
ModeloWorkspaceRelationSourceEndpointReferenceV1(relation_id=binding_id, casilla_id=source_casilla_id)
for binding_id, provider in fold_slots(bindings)
if str(provider.source_modelo) == source_modelo
for source_casilla_id in provider.declared_source_casilla_ids
if source_casilla_id == casilla_id
)
[docs]
def relation_target_endpoints_for_binding(
bindings: tuple[BindingDefinition, ...],
binding_id: str,
) -> tuple[ModeloWorkspaceRelationTargetEndpointReferenceV1, ...]:
"""Return the fold-target-endpoint row for a binding that is itself a fold slot."""
return tuple(
ModeloWorkspaceRelationTargetEndpointReferenceV1(relation_id=slot_id, binding_id=slot_id)
for slot_id, _ in fold_slots(bindings)
if slot_id == binding_id
)
[docs]
def resolve_static_inspection_schema_identity(
inspection: RegistryRevisionInspection,
) -> ModeloWorkspaceSchemaIdentityV1:
"""Build the STATIC_INSPECTION schema identity from the one REGISTRY capture already held.
``schema_fingerprint`` is a content digest over the inspection's own
declared identity sets (casilla and binding ids) -- the same shape
``_edit_services.py`` uses for its own, differently-typed
``ModeloEditSchemaIdentityV1`` (interface-ADR-governed), adapted to the
flatter STATIC_INSPECTION type. ``field_manifest_digest`` is exclusively
the inspection-rooted field manifest's own digest; the edit
contract's ``CalculationCompletenessManifest`` digest has its own field
(``completeness_manifest_digest``) on its own type and no longer shares
this one.
"""
field_manifest_port = ModeloWorkspaceFieldManifestPortV1(authority=inspection)
field_manifest_capture = field_manifest_port.capture_projection_with_epoch()
return ModeloWorkspaceSchemaIdentityV1(
schema_id=f"modelo-{inspection.modelo_id}-{inspection.revision_id}".lower(),
schema_fingerprint=content_hash_hex(
{
"casilla_ids": sorted(inspection.casilla_ids),
"binding_ids": sorted(inspection.binding_ids),
}
),
field_manifest_digest=field_manifest_capture.projection.manifest_digest,
)
[docs]
def static_inspection_evidence_horizon(inspection: RegistryRevisionInspection) -> ModeloWorkspaceEvidenceHorizonV1:
"""Build the evidence horizon straight from the inspection's own retained source catalogue."""
source_refs = tuple(sorted(inspection.source_ref_ids))
return ModeloWorkspaceEvidenceHorizonV1(
source_refs=source_refs,
evidence_digest=content_hash_hex({"source_refs": source_refs}),
)
[docs]
def static_inspection_contributors() -> tuple[ModeloWorkspaceContributorIdentityV1, ...]:
"""Return the four contributor identities STATIC_INSPECTION actually reads.
Matches the ADR's own admission-scope sentence exactly: "Static inspection
captures exactly registry, work, locale_catalogue, and field_manifest."
"""
return tuple(
sorted(
(
MODELO_WORKSPACE_WORK_PRODUCER_CONTRACT_V1.contributor,
MODELO_WORKSPACE_LOCALE_CATALOGUE_PRODUCER_CONTRACT_V1.contributor,
MODELO_WORKSPACE_FIELD_MANIFEST_PRODUCER_CONTRACT_V1.contributor,
MODELO_WORKSPACE_REGISTRY_PRODUCER_CONTRACT_V1.contributor,
),
key=lambda contributor: (contributor.owner, contributor.producer),
)
)
[docs]
def resolve_static_inspection_baseline(
target: ModeloWorkspaceResolvedTargetV1,
*,
schema_identity: ModeloWorkspaceSchemaIdentityV1,
locale: ModeloWorkspaceLocaleSummaryV1,
work_stamp: ModeloWorkspaceProducerStampV1,
work_epoch: ModeloWorkspaceEpochV1,
registry_stamp: ModeloWorkspaceProducerStampV1,
registry_epoch: ModeloWorkspaceEpochV1,
locale_stamp: ModeloWorkspaceProducerStampV1,
locale_epoch: ModeloWorkspaceEpochV1,
field_manifest_stamp: ModeloWorkspaceProducerStampV1,
field_manifest_epoch: ModeloWorkspaceEpochV1,
) -> ModeloWorkspaceBaselineV1:
"""Assemble the STATIC_INSPECTION baseline from the four contributors' own stamps and epochs.
Every stamp/epoch pair passed in MUST come from the exact same captures
that produced ``target``, ``schema_identity`` and ``locale`` -- this
function performs no capture of its own, only digesting what the caller
already atomically observed.
"""
stamps = (work_stamp, registry_stamp, locale_stamp, field_manifest_stamp)
epochs = (work_epoch, registry_epoch, locale_epoch, field_manifest_epoch)
contributor_stamp_digest = content_hash_hex([stamp.model_dump(mode="json") for stamp in stamps])
contributor_epoch_digest = content_hash_hex([epoch.model_dump(mode="json") for epoch in epochs])
token = content_hash_hex(
{
"contributor_stamp_digest": contributor_stamp_digest,
"contributor_epoch_digest": contributor_epoch_digest,
"target": target.model_dump(mode="json"),
"selected_revision_id": target.law_selected_revision_id,
"schema_identity": schema_identity.model_dump(mode="json"),
"locale_catalogue_digest": locale.catalogue_digest,
}
)
return ModeloWorkspaceBaselineV1(
token=token,
contributor_stamp_digest=contributor_stamp_digest,
contributor_epoch_digest=contributor_epoch_digest,
target=target,
selected_revision_id=target.law_selected_revision_id,
schema_identity=schema_identity,
locale_catalogue_digest=locale.catalogue_digest,
)
def _resolved_obligation(resolution: ModeloWorkResolution) -> tuple[ModeloCode, int, Period]:
"""Return the modelo, ejercicio and period a resolved selection names.
``ModeloWorkResolution`` leaves all three optional because an unresolved
selection carries none of them. Every caller here has already selected a
work unit, so absence is a defect rather than a state to render -- and it
is refused rather than asserted, because an ``assert`` disappears under
``python -O`` and would let a partial obligation reach the registry port.
"""
if resolution.modelo is None or resolution.filing_year is None or resolution.period is None:
raise ModeloWorkspaceUnresolvedWorkError(
"work selection resolved without a complete modelo/ejercicio/period address",
)
return resolution.modelo, resolution.filing_year, resolution.period
[docs]
class ModeloWorkspaceStaleCursorError(CadrumoError):
"""Raised when a cursor's pinned coordinate no longer matches the current baseline.
A stale cursor MUST refuse rather than silently return a different page:
resuming it against data that moved would return records the caller did
not ask for and has no way to detect.
"""
[docs]
def static_inspection_casilla_schema_records(
inspection: RegistryRevisionInspection,
target: ModeloWorkspaceResolvedTargetV1,
*,
output_language: OutputLanguage,
) -> tuple[ModeloWorkspaceSchemaRecordV1, ...]:
"""Build one schema record per casilla identity, sorted for stable pagination.
Bounded to identity: ``legal_refs`` and ``constraints`` are
``None`` (this admission's producer never carries `CasillaDefinition`
data), never ``()``. ``formula_operands`` and ``relation_endpoints``
consume the shared join functions directly rather than re-deriving either
edge here.
"""
formulas = inspection.formulas
records: list[ModeloWorkspaceSchemaRecordV1] = []
for casilla_id in sorted(inspection.casilla_ids):
chain = inspection.casilla_localization_keys.get(casilla_id) or (
casilla_occurrence_locale_key(
target.modelo, target.law_selected_revision_id, casilla_id, ModeloLocalizationFieldKind.LABEL
),
)
source = modelo_localization_source(chain, locale=output_language.value)
key = chain[0] if source is None else source[0]
locale_summary, value = _resolve_locale_summary_and_value(key, output_language=output_language)
records.append(
ModeloWorkspaceSchemaRecordV1(
reference=ModeloWorkspaceCasillaReferenceV1(casilla_id=casilla_id),
record_family=("casillas",),
section_path=inspection.casilla_sections.get(casilla_id, ()),
data_type="casilla_id",
label=ModeloWorkspaceLocalizedTextV1(
locale_key=key,
value=value if value is not None else casilla_id,
locale=locale_summary,
),
classification=ModeloWorkspaceSchemaClassification.PROJECTED,
family_disposition=RegistrySchemaFamilyDisposition.POPULATED,
legal_refs=None,
constraints=None,
formula_operands=formula_operand_references_for_casilla(formulas, casilla_id),
relation_endpoints=relation_source_endpoints_for_casilla(
inspection.bindings,
casilla_id,
source_modelo=str(target.modelo),
),
)
)
return tuple(records)
[docs]
def binding_schema_records(
binding_ids: frozenset[BindingId],
bindings: tuple[BindingDefinition, ...],
) -> tuple[ModeloWorkspaceSchemaRecordV1, ...]:
"""Build one schema record per binding identity, sorted for stable pagination.
Narrowed from ``inspection: RegistryRevisionInspection`` to the raw
tuples it read internally -- ``BindingDefinition`` is the identical type
on both ``RegistryRevisionInspection`` and ``RegistrySnapshot.revision``, so this
is ONE shared implementation both admissions call, never two copies that
could drift. Unlike a casilla, ``BindingDefinition`` IS retained
whole by both admissions, so ``legal_refs`` is the binding's own real
(possibly empty) tuple, never ``None`` -- the absence rule applies
only where an admission genuinely carries no such data. The
label is ``ModeloWorkspaceTechnicalLabelV1``: no locale convention exists
for binding identities.
"""
bindings_by_id = {binding.id: binding for binding in bindings}
records: list[ModeloWorkspaceSchemaRecordV1] = []
for binding_id in sorted(binding_ids):
binding = bindings_by_id.get(binding_id)
legal_refs = tuple(binding.legal_refs) if binding is not None else None
records.append(
ModeloWorkspaceSchemaRecordV1(
reference=ModeloWorkspaceBindingReferenceV1(binding_id=binding_id),
record_family=("bindings",),
data_type="binding_id",
label=ModeloWorkspaceTechnicalLabelV1(identifier=binding_id),
classification=ModeloWorkspaceSchemaClassification.PROJECTED,
family_disposition=RegistrySchemaFamilyDisposition.POPULATED,
legal_refs=legal_refs,
constraints=(),
relation_endpoints=relation_target_endpoints_for_binding(bindings, binding_id),
)
)
return tuple(records)
[docs]
def parameter_schema_records(
parameters: tuple[ParameterDefinition, ...],
formulas: tuple[FormulaDefinition, ...],
) -> tuple[ModeloWorkspaceSchemaRecordV1, ...]:
"""Build one schema record per parameter, keyed off every formula that dispatches to it.
Narrowed from ``inspection: RegistryRevisionInspection`` to the raw
``parameters``/``formulas`` tuples -- ``ParameterDefinition`` and
``FormulaDefinition`` are the identical type on both admissions, so this
is ONE shared implementation. A parameter has no direct outbound edge of
its own in the registry schema; the only declared connection is a
formula's own ``dispatch_table`` operand naming it, which
:func:`formula_expression_operand_references` already extracts as
``ModeloWorkspaceFormulaParameterOperandReferenceV1`` and
``ModeloWorkspaceFormulaDispatchOperandReferenceV1`` entries.
"""
parameter_operands: dict[str, list[ModeloWorkspaceFormulaOperandReferenceV1]] = {}
for formula in formulas:
for reference in formula_expression_operand_references(formula.id, formula.expression):
if isinstance(reference, ModeloWorkspaceFormulaParameterOperandReferenceV1):
parameter_operands.setdefault(reference.parameter_id, []).append(reference)
elif isinstance(reference, ModeloWorkspaceFormulaDispatchOperandReferenceV1):
for parameter_id in reference.parameter_ids:
parameter_operands.setdefault(parameter_id, []).append(reference)
records: list[ModeloWorkspaceSchemaRecordV1] = []
for parameter in sorted(parameters, key=lambda item: item.id):
records.append(
ModeloWorkspaceSchemaRecordV1(
reference=ModeloWorkspaceParameterReferenceV1(parameter_id=parameter.id),
record_family=("parameters",),
data_type="parameter_id",
label=ModeloWorkspaceTechnicalLabelV1(identifier=parameter.id),
classification=ModeloWorkspaceSchemaClassification.PROJECTED,
family_disposition=RegistrySchemaFamilyDisposition.POPULATED,
legal_refs=tuple(parameter.legal_refs),
constraints=(),
formula_operands=tuple(parameter_operands.get(parameter.id, ())),
)
)
return tuple(records)
[docs]
def static_inspection_schema_records(
inspection: RegistryRevisionInspection,
target: ModeloWorkspaceResolvedTargetV1,
*,
output_language: OutputLanguage,
) -> tuple[ModeloWorkspaceSchemaRecordV1, ...]:
"""Return the complete STATIC_INSPECTION schema_facet across all five reference kinds.
Sorted by ``(reference.kind, identity)`` so the whole sequence, and
therefore pagination over it, is deterministic and stable across
identical repeated reads.
"""
records = (
static_inspection_casilla_schema_records(inspection, target, output_language=output_language)
+ binding_schema_records(inspection.binding_ids, inspection.bindings)
+ formula_schema_records(inspection.formulas)
+ parameter_schema_records(inspection.parameters, inspection.formulas)
)
return tuple(sorted(records, key=lambda record: (record.reference.kind, str(record.reference))))
def _facet_cursor(
cursor: ModeloWorkspaceCursorV1 | None,
facet: ModeloWorkspaceFacetName,
) -> ModeloWorkspaceCursorV1 | None:
"""Return ``cursor`` when it addresses ``facet``, otherwise ``None``.
A result assembles several facets from one call, so a caller holds at
most one cursor at a time and the cursor itself names which facet it
continues. Routing on that name keeps the other facets at their first
page instead of applying one facet's offset to another's records.
"""
return cursor if cursor is not None and cursor.facet is facet else None
[docs]
def paginate_modelo_workspace_facet[RecordT](
facet_type: type[ModeloWorkspaceBoundedFacetV1[RecordT]],
records: tuple[RecordT, ...],
*,
facet: ModeloWorkspaceFacetName,
target: ModeloWorkspaceResolvedTargetV1,
schema_identity: ModeloWorkspaceSchemaIdentityV1,
baseline: ModeloWorkspaceBaselineV1,
contributors: tuple[ModeloWorkspaceContributorIdentityV1, ...],
disposition: ModeloWorkspaceCapabilityDisposition,
page_size: int,
cursor: ModeloWorkspaceCursorV1 | None = None,
) -> ModeloWorkspaceBoundedFacetV1[RecordT]:
"""Return one bounded, cursor-consistent page from the complete ``records`` sequence.
``records`` MUST already be in the caller's canonical stable order --
pagination consumes an offset over that fixed order, never re-derives it.
A ``cursor`` from a DIFFERENT baseline, revision, schema identity, facet,
or contributor epoch refuses outright rather than silently starting over
or returning a page from the wrong coordinate.
This is the ONE paginator for every bounded facet. It exists as a single
authority because ``ModeloWorkspaceBoundedFacetV1`` requires ``has_more``
to agree with ``next_cursor``: a facet built by truncating records and
setting ``has_more`` without minting the matching cursor does not merely
lose pagination, it fails model validation outright and takes the whole
projection down with it. Minting the cursor is therefore not a
convenience this helper offers, it is the only way to construct an
overflowing facet at all, and every facet routes through here so that no
call site can rediscover that the hard way.
``facet_type`` is the caller's own concrete parametrization (for example
``ModeloWorkspaceBoundedFacetV1[ModeloWorkspaceSchemaRecordV1]``). It is
passed rather than parametrized here from ``RecordT`` because a type
parameter is only a ``TypeVar`` at runtime: pydantic would build the
model with the record field effectively unvalidated, silently trading
this boundary's strictness for the generalisation.
"""
if cursor is not None:
if (
cursor.baseline != baseline
or cursor.selected_revision_id != target.law_selected_revision_id
or cursor.schema_identity != schema_identity
or cursor.facet is not facet
or cursor.contributor_epoch_digest != baseline.contributor_epoch_digest
):
raise ModeloWorkspaceStaleCursorError(
f"workspace {facet.value} facet cursor no longer matches the current baseline coordinate"
)
offset = int(cursor.continuation)
else:
offset = 0
page = records[offset : offset + page_size]
next_offset = offset + len(page)
has_more = next_offset < len(records)
next_cursor = (
ModeloWorkspaceCursorV1(
baseline=baseline,
selected_revision_id=target.law_selected_revision_id,
schema_identity=schema_identity,
facet=facet,
contributor_epoch_digest=baseline.contributor_epoch_digest,
continuation=str(next_offset),
)
if has_more
else None
)
return facet_type(
selected_revision_id=target.law_selected_revision_id,
schema_identity=schema_identity,
baseline=baseline,
contributor_epoch_digest=baseline.contributor_epoch_digest,
contributors=contributors,
facet=facet,
disposition=disposition,
records=page,
page_size=page_size,
next_cursor=next_cursor,
has_more=has_more,
)
def _not_applicable_family_dispositions(
family_dispositions: Mapping[str, SchemaFamilyDispositionDeclaration],
) -> tuple[ModeloWorkspaceFamilyDispositionV1, ...]:
"""Project only the family dispositions the declarations mapping can honestly attest to.
``RegistryRevisionInspection.family_dispositions`` and
``ModeloRevision.family_dispositions`` are the identical mapping (the
inspection copies it straight from the revision at construction), so
this one function is shared by both admissions rather than written
twice. It carries exactly the families the revision has explicitly
declared NOT_APPLICABLE, each grounded with its own
reason/legal_refs/source_refs -- a substantive claim the registry itself
made. A family absent from this mapping is not reported here at all:
silently defaulting an unreported family to POPULATED or
BLOCKED_PENDING_EVIDENCE would assert a fact this data has no basis for.
Reporting nothing is honest; guessing is not.
"""
return tuple(
sorted(
(
ModeloWorkspaceFamilyDispositionV1(
family=family,
disposition=RegistrySchemaFamilyDisposition.NOT_APPLICABLE,
legal_refs=tuple(declaration.legal_refs),
source_refs=tuple(declaration.source_refs),
)
for family, declaration in family_dispositions.items()
),
key=lambda item: item.family,
)
)
[docs]
def static_inspection_family_dispositions(
inspection: RegistryRevisionInspection,
) -> tuple[ModeloWorkspaceFamilyDispositionV1, ...]:
"""Project only the family dispositions the inspection can honestly attest to.
See :func:`_not_applicable_family_dispositions` for the shared logic;
the inspection carries no data for most schema families (it strips
everything but casilla/binding/formula/relation/parameter/projection-endpoint/
workbook-parity/live-cross-reference identifiers).
"""
return _not_applicable_family_dispositions(inspection.family_dispositions)
[docs]
def resolve_static_inspection_result(
target: ModeloWorkspaceTargetV1,
*,
bucket_id: str,
catalogue_repository: WorkUnitCatalogueRepositoryProtocol,
authority: RegistryAuthorityCapturePort,
output_language: OutputLanguage,
page_size: int = 200,
cursor: ModeloWorkspaceCursorV1 | None = None,
) -> ModeloWorkspaceStaticInspectionResultV1:
"""Assemble the complete, single-page STATIC_INSPECTION result for one target.
``page_size`` defaults to 200, the schema facet's own maximum page size
(``ModeloWorkspaceBoundedFacetV1``'s ``_MAX_FACET_PAGE_SIZE``), so a
revision whose schema fits within that bound returns in one page; a
caller working through a larger schema paginates via ``next_cursor``
exactly as :func:`paginate_modelo_workspace_facet` proves.
Captures WORK then REGISTRY exactly once each (the ordering-critical
core), then builds every remaining piece from that one REGISTRY
capture's inspection: schema identity, locale summary, evidence horizon,
family dispositions, contributors, baseline, the five-kind schema_facet,
the fixed work_review facet, and the capability denominator. No second
registry or work read occurs anywhere in this function.
"""
if cursor is not None and cursor.facet is not ModeloWorkspaceFacetName.SCHEMA:
raise ModeloWorkspaceStaleCursorError(
f"static inspection paginates only the schema facet; cursor names {cursor.facet.value}"
)
work_capture, registry_capture, axes = capture_modelo_workspace_target_captures(
target,
bucket_id=bucket_id,
catalogue_repository=catalogue_repository,
authority=authority,
)
resolution = work_capture.projection
registry_projection = registry_capture.projection
inspection = registry_projection.inspection
if inspection is None:
raise ModeloWorkspaceAbsentRegistryProjectionError(
"registry capture returned no inspection for the resolved revision",
)
work_unit = resolution.work_unit
resolved_modelo, resolved_filing_year, resolved_period = _resolved_obligation(resolution)
resolved_target = ModeloWorkspaceResolvedTargetV1(
bucket_id=resolution.bucket_id,
modelo=resolved_modelo,
filing_year=resolved_filing_year,
period=resolved_period,
law_selected_revision_id=axes.law_selected_revision_id,
review_status=registry_projection.review_status,
requested_revision_assertion=axes.requested_revision_assertion,
stored_revision_assertion=axes.stored_revision_assertion,
work_unit_id=work_unit.work_unit_id if work_unit is not None else None,
work_state=work_unit.state if work_unit is not None else None,
)
schema_identity = resolve_static_inspection_schema_identity(inspection)
locale = capture_modelo_workspace_locale_summary(resolved_target, output_language=output_language)
locale_key = revision_locale_key(resolved_target.modelo, resolved_target.law_selected_revision_id)
locale_capture = ModeloWorkspaceLocaleCataloguePortV1(
translation_key=locale_key,
locale=output_language.value,
).capture_projection_with_epoch()
field_manifest_port = ModeloWorkspaceFieldManifestPortV1(authority=inspection)
field_manifest_capture = field_manifest_port.capture_projection_with_epoch()
baseline = resolve_static_inspection_baseline(
resolved_target,
schema_identity=schema_identity,
locale=locale,
work_stamp=work_capture.stamp,
work_epoch=work_capture.epoch,
registry_stamp=registry_capture.stamp,
registry_epoch=registry_capture.epoch,
locale_stamp=locale_capture.stamp,
locale_epoch=locale_capture.epoch,
field_manifest_stamp=field_manifest_capture.stamp,
field_manifest_epoch=field_manifest_capture.epoch,
)
contributors = static_inspection_contributors()
records = static_inspection_schema_records(inspection, resolved_target, output_language=output_language)
schema_facet = paginate_modelo_workspace_facet(
ModeloWorkspaceBoundedFacetV1[ModeloWorkspaceSchemaRecordV1],
records,
facet=ModeloWorkspaceFacetName.SCHEMA,
cursor=_facet_cursor(cursor, ModeloWorkspaceFacetName.SCHEMA),
target=resolved_target,
schema_identity=schema_identity,
baseline=baseline,
contributors=contributors,
disposition=ModeloWorkspaceCapabilityDisposition.AVAILABLE,
page_size=page_size,
)
evidence_horizon = static_inspection_evidence_horizon(inspection)
family_dispositions = static_inspection_family_dispositions(inspection)
capabilities = static_inspection_modelo_workspace_capabilities(resolved_target)
projection = ModeloWorkspaceProjectionV1(
admission=ModeloWorkspaceStaticInspectionScopeV1(),
target=resolved_target,
schema_identity=schema_identity,
locale=locale,
evidence_horizon=evidence_horizon,
family_dispositions=family_dispositions,
contributors=contributors,
baseline=baseline,
schema_facet=schema_facet,
work_review=STATIC_INSPECTION_WORK_REVIEW_FACET,
capabilities=capabilities,
)
return ModeloWorkspaceStaticInspectionResultV1(projection=projection)