"""Pure assembly of one installed-workbench projection generation.
The installed-session composition root owns secure readers and source
projectors. It supplies the already-loaded, frontend-neutral inputs declared
here; this module joins those inputs into one immutable generation and derives
the Home and search projections through their existing application composers.
An absent reader result is deliberately not represented by an empty tuple or
an empty projection. ``LOCKED``, ``NEVER_CAPTURED`` and ``UNAVAILABLE`` are
separate source outcomes and carry an explicit refusal. The output contract
does not retain the source-input models, so raw source facts cannot cross this
assembly boundary accidentally.
Core types:
:class:`~cadrumo.domain.modelos.calculation_revision.CalculationRevision`,
:class:`~cadrumo.domain.invoices.models.InvoiceCatalogue`,
:class:`~cadrumo.domain.modelos.filing_record.ModeloRecord`,
:class:`~cadrumo.domain.deadlines.models.TaxpayerProfile`,
:class:`~cadrumo.domain.transactions.models.TransactionCatalogue`,
:class:`~cadrumo.domain.user_profile.values.UserProfileRecord`.
"""
from __future__ import annotations
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from datetime import date
from enum import StrEnum
from typing import TYPE_CHECKING, Final, Literal, Protocol, Self, runtime_checkable
from zoneinfo import ZoneInfo
from pydantic import BaseModel, ValidationError, model_validator
from ..core.errors.hierarchy import InternalInvariantError, pydantic_validation_boundary
from ..core.identifier_grammar import NamespacedId
from ..core.models import STRICT_FROZEN_CONFIG
from ..core.time.utc import UtcInstant
from ..domain.buckets.event import (
BucketEvent,
BucketEventHistoryCatalogue,
BucketEventObjectType,
BucketEventType,
bucket_event_order_key,
)
from ..domain.buckets.protocols import BucketEventHistoryRepositoryProtocol
from ..domain.deadlines.errors import ProfileError
from ..domain.deadlines.models import TaxpayerProfile
from ..domain.invoices.models import InvoiceCatalogue
from ..domain.invoices.protocols import InvoiceCatalogueRepositoryProtocol
from ..domain.modelos.calculation_revision import (
CalculationRevision,
CalculationRevisionCatalogue,
CalculationRevisionState,
)
from ..domain.modelos.filing_record import ModeloRecord, ModeloRecordCatalogue
from ..domain.modelos.protocols import (
CalculationRevisionCatalogueRepositoryProtocol,
ModeloRecordCatalogueRepositoryProtocol,
VerificationReportCatalogueRepositoryProtocol,
)
from ..domain.modelos.verification_report import (
ModeloVerificationFindingKind,
ModeloVerificationFindingSeverity,
VerificationCompletenessStatus,
VerificationReportCatalogue,
)
from ..domain.modelos.work_unit import WorkUnit, WorkUnitCatalogue, WorkUnitState
from ..domain.modelos.work_unit_repository import WorkUnitCatalogueRepositoryProtocol
from ..domain.transactions.models import TransactionCatalogue
from ..domain.transactions.protocols import TransactionCatalogueRepositoryProtocol
from ..domain.user_profile.errors import UserProfileValidationError
from ..domain.user_profile.values import UserProfileRecord
from .aeat_sync.workspace import AeatSyncWorkspaceProjectionError, AeatSyncWorkspaceProjectionV1
from .calculations.verification_report_gate import require_verification_report_coordinates_current
from .ledger.workspace import (
LedgerWorkspaceArea,
LedgerWorkspaceProjectionV1,
LedgerWorkspaceStatus,
)
from .modelo.declarations_calendar import (
DeclarationsCalendarProjectionV1,
DeclarationsCalendarSource,
DeclarationsCalendarSourceObservationV1,
project_declarations_calendar,
)
from .modelo.declarations_workspace import (
DeclarationResultCasillaReaderV1,
DeclarationsLifecycleKind,
DeclarationsSanitizedLifecycleFactV1,
DeclarationsWorkspaceAvailability,
DeclarationsWorkspaceProjectionError,
DeclarationsWorkspaceProjectionV1,
DeclarationsWorkspaceZone,
DeclarationsWorkspaceZoneObservationV1,
project_declarations_workspace,
)
from .modelo.workspace_models import ModeloWorkspaceProjectionV1
from .operations.registry import OperationPublicContractSetV1
from .overview.agenda import OverviewAgenda, build_overview_agenda
from .overview.calendar import build_overview_calendar
from .overview.calendar_models import OverviewCalendar, OverviewCalendarRange
from .overview.evidence import (
AeatCalendarEvidenceSources,
CalendarEvidenceProjection,
CalendarEvidenceReadOutcome,
LocalCalendarEvidenceSources,
build_calendar_evidence_projection,
)
from .overview.home import (
HOME_REASON_BLOCKED_DEPENDENCY,
HOME_REASON_DECLARATION_NEEDS_REVIEW,
HOME_REASON_EVIDENCE_MISSING,
HOME_REASON_LEDGER_CLASSIFICATION_PENDING,
HomeAccountSession,
HomeAvailability,
HomeDeclarationResume,
HomeDeclarationState,
HomeLedgerReadiness,
HomeNextAction,
HomeProjectionInput,
HomeProjectionV1,
HomeZoneState,
compose_home_projection,
)
from .overview.next_actions import declare_next_action
from .search.installed_workbench import (
InstalledWorkbenchSearchSnapshotV1,
assemble_installed_workbench_search_snapshot,
)
from .search.workbench import WorkbenchDestinationAdmission, WorkbenchDestinationAdmissionState
from .user_profile.projections import projection_for_taxpayer, record_to_path_values
from .workbench_capture_memory import WorkbenchCalendarMemoKey, WorkbenchCalendarWork, WorkbenchCaptureMemory
WORKBENCH_GENERATION_CONTRACT_VERSION: Literal[1] = 1
_AEAT_SYNC_READER_UNAVAILABLE: Final[str] = "workbench.aeat_sync.reader_unavailable"
_AEAT_SYNC_SNAPSHOT_PROJECTOR_UNAVAILABLE: Final[str] = "workbench.aeat_sync.snapshot_projector_unavailable"
if TYPE_CHECKING:
from ..domain.calculations.registry.authority import PinnedAuthorityOperation
from .ledger.action_ports import LedgerActionPorts
[docs]
class WorkbenchGenerationAvailability(StrEnum):
"""Truthful state of one already-captured source result."""
AVAILABLE = "available"
STALE = "stale"
LOCKED = "locked"
NEVER_CAPTURED = "never_captured"
UNAVAILABLE = "unavailable"
_OBSERVABLE = frozenset(
{
WorkbenchGenerationAvailability.AVAILABLE,
WorkbenchGenerationAvailability.STALE,
}
)
def _validate_result_state(
*,
availability: WorkbenchGenerationAvailability,
observed_at: UtcInstant | None,
refusal: NamespacedId | None,
has_value: bool,
label: str,
) -> None:
"""Enforce the no-false-empty state machine shared by input/output rows."""
if availability in _OBSERVABLE:
_validate_observable_result_state(availability, observed_at, refusal, has_value, label)
return
_validate_unobservable_result_state(availability, observed_at, refusal, has_value, label)
def _validate_observable_result_state(
availability: WorkbenchGenerationAvailability,
observed_at: UtcInstant | None,
refusal: NamespacedId | None,
has_value: bool,
label: str,
) -> None:
if observed_at is None:
raise ValueError(f"{label} {availability.value} result requires an observation time")
if availability is WorkbenchGenerationAvailability.AVAILABLE and refusal is not None:
raise ValueError(f"{label} available result cannot carry a refusal")
if not has_value:
raise ValueError(f"{label} {availability.value} result requires a value")
if availability is WorkbenchGenerationAvailability.STALE and refusal is None:
raise ValueError(f"{label} stale result requires a refusal")
def _validate_unobservable_result_state(
availability: WorkbenchGenerationAvailability,
observed_at: UtcInstant | None,
refusal: NamespacedId | None,
has_value: bool,
label: str,
) -> None:
if observed_at is not None:
raise ValueError(f"{label} {availability.value} result cannot carry an observation time")
if refusal is None:
raise ValueError(f"{label} {availability.value} result requires a refusal")
if has_value:
raise ValueError(f"{label} {availability.value} result cannot carry a value")
_LIFECYCLE_KIND_BY_EVENT: Final[Mapping[BucketEventType, DeclarationsLifecycleKind]] = {
BucketEventType.MODELO_WORK_UNIT_CREATED: DeclarationsLifecycleKind.CREATED,
BucketEventType.MODELO_WORK_UNIT_RENAMED: DeclarationsLifecycleKind.RENAMED,
BucketEventType.MODELO_CALCULATION_CREATED: DeclarationsLifecycleKind.CALCULATED,
BucketEventType.MODELO_VERIFICATION_PASSED: DeclarationsLifecycleKind.VERIFIED,
BucketEventType.MODELO_VERIFICATION_REFUSED: DeclarationsLifecycleKind.VERIFICATION_REFUSED,
BucketEventType.MODELO_FILED: DeclarationsLifecycleKind.FILED,
BucketEventType.MODELO_FILING_RECONCILED: DeclarationsLifecycleKind.RECONCILED,
BucketEventType.MODELO_OBSERVATION_OVERRIDDEN: DeclarationsLifecycleKind.OBSERVATION_OVERRIDDEN,
BucketEventType.MODELO_OBSERVATION_OVERRIDE_CLEARED: DeclarationsLifecycleKind.OBSERVATION_OVERRIDE_CLEARED,
}
"""The bucket events the filing history reports, by the lifecycle meaning they carry.
Every other ``MODELO_*`` event is outside this history's closed vocabulary --
exports, revision reconciliations, wallet corrections -- and is left to the
per-modelo history view that renders raw events.
"""
def _lifecycle_work_unit_id(
event: BucketEvent,
*,
revisions: CalculationRevisionCatalogue,
verification: VerificationReportCatalogue | None,
filings: ModeloRecordCatalogue,
) -> str | None:
"""Resolve the declaration an event belongs to.
An event about an observation has no declaration object of its own, so it
names its work unit in the payload; every other event resolves through the
object it references.
"""
payload_work_unit_id = event.payload.get("work_unit_id")
if payload_work_unit_id:
return payload_work_unit_id
object_id = event.object_id
if event.object_type is BucketEventObjectType.WORK_UNIT:
return object_id
if event.object_type is BucketEventObjectType.CALCULATION_REVISION:
revision = revisions.revisions.get(object_id)
return None if revision is None else str(revision.work_unit_id)
if event.object_type is BucketEventObjectType.VERIFICATION_REPORT:
report = None if verification is None else verification.get(object_id)
if report is None:
return None
revision = revisions.revisions.get(report.calculation_revision_id)
return None if revision is None else str(revision.work_unit_id)
if event.object_type is BucketEventObjectType.FILING_RECORD:
record = filings.records.get(object_id)
return None if record is None else str(record.work_unit_id)
return None
def _declarations_lifecycle_facts(
events: BucketEventHistoryCatalogue,
*,
work_units: WorkUnitCatalogue,
revisions: CalculationRevisionCatalogue,
verification: VerificationReportCatalogue | None,
filings: ModeloRecordCatalogue,
) -> tuple[DeclarationsSanitizedLifecycleFactV1, ...]:
"""Project the persisted event log into the Declarations filing history.
Only events whose declaration is still in the catalogue are reported: the
workspace joins every fact to a live declaration address, and a fact for a
work unit that is no longer there has no row to belong to.
"""
known = {str(unit.work_unit_id) for unit in work_units.values()}
facts: list[DeclarationsSanitizedLifecycleFactV1] = []
for event in sorted(events.events.values(), key=bucket_event_order_key):
kind = _LIFECYCLE_KIND_BY_EVENT.get(event.event_type)
if kind is None:
continue
work_unit_id = _lifecycle_work_unit_id(event, revisions=revisions, verification=verification, filings=filings)
if work_unit_id is None or work_unit_id not in known:
continue
facts.append(
DeclarationsSanitizedLifecycleFactV1(
fact_id=event.event_id,
work_unit_id=work_unit_id,
occurred_at=event.occurred_at,
kind=kind,
)
)
return tuple(facts)
def _read_declarations_workspace(
*,
operation: PinnedAuthorityOperation,
bucket_id: str,
work_units: WorkUnitCatalogue,
calculation_revisions: CalculationRevisionCatalogue,
filing_records: ModeloRecordCatalogue,
observed_at: UtcInstant,
result_casilla_reader: DeclarationResultCasillaReaderV1 | None,
lifecycle_facts: tuple[DeclarationsSanitizedLifecycleFactV1, ...] | None,
) -> DeclarationsWorkspaceProjectionV1 | None:
history_observation = (
DeclarationsWorkspaceZoneObservationV1(
zone=DeclarationsWorkspaceZone.FILING_HISTORY,
availability=DeclarationsWorkspaceAvailability.UNAVAILABLE,
reason_code="workbench.declarations.lifecycle_reader_unavailable",
)
if lifecycle_facts is None
else _declarations_observation(DeclarationsWorkspaceZone.FILING_HISTORY, observed_at)
)
try:
return project_declarations_workspace(
operation=operation,
bucket_id=bucket_id,
work_units=work_units,
calculation_revisions=calculation_revisions,
filing_records=filing_records,
lifecycle_facts=() if lifecycle_facts is None else lifecycle_facts,
result_casilla_reader=result_casilla_reader,
zone_observations=(
_declarations_observation(DeclarationsWorkspaceZone.DECLARATIONS, observed_at),
_declarations_observation(DeclarationsWorkspaceZone.CALCULATION_REVISIONS, observed_at),
history_observation,
),
)
except DeclarationsWorkspaceProjectionError:
return None
[docs]
class WorkbenchGenerationSourceResultV1[SourceT](BaseModel):
"""Typed result admitted from one preloaded application read door.
``value`` is input-only: it is either a safe ``HomeProjectionInput`` or an
already-built safe workspace projection. The assembler copies only the
resulting projection into :class:`WorkbenchGenerationV1`.
"""
model_config = STRICT_FROZEN_CONFIG
availability: WorkbenchGenerationAvailability
observed_at: UtcInstant | None = None
refusal: NamespacedId | None = None
value: SourceT | None = None
@model_validator(mode="after")
@pydantic_validation_boundary
def _state_is_truthful(self) -> Self:
_validate_result_state(
availability=self.availability,
observed_at=self.observed_at,
refusal=self.refusal,
has_value=self.value is not None,
label="source",
)
return self
[docs]
@classmethod
def available(cls, value: SourceT, *, observed_at: UtcInstant) -> Self:
"""Construct a source result with a measured value."""
return cls(availability=WorkbenchGenerationAvailability.AVAILABLE, observed_at=observed_at, value=value)
[docs]
@classmethod
def stale(cls, value: SourceT, *, observed_at: UtcInstant, refusal: NamespacedId) -> Self:
"""Construct a source result retaining a known but stale value."""
return cls(
availability=WorkbenchGenerationAvailability.STALE,
observed_at=observed_at,
refusal=refusal,
value=value,
)
[docs]
@classmethod
def locked(cls, *, refusal: NamespacedId) -> Self:
"""Construct a source result blocked by local custody."""
return cls(availability=WorkbenchGenerationAvailability.LOCKED, refusal=refusal)
[docs]
@classmethod
def never_captured(cls, *, refusal: NamespacedId) -> Self:
"""Construct a source result for a source not read in this session."""
return cls(availability=WorkbenchGenerationAvailability.NEVER_CAPTURED, refusal=refusal)
[docs]
@classmethod
def unavailable(cls, *, refusal: NamespacedId) -> Self:
"""Construct a source result whose reader cannot currently answer."""
return cls(availability=WorkbenchGenerationAvailability.UNAVAILABLE, refusal=refusal)
[docs]
class WorkbenchGenerationProjectionResultV1[ProjectionT](BaseModel):
"""One immutable safe projection plus its source availability evidence."""
model_config = STRICT_FROZEN_CONFIG
availability: WorkbenchGenerationAvailability
observed_at: UtcInstant | None = None
refusal: NamespacedId | None = None
projection: ProjectionT | None = None
@model_validator(mode="after")
@pydantic_validation_boundary
def _state_is_truthful(self) -> Self:
_validate_result_state(
availability=self.availability,
observed_at=self.observed_at,
refusal=self.refusal,
has_value=self.projection is not None,
label="projection",
)
return self
[docs]
class WorkbenchGenerationV1(BaseModel):
"""Immutable public generation consumed by an installed workbench root."""
model_config = STRICT_FROZEN_CONFIG
contract_version: Literal[1] = WORKBENCH_GENERATION_CONTRACT_VERSION
assembled_at: UtcInstant
home: WorkbenchGenerationProjectionResultV1[HomeProjectionV1]
ledger: WorkbenchGenerationProjectionResultV1[LedgerWorkspaceProjectionV1]
declarations: WorkbenchGenerationProjectionResultV1[DeclarationsWorkspaceProjectionV1]
declarations_calendar: WorkbenchGenerationProjectionResultV1[DeclarationsCalendarProjectionV1]
aeat_sync: WorkbenchGenerationProjectionResultV1[AeatSyncWorkspaceProjectionV1]
modelo: WorkbenchGenerationProjectionResultV1[tuple[ModeloWorkspaceProjectionV1, ...]]
search: WorkbenchGenerationProjectionResultV1[InstalledWorkbenchSearchSnapshotV1]
ledger_admission: WorkbenchDestinationAdmission
declarations_admission: WorkbenchDestinationAdmission
aeat_sync_admission: WorkbenchDestinationAdmission
[docs]
class WorkbenchGenerationReadDoorV1(Protocol):
"""Injected child-composition door returning one already-loaded generation."""
[docs]
class ProfileRecordReadRepositoryV1(Protocol):
"""Narrow read-only door for the current encrypted profile record."""
[docs]
def load(self, profile_id: str) -> UserProfileRecord:
"""Load the record served by the already-open custody session."""
...
[docs]
@dataclass(frozen=True, slots=True)
class SecureProfileWorkbenchGenerationReadDoorV1:
"""Read one generation from explicit secure profile repositories.
Every repository is session-bound by the child composition root. The door
loads each authority once, projects only safe application DTOs, and marks
authorities lacking an installed-session reader as unavailable instead of
manufacturing empty fixtures.
"""
profile_id: str
operation: PinnedAuthorityOperation
profile_repository: ProfileRecordReadRepositoryV1
work_unit_repository: WorkUnitCatalogueRepositoryProtocol
calculation_repository: CalculationRevisionCatalogueRepositoryProtocol
filing_repository: ModeloRecordCatalogueRepositoryProtocol
clock: Callable[[], UtcInstant]
account_session_reader: Callable[[], HomeAccountSession]
transaction_repository: TransactionCatalogueRepositoryProtocol | None = None
invoice_repository: InvoiceCatalogueRepositoryProtocol | None = None
bucket_event_repository: BucketEventHistoryRepositoryProtocol | None = None
verification_repository: VerificationReportCatalogueRepositoryProtocol | None = None
notification_custody_reader: Callable[[], int] | None = None
"""Counts the notification documents this profile already holds locally.
Unbound means this session did not read the store, which AEAT Sync reports
as a composition gap -- distinct from a count of zero, which is a proven
zero the operator can act on.
"""
result_casilla_reader: DeclarationResultCasillaReaderV1 | None = None
"""Names the casilla that settles one modelo revision, or nothing.
Injected rather than resolved in the door, which holds repositories and no
registry access. An unbound reader leaves every declaration's result
unknown, which is the same honest absence a modelo with no declared
settlement role produces -- not a zero.
"""
operation_contracts: OperationPublicContractSetV1 | None = None
modelo_projection_reader: Callable[[WorkUnit], ModeloWorkspaceProjectionV1] | None = None
ledger_action_ports: LedgerActionPorts | None = None
"""Outer-composed ledger ports for this profile, when the ledger is bound."""
capture_memory: WorkbenchCaptureMemory | None = None
"""The session's reusable capture work; every capture recomputes everything without one."""
"""An absent reader below is a composition fact, not a data fact.
A host that did not bind a ledger store or an operation contract set
cannot observe those authorities, and the generation says so with an
explicit refusal rather than publishing an empty workspace that would be
indistinguishable from a profile holding nothing.
"""
def _capture_is_unchanged(
self,
*,
record: UserProfileRecord,
work_units_revision: str,
calculations_revision: str,
filings_revision: str,
ledger_revision: tuple[str, str] | None,
ledger_sources: tuple[TransactionCatalogue, InvoiceCatalogue] | None,
verification: VerificationReportCatalogue | None,
custody_count: int | None,
) -> bool:
final_record = self.profile_repository.load(self.profile_id)
_, final_work_units_revision = self.work_unit_repository.load_revisioned()
_, final_calculations_revision = self.calculation_repository.load_revisioned()
_, final_filings_revision = self.filing_repository.load_revisioned()
return (
final_record.content_digest == record.content_digest
and final_work_units_revision == work_units_revision
and final_calculations_revision == calculations_revision
and final_filings_revision == filings_revision
and self._ledger_is_unchanged(ledger_revision, ledger_sources)
and self._load_verification_reports() == verification
and self._load_custody_count() == custody_count
)
def _ledger_revision(self) -> tuple[str, str] | None:
"""State both ledger stores' revisions without decrypting them, when both can.
Read before the catalogues themselves, so a write landing between the
two reads shows up as a changed revision at the close of the capture.
``None`` means a store cannot state one, and the close compares the
decoded catalogues instead.
"""
ports = self.ledger_action_ports
if ports is None:
return None
transactions, invoices = ports.transaction_repository, ports.invoice_repository
if not isinstance(transactions, RevisionedCatalogueStore) or not isinstance(invoices, RevisionedCatalogueStore):
return None
transaction_revision, invoice_revision = transactions.load_revision(), invoices.load_revision()
if transaction_revision is None or invoice_revision is None:
return None
return transaction_revision, invoice_revision
def _ledger_is_unchanged(
self,
revision: tuple[str, str] | None,
sources: tuple[TransactionCatalogue, InvoiceCatalogue] | None,
) -> bool:
if revision is not None:
return self._ledger_revision() == revision
return self._load_ledger_sources() == sources
def _load_custody_count(self) -> int | None:
"""Count documents in local custody, or nothing when no reader is bound.
Read inside the capture window and re-read at its close like every
other source: a document landing mid-capture would otherwise let AEAT
Sync publish a count that was never true at any single instant.
"""
if self.notification_custody_reader is None:
return None
return self.notification_custody_reader()
def _load_verification_reports(self) -> VerificationReportCatalogue | None:
"""Read the verification catalogue, or nothing when no host bound one.
Loaded inside the capture window and re-read at its close like every
other source, because a report that lands mid-capture would let Home
offer work against a blocker that no longer exists.
"""
if self.verification_repository is None:
return None
return require_verification_report_coordinates_current(
self.verification_repository.load(),
operation=self.operation,
)
def _load_ledger_sources(self) -> tuple[TransactionCatalogue, InvoiceCatalogue] | None:
"""Read the ledger stores once, as the value the guard compares.
Neither store exposes a revision handle the way the work-unit,
calculation and filing catalogues do, so the snapshot itself is the
identity: an equal pair means nothing was written between the two
reads. Bucket events are deliberately outside it -- they only supply
review context and have no whole-catalogue read to compare.
"""
if self.ledger_action_ports is None:
return None
return (
self.ledger_action_ports.transaction_repository.load(),
self.ledger_action_ports.invoice_repository.load(),
)
def _read_ledger(
self,
calculation_revisions: Mapping[str, CalculationRevision],
work_units: WorkUnitCatalogue,
*,
sources: tuple[TransactionCatalogue, InvoiceCatalogue] | None,
ports: LedgerActionPorts,
) -> LedgerWorkspaceProjectionV1 | None:
"""Project the Ledger workspace only when its stores were bound."""
if sources is None:
return None
from .ledger.workspace_reader import read_ledger_workspace_projection
return read_ledger_workspace_projection(
bucket_id=self.profile_id,
ports=ports,
calculation_revisions=calculation_revisions,
work_units=work_units,
transactions=sources[0],
invoices=sources[1],
)
def _read_modelo(self, work_units: WorkUnitCatalogue) -> tuple[ModeloWorkspaceProjectionV1, ...] | None:
"""Project every current work unit, or refuse the whole Modelo source.
A profile holding no work yields an empty tuple, which is a proven
empty portfolio rather than an unread one.
A unit the bundled registry cannot inspect refuses the SOURCE, not the
session: a partial tuple would silently omit a declaration the profile
holds, and letting the failure escape would take Home, Ledger,
Declarations and AEAT Sync down with it for one unsupported modelo.
"""
if self.modelo_projection_reader is None:
return None
reader = self.modelo_projection_reader
try:
return tuple(reader(unit) for unit in work_units.values())
except (ValueError, LookupError):
return None
def _read_aeat_sync(
self,
subject_key: str | None,
*,
observed_at: UtcInstant,
filings: tuple[ModeloRecord, ...],
custody_count: int | None,
censo_values: Mapping[str, object],
) -> tuple[AeatSyncWorkspaceProjectionV1 | None, NamespacedId]:
"""Project the pre-pull AEAT Sync workspace against composed contracts.
A profile carrying no NIF has no subject to scope AEAT evidence to.
Scoping it to the schema's placeholder would produce a workspace whose
rows a later real pull would refuse as a mixed subject, so the source
stays unavailable until the profile declares its identity.
A malformed local projection is likewise a source refusal. The
workspace projector remains the boundary that detects it, while this
installed-session reader prevents one rejected AEAT Sync projection
from preventing Home and the other workbench sources from rendering.
Its projector-specific refusal stays distinct from the missing-reader
refusal so downstream source, admission, and search semantics retain
the actual cause.
"""
if self.operation_contracts is None or subject_key is None:
return None, _AEAT_SYNC_READER_UNAVAILABLE
from .aeat_sync.workspace_reader import read_local_aeat_sync_workspace_projection
try:
return (
read_local_aeat_sync_workspace_projection(
bucket_id=self.profile_id,
subject_key=subject_key,
observed_at=observed_at,
filings=filings,
operation_contracts=self.operation_contracts,
custody_count=custody_count,
censo_values={key: value for key, value in censo_values.items() if isinstance(value, str)},
),
_AEAT_SYNC_READER_UNAVAILABLE,
)
except (AeatSyncWorkspaceProjectionError, ValidationError):
return None, _AEAT_SYNC_SNAPSHOT_PROJECTOR_UNAVAILABLE
def _declared_tax_id(raw_values: Mapping[str, object]) -> str | None:
"""Return the profile's own NIF, never the schema's placeholder default."""
declared = raw_values.get("identity.tax_id")
if not isinstance(declared, str) or not declared.strip():
return None
return declared.strip()
def _declarations_observation(
zone: DeclarationsWorkspaceZone,
observed_at: UtcInstant,
) -> DeclarationsWorkspaceZoneObservationV1:
return DeclarationsWorkspaceZoneObservationV1(
zone=zone,
availability=DeclarationsWorkspaceAvailability.AVAILABLE,
observed_at=observed_at,
)
def _scope_filing_records(
filing_records: tuple[ModeloRecord, ...],
schedule_calendar: OverviewCalendar,
) -> tuple[ModeloRecord, ...]:
"""Keep only evidence addressed by this legal-calendar query.
The local source remains observed and available even when none of its
profile-wide filing records belongs to the requested window. That is a
measured empty query result, not an invented empty authority.
"""
addresses = {
(entry.modelo, entry.period.filing_year, entry.period.registry_token) for entry in schedule_calendar.entries
}
return tuple(
record
for record in filing_records
if (str(record.modelo), record.filing_year, record.period.registry_token) in addresses
)
def _schedule_observation(
calendar: OverviewCalendar,
observed_at: UtcInstant,
) -> DeclarationsCalendarSourceObservationV1:
if not calendar.taxpayer_model_declared:
return DeclarationsCalendarSourceObservationV1(
source=DeclarationsCalendarSource.SCHEDULE,
availability=HomeAvailability.UNAVAILABLE,
reason_code=_TAXPAYER_MODEL_UNDECLARED,
)
return DeclarationsCalendarSourceObservationV1(
source=DeclarationsCalendarSource.SCHEDULE,
availability=HomeAvailability.AVAILABLE,
observed_at=observed_at,
)
_TAXPAYER_PROFILE_INCOMPLETE: Final = "workbench.home.taxpayer_profile_incomplete"
_TAXPAYER_MODEL_UNDECLARED: Final = "workbench.calendar.taxpayer_model_undeclared"
@dataclass(frozen=True, slots=True)
class _WorkbenchCalendarInputs:
"""The calendar-derived Home and Declarations inputs, or why there are none.
``refusal`` is set exactly when the calendar could not be built at all;
``agenda_refusal`` is set whenever the agenda cannot be offered, which also
covers a calendar that was built but has no taxpayer model to schedule.
"""
agenda_evidence_state: HomeZoneState
declarations_calendar: DeclarationsCalendarProjectionV1
agenda: OverviewAgenda | None
agenda_refusal: HomeZoneState | None
refusal: HomeZoneState | None
def _taxpayer_profile_refusal(
raw_values: Mapping[str, str],
*,
operation: PinnedAuthorityOperation,
) -> HomeZoneState:
"""Name the profile paths the operator still owes, as one Home zone state.
The paths come from the profile completeness contract, which the manager
reads too, so the zone points the operator at exactly the fields the
Profile destination lists as required.
"""
from .user_profile.completeness import conditional_profile_missing_required, missing_required_field_paths
missing = (
*missing_required_field_paths(operation.profile_schema(), raw_values),
*conditional_profile_missing_required(raw_values),
)
return HomeZoneState(
availability=HomeAvailability.UNAVAILABLE,
reason_code=_TAXPAYER_PROFILE_INCOMPLETE,
missing_profile_paths=tuple(dict.fromkeys(missing)),
)
[docs]
@runtime_checkable
class RevisionedCatalogueStore(Protocol):
"""A store that can state a revision of its whole catalogue without decrypting it.
The revision changes with every committed write; ``None`` means the store
cannot state one right now.
"""
[docs]
def load_revision(self) -> str | None:
"""Return the current catalogue revision, or ``None`` when it cannot be stated."""
...
@dataclass(frozen=True, slots=True)
class _CalendarMemo:
"""Where one capture may reuse its calendar work, and under which input identity."""
memory: WorkbenchCaptureMemory | None
key: WorkbenchCalendarMemoKey
def reuse(self, build: Callable[[], WorkbenchCalendarWork]) -> WorkbenchCalendarWork:
return build() if self.memory is None else self.memory.calendar.reuse(self.key, build)
def _read_workbench_calendar_inputs(
*,
record: UserProfileRecord,
raw_values: Mapping[str, str],
as_of: date,
work_units: WorkUnitCatalogue,
filings: ModeloRecordCatalogue,
observed_at: UtcInstant,
operation: PinnedAuthorityOperation,
memo: _CalendarMemo,
) -> _WorkbenchCalendarInputs:
"""Project the taxpayer once and build every calendar-derived input from it.
A profile the taxpayer projection refuses is an incomplete profile, not a
broken session: the operator reached it by editing a field. The refusal is
published on the zones that need the projection, and every other zone of
the generation is served, rather than the whole workbench failing to start.
"""
try:
taxpayer = projection_for_taxpayer(
record,
tax_id_default="00000000T",
schema=operation.profile_schema(),
)
except (ProfileError, UserProfileValidationError):
refusal = _taxpayer_profile_refusal(raw_values, operation=operation)
return _WorkbenchCalendarInputs(
agenda_evidence_state=refusal,
declarations_calendar=_refused_declarations_calendar(refusal, as_of=as_of, observed_at=observed_at),
agenda=None,
agenda_refusal=refusal,
refusal=refusal,
)
evidence, declarations_calendar, agenda, model_declared = _build_workbench_calendar_inputs(
taxpayer=taxpayer,
raw_values=raw_values,
as_of=as_of,
work_units=work_units,
filings=filings,
observed_at=observed_at,
operation=operation,
memo=memo,
)
if not model_declared:
# The schedule observation already says why the calendar is empty; the
# agenda derived from the same missing model must say so too rather
# than read as a verified absence of dates.
return _WorkbenchCalendarInputs(
agenda_evidence_state=evidence.aeat_state,
declarations_calendar=declarations_calendar,
agenda=None,
agenda_refusal=HomeZoneState(
availability=HomeAvailability.UNAVAILABLE,
reason_code=_TAXPAYER_MODEL_UNDECLARED,
),
refusal=None,
)
return _WorkbenchCalendarInputs(
agenda_evidence_state=evidence.aeat_state,
declarations_calendar=declarations_calendar,
agenda=agenda,
agenda_refusal=None,
refusal=None,
)
def _refused_declarations_calendar(
refusal: HomeZoneState,
*,
as_of: date,
observed_at: UtcInstant,
) -> DeclarationsCalendarProjectionV1:
"""Publish a Declarations calendar that states why it has no schedule.
Declarations stays admitted when only the calendar is refused, and an
admitted Declarations destination always carries a calendar. The honest
calendar here is an empty one whose every source says why it is empty.
"""
reason_code = refusal.reason_code
if reason_code is None:
raise InternalInvariantError("a refused Declarations calendar requires a reason")
unavailable = HomeZoneState(availability=HomeAvailability.UNAVAILABLE, reason_code=reason_code)
evidence = build_calendar_evidence_projection(
local=CalendarEvidenceReadOutcome[LocalCalendarEvidenceSources](state=unavailable),
aeat=CalendarEvidenceReadOutcome[AeatCalendarEvidenceSources](state=unavailable),
)
calendar = OverviewCalendar(
range=_calendar_query_range(as_of),
entries=(),
generated_at=observed_at,
taxpayer_model_declared=False,
)
return project_declarations_calendar(
calendar=calendar,
evidence=evidence,
as_of=as_of,
schedule_observation=DeclarationsCalendarSourceObservationV1(
source=DeclarationsCalendarSource.SCHEDULE,
availability=HomeAvailability.UNAVAILABLE,
reason_code=reason_code,
),
)
def _calendar_query_range(as_of: date) -> OverviewCalendarRange:
"""The calendar year the workbench schedules, containing ``as_of``."""
return OverviewCalendarRange(
from_date=date(as_of.year, 1, 1),
to_date=date(as_of.year, 12, 31),
)
def _build_workbench_calendar_inputs(
*,
taxpayer: TaxpayerProfile,
raw_values: Mapping[str, object],
as_of: date,
work_units: WorkUnitCatalogue,
filings: ModeloRecordCatalogue,
observed_at: UtcInstant,
operation: PinnedAuthorityOperation,
memo: _CalendarMemo,
) -> tuple[CalendarEvidenceProjection, DeclarationsCalendarProjectionV1, OverviewAgenda, bool]:
query_range = _calendar_query_range(as_of)
def evidence_for(schedule_calendar: OverviewCalendar) -> CalendarEvidenceProjection:
return build_calendar_evidence_projection(
local=CalendarEvidenceReadOutcome(
state=HomeZoneState(availability=HomeAvailability.AVAILABLE, observed_at=observed_at),
value=LocalCalendarEvidenceSources(
filing_records=_scope_filing_records(
tuple(filings.records.values()),
schedule_calendar,
),
),
),
aeat=CalendarEvidenceReadOutcome[AeatCalendarEvidenceSources](
state=HomeZoneState(
availability=HomeAvailability.NEVER_CAPTURED,
reason_code="workbench.calendar.aeat_reader_unavailable",
),
),
expected_tax_id=taxpayer.tax_id,
)
def compute() -> WorkbenchCalendarWork:
schedule_calendar = build_overview_calendar(
taxpayer,
query_range,
today=as_of,
raw_values=raw_values,
work_units=tuple(work_units.values()),
operation=operation,
)
return WorkbenchCalendarWork(
schedule_calendar=schedule_calendar,
calendar=build_overview_calendar(
taxpayer,
query_range,
today=as_of,
raw_values=raw_values,
filing_evidence=evidence_for(schedule_calendar).evidence,
work_units=tuple(work_units.values()),
operation=operation,
),
agenda=build_overview_agenda(taxpayer, as_of=as_of, raw_values=raw_values, operation=operation),
)
work = memo.reuse(compute)
# The evidence rows depend only on the memo key's inputs; their observation
# state is this capture's own, so the projection is rebuilt every time.
evidence = evidence_for(work.schedule_calendar)
declarations_calendar = project_declarations_calendar(
calendar=work.calendar,
evidence=evidence,
as_of=as_of,
schedule_observation=_schedule_observation(work.calendar, observed_at),
)
return evidence, declarations_calendar, work.agenda, work.calendar.taxpayer_model_declared
def _secure_profile_home_input(
*,
observed_at: UtcInstant,
account_session: HomeAccountSession,
agenda: OverviewAgenda | None,
agenda_evidence_state: HomeZoneState,
ledger: LedgerWorkspaceProjectionV1 | None,
declarations: tuple[HomeDeclarationResume, ...] | None,
blocked_revision_ids: frozenset[str] = frozenset(),
agenda_refusal: HomeZoneState | None = None,
) -> HomeProjectionInput:
"""Assemble Home from the authorities this session actually read.
Each zone carries its own state. A zone with no installed reader stays
UNAVAILABLE and names what is missing, because a Home that renders an
unread authority as an empty list is indistinguishable from one whose
operator genuinely has nothing outstanding.
"""
def unavailable(reason_code: str) -> HomeZoneState:
return HomeZoneState(availability=HomeAvailability.UNAVAILABLE, reason_code=reason_code)
def never_captured(reason_code: str) -> HomeZoneState:
"""A zone whose source is readable but has never been captured.
Distinct from `unavailable`, which says the reader cannot answer. An
operator reading "unavailable" looks for something broken; one reading
"never captured" knows the data is simply not here yet and that a pull
is what produces it. Collapsing the two hides the only action that
would resolve the zone.
"""
return HomeZoneState(availability=HomeAvailability.NEVER_CAPTURED, reason_code=reason_code)
declarations_state = (
HomeZoneState(availability=HomeAvailability.AVAILABLE, observed_at=observed_at)
if declarations is not None
else unavailable("workbench.home.declarations_resume_projector_unavailable")
)
addressed = _home_declaration_actions(declarations, blocked_revision_ids)
cross_cutting = _home_ledger_actions(ledger)
actions = (
None
if cross_cutting is None
else (
*addressed,
*(item.model_copy(update={"rank": len(addressed) + offset}) for offset, item in enumerate(cross_cutting)),
)
)
actions_state = (
HomeZoneState(availability=HomeAvailability.AVAILABLE, observed_at=observed_at)
if actions is not None
else unavailable("workbench.home.actions_projector_unavailable")
)
readiness = _home_ledger_readiness(ledger)
ledger_state = (
HomeZoneState(availability=HomeAvailability.AVAILABLE, observed_at=observed_at)
if readiness is not None
else unavailable("workbench.ledger.snapshot_projector_unavailable")
)
return HomeProjectionInput(
generated_at=observed_at,
account=account_session,
actions_state=actions_state,
actions=actions or (),
declarations_state=declarations_state,
declarations=declarations or (),
ledger_state=ledger_state,
agenda_state=(
HomeZoneState(availability=HomeAvailability.AVAILABLE, observed_at=observed_at)
if agenda is not None
else agenda_refusal or unavailable("workbench.home.agenda_projector_unavailable")
),
ledger_readiness=readiness,
overview_agenda=agenda,
# The AEAT side of the agenda is whatever the evidence read concluded,
# not a separate refusal: before any pull it is NEVER CAPTURED, and
# Home must say that rather than call the whole zone unavailable.
agenda_evidence_state=agenda_evidence_state,
# AEAT notifications exist only once a pull has persisted a snapshot, so
# before that the reader is fine and the DATA is absent. The previous
# code called this reader-unavailable, which named the wrong thing and
# pointed the operator at a fault that does not exist.
messages_state=never_captured("workbench.home.messages_never_pulled"),
)
_HOME_LEDGER_AREAS: Final = (
(LedgerWorkspaceArea.ENTRIES, "entries"),
(LedgerWorkspaceArea.REVIEW, "requiring_review"),
(LedgerWorkspaceArea.CLASSIFICATION, "unclassified"),
(LedgerWorkspaceArea.EVIDENCE, "missing_evidence"),
)
def _home_ledger_readiness(ledger: LedgerWorkspaceProjectionV1 | None) -> HomeLedgerReadiness | None:
"""Home's four Ledger counts, or nothing when any of them was not measured.
`item_count` is always an integer, so a zero alone cannot be trusted: an
area whose status is UNMEASURED reports zero for "nobody looked", the same
value a genuinely empty area reports for "I looked and there was nothing".
Home renders these as bare numbers with no room to qualify them, so an
unmeasured area makes the whole readiness block refuse and name itself
rather than publish a zero the operator would read as a finding.
"""
if ledger is None:
return None
by_area = {state.area: state for state in ledger.areas}
counts: dict[str, int] = {}
for area, field in _HOME_LEDGER_AREAS:
state = by_area.get(area)
if state is None or state.status is LedgerWorkspaceStatus.UNMEASURED:
return None
counts[field] = state.item_count
return HomeLedgerReadiness(**counts)
_HOME_DECLARATION_STATES: Final[dict[CalculationRevisionState, HomeDeclarationState]] = {
# Read from the domain's own vocabulary, not invented for Home.
# `VERIFICADO_COMPLETO` says verified and complete, which is the only thing
# READY can honestly mean; `BORRADOR` is a calculation that exists and has
# not been verified, which is exactly NEEDS_REVIEW; both PRESENTADO forms
# are filed, superseded or not.
CalculationRevisionState.BORRADOR: HomeDeclarationState.NEEDS_REVIEW,
CalculationRevisionState.VERIFICADO_COMPLETO: HomeDeclarationState.READY,
CalculationRevisionState.PRESENTADO: HomeDeclarationState.FILED,
CalculationRevisionState.PRESENTADO_SUPERSEDIDO: HomeDeclarationState.FILED,
CalculationRevisionState.DESCARTADO: HomeDeclarationState.DISCARDED,
}
"""How a calculation's state reads on Home.
The mapping ERRS SAFE in the one direction that matters: nothing becomes READY
except the state that literally says verified and complete. Telling an operator
a declaration is ready to file when nobody verified it is a filing-grade harm,
so every other state resolves to something that keeps work in front of them.
"""
def _home_declarations(
declarations: DeclarationsWorkspaceProjectionV1 | None,
work_units: WorkUnitCatalogue,
) -> tuple[HomeDeclarationResume, ...] | None:
"""Home's resumable declarations, or nothing when the workspace is refused."""
if declarations is None:
return None
current_by_unit = {
revision.work_unit_id: revision for revision in declarations.calculation_revisions if revision.is_current
}
resumes: list[HomeDeclarationResume] = []
for ref in declarations.declarations:
unit = work_units.work_units.get(ref.work_unit_id)
if unit is None:
# The declaration names a work unit this session did not load. Its
# display name lives on the unit, so the row cannot be rendered
# honestly and the zone refuses rather than inventing a label.
return None
revision = current_by_unit.get(ref.work_unit_id)
if ref.state is WorkUnitState.DESCARTADO:
state = HomeDeclarationState.DISCARDED
elif ref.has_current_filing:
state = HomeDeclarationState.FILED
elif revision is None:
# No calculation exists yet, so there is nothing to review or file.
state = HomeDeclarationState.DRAFT
else:
state = _HOME_DECLARATION_STATES[revision.state]
resumes.append(
HomeDeclarationResume(
work_unit_id=ref.work_unit_id,
modelo=ref.modelo,
filing_year=ref.filing_year,
period=ref.period,
name=unit.name,
state=state,
revision_id=revision.calculation_revision_id if revision is not None else None,
)
)
return tuple(resumes)
_HOME_LEDGER_ACTIONS: Final = (
(LedgerWorkspaceArea.CLASSIFICATION, "operator.ledger.classify", HOME_REASON_LEDGER_CLASSIFICATION_PENDING),
(LedgerWorkspaceArea.EVIDENCE, "operator.ledger.evidence.review.list", HOME_REASON_EVIDENCE_MISSING),
)
"""Cross-cutting Ledger work Home can offer, in the order it is offered."""
def _dependency_blocked_revisions(
verification: VerificationReportCatalogue | None,
) -> frozenset[str]:
"""Calculation revisions blocked by an unclean cross-period dependency.
ONE finding kind is read, deliberately. `CROSS_PERIOD_DEPENDENCY_UNCLEAN`
is `blocked_dependency` by its own name, so the reason Home renders is the
domain's own word for the condition. The other two blocked reason codes
Home declares are left unproduced: nothing in
`ModeloVerificationFindingKind` names evidence, and routing
`blocked_review` to BLOCKING_RULE or MISSING_REQUIRED_CASILLA would be a
guess dressed as a finding.
Only BLOCKING severity counts, and only on a report whose completeness is
BLOCKED. An advisory finding of the same kind is information, not a
blocker, and offering it as one would put work in front of an operator that
nothing is actually waiting on.
"""
if verification is None:
return frozenset[str]()
return frozenset(
report.calculation_revision_id
for report in verification.reports.values()
if report.completeness_status is VerificationCompletenessStatus.BLOCKED
and any(
finding.kind is ModeloVerificationFindingKind.CROSS_PERIOD_DEPENDENCY_UNCLEAN
and finding.severity is ModeloVerificationFindingSeverity.BLOCKING
for finding in report.findings
)
)
def _home_declaration_actions(
resumes: tuple[HomeDeclarationResume, ...] | None,
blocked_revision_ids: frozenset[str] = frozenset(),
) -> tuple[HomeNextAction, ...]:
"""One action per declaration whose calculation is not yet verified.
`operator.modelo.work.revisions` takes the work unit id the resume already
carries, and listing a work unit's calculation revisions is precisely what
`declaration_needs_review` asks the operator to do -- so the action is the
catalogue's, the reason is Home's, and the address is the declaration's.
Nothing here is minted for the zone.
Ranked ahead of the cross-cutting Ledger offers because these name a single
declaration the operator can finish, while "classify the ledger" is work
spread across every record.
"""
if resumes is None:
return ()
offered: list[HomeNextAction] = []
for resume in resumes:
blocked = resume.revision_id is not None and resume.revision_id in blocked_revision_ids
if not blocked and resume.state is not HomeDeclarationState.NEEDS_REVIEW:
continue
offered.append(
HomeNextAction(
rank=len(offered),
# A blocked declaration is named by its blocker rather than by
# the generic review prompt: "a dependency is blocked" tells
# the operator why the work will not close, where "needs
# review" invites them to try and find out.
reason_code=(HOME_REASON_BLOCKED_DEPENDENCY if blocked else HOME_REASON_DECLARATION_NEEDS_REVIEW),
action=declare_next_action("operator.modelo.work.revisions", work_unit_id=resume.work_unit_id),
modelo=resume.modelo,
filing_year=resume.filing_year,
period=resume.period,
)
)
return tuple(offered)
def _home_ledger_actions(ledger: LedgerWorkspaceProjectionV1 | None) -> tuple[HomeNextAction, ...] | None:
"""Offer the Ledger work that is outstanding, or nothing when unmeasured.
An UNMEASURED area yields no action rather than an action for zero items:
`item_count` is a plain integer, so an unmeasured area reports the same
zero a finished one does, and offering "classify 0 entries" is worse than
offering nothing. The zone refuses as a whole under the same rule the
readiness block uses.
"""
if ledger is None:
return None
by_area = {state.area: state for state in ledger.areas}
actions: list[HomeNextAction] = []
for area, action_id, reason_code in _HOME_LEDGER_ACTIONS:
state = by_area.get(area)
if state is None or state.status is LedgerWorkspaceStatus.UNMEASURED:
return None
if state.item_count:
actions.append(
HomeNextAction(
rank=len(actions),
action=declare_next_action(action_id),
reason_code=reason_code,
)
)
return tuple(actions)
def _generation_admission(
destination: str,
state: WorkbenchDestinationAdmissionState,
*,
reason_code: str | None = None,
) -> WorkbenchDestinationAdmission:
return WorkbenchDestinationAdmission(destination=destination, state=state, reason_code=reason_code)
def _source_result[SourceT](
value: SourceT | None,
*,
observed_at: UtcInstant,
refusal: NamespacedId,
) -> WorkbenchGenerationSourceResultV1[SourceT]:
if value is None:
return WorkbenchGenerationSourceResultV1[SourceT].unavailable(refusal=refusal)
return WorkbenchGenerationSourceResultV1[SourceT].available(value, observed_at=observed_at)
def _source_admission(
destination: str,
value: object | None,
*,
unavailable_reason: str,
) -> WorkbenchDestinationAdmission:
if value is None:
return _generation_admission(
destination,
WorkbenchDestinationAdmissionState.UNAVAILABLE,
reason_code=unavailable_reason,
)
return _generation_admission(destination, WorkbenchDestinationAdmissionState.AVAILABLE)
def _build_workbench_generation_inputs(
*,
observed_at: UtcInstant,
account_session: HomeAccountSession,
calendar_inputs: _WorkbenchCalendarInputs,
ledger: LedgerWorkspaceProjectionV1 | None,
declarations: DeclarationsWorkspaceProjectionV1 | None,
aeat_sync: AeatSyncWorkspaceProjectionV1 | None,
aeat_sync_refusal: NamespacedId,
modelo: tuple[ModeloWorkspaceProjectionV1, ...] | None,
work_units: WorkUnitCatalogue,
verification: VerificationReportCatalogue | None,
) -> WorkbenchGenerationInputsV1:
return WorkbenchGenerationInputsV1(
assembled_at=observed_at,
home=WorkbenchGenerationSourceResultV1[HomeProjectionInput].available(
_secure_profile_home_input(
observed_at=observed_at,
account_session=account_session,
agenda=calendar_inputs.agenda,
agenda_evidence_state=calendar_inputs.agenda_evidence_state,
ledger=ledger,
declarations=_home_declarations(declarations, work_units),
blocked_revision_ids=_dependency_blocked_revisions(verification),
agenda_refusal=calendar_inputs.agenda_refusal,
),
observed_at=observed_at,
),
ledger=_source_result(
ledger,
observed_at=observed_at,
refusal="workbench.ledger.snapshot_projector_unavailable",
),
declarations=_source_result(
declarations,
observed_at=observed_at,
refusal="workbench.declarations.snapshot_projector_unavailable",
),
declarations_calendar=WorkbenchGenerationSourceResultV1[DeclarationsCalendarProjectionV1].available(
calendar_inputs.declarations_calendar,
observed_at=observed_at,
),
aeat_sync=_source_result(aeat_sync, observed_at=observed_at, refusal=aeat_sync_refusal),
modelo=_source_result(
modelo,
observed_at=observed_at,
refusal="workbench.modelo.bulk_reader_unavailable",
),
ledger_admission=_source_admission(
"workbench.ledger",
ledger,
unavailable_reason="workbench.ledger.snapshot_projector_unavailable",
),
declarations_admission=_source_admission(
"workbench.declarations",
declarations,
unavailable_reason="workbench.declarations.snapshot_projector_unavailable",
),
aeat_sync_admission=_source_admission(
"workbench.aeat_sync",
aeat_sync,
unavailable_reason=aeat_sync_refusal,
),
)
[docs]
@dataclass(frozen=True, slots=True)
class InstalledWorkbenchGenerationProviderV1:
"""Child-owned provider for one immutable installed-session generation.
The read door is composed with secure repositories and application
projectors by the process owner. Calling the provider captures that door
once and immediately discards its source bundle after projection.
"""
read_door: WorkbenchGenerationReadDoorV1
def __call__(self) -> WorkbenchGenerationV1:
"""Capture and assemble exactly one coherent generation."""
return assemble_workbench_generation_from(self.read_door)
[docs]
def assemble_workbench_generation(inputs: WorkbenchGenerationInputsV1) -> WorkbenchGenerationV1:
"""Build one immutable generation from already-loaded public inputs.
No area is changed into a known empty projection when its source result is
missing. Search is assembled only when every required projection exists;
otherwise its own result preserves the strongest truthful source refusal.
"""
home = _project_home(inputs.home)
ledger = _carry_projection(inputs.ledger)
declarations = _carry_projection(inputs.declarations)
declarations_calendar = _carry_projection(inputs.declarations_calendar)
aeat_sync = _carry_projection(inputs.aeat_sync)
modelo = _carry_projection(inputs.modelo)
search = _assemble_search(
ledger=ledger,
declarations=declarations,
aeat_sync=aeat_sync,
modelo=modelo,
ledger_admission=inputs.ledger_admission,
declarations_admission=inputs.declarations_admission,
aeat_sync_admission=inputs.aeat_sync_admission,
)
return WorkbenchGenerationV1(
assembled_at=inputs.assembled_at,
home=home,
ledger=ledger,
declarations=declarations,
declarations_calendar=declarations_calendar,
aeat_sync=aeat_sync,
modelo=modelo,
search=search,
ledger_admission=inputs.ledger_admission,
declarations_admission=inputs.declarations_admission,
aeat_sync_admission=inputs.aeat_sync_admission,
)
[docs]
def assemble_workbench_generation_from(read_door: WorkbenchGenerationReadDoorV1) -> WorkbenchGenerationV1:
"""Assemble one generation by invoking an injected read door exactly once."""
return assemble_workbench_generation(read_door.read_workbench_generation_inputs())
def _project_home(
source: WorkbenchGenerationSourceResultV1[HomeProjectionInput],
) -> WorkbenchGenerationProjectionResultV1[HomeProjectionV1]:
if source.value is None:
return WorkbenchGenerationProjectionResultV1(
availability=source.availability,
observed_at=source.observed_at,
refusal=source.refusal,
)
projection = compose_home_projection(source.value)
return WorkbenchGenerationProjectionResultV1(
availability=source.availability,
observed_at=source.observed_at,
refusal=source.refusal,
projection=projection,
)
def _carry_projection[ProjectionT](
source: WorkbenchGenerationSourceResultV1[ProjectionT],
) -> WorkbenchGenerationProjectionResultV1[ProjectionT]:
"""Copy an already-built safe projection without retaining input wrappers."""
return WorkbenchGenerationProjectionResultV1(
availability=source.availability,
observed_at=source.observed_at,
refusal=source.refusal,
projection=source.value,
)
def _search_projections(
sources: tuple[
WorkbenchGenerationProjectionResultV1[LedgerWorkspaceProjectionV1],
WorkbenchGenerationProjectionResultV1[DeclarationsWorkspaceProjectionV1],
WorkbenchGenerationProjectionResultV1[AeatSyncWorkspaceProjectionV1],
WorkbenchGenerationProjectionResultV1[tuple[ModeloWorkspaceProjectionV1, ...]],
],
) -> (
tuple[
LedgerWorkspaceProjectionV1,
DeclarationsWorkspaceProjectionV1,
AeatSyncWorkspaceProjectionV1,
tuple[ModeloWorkspaceProjectionV1, ...],
]
| None
):
if any(source.projection is None for source in sources):
return None
ledger_projection = sources[0].projection
declarations_projection = sources[1].projection
aeat_projection = sources[2].projection
modelo_projection = sources[3].projection
# The ``None`` branch above makes these values present; the explicit
# narrowing keeps the call boundary honest for static type checkers.
if (
ledger_projection is None
or declarations_projection is None
or aeat_projection is None
or modelo_projection is None
): # pragma: no cover - guarded by the branch above
return None
return ledger_projection, declarations_projection, aeat_projection, modelo_projection
def _search_availability(
sources: tuple[
WorkbenchGenerationProjectionResultV1[LedgerWorkspaceProjectionV1],
WorkbenchGenerationProjectionResultV1[DeclarationsWorkspaceProjectionV1],
WorkbenchGenerationProjectionResultV1[AeatSyncWorkspaceProjectionV1],
WorkbenchGenerationProjectionResultV1[tuple[ModeloWorkspaceProjectionV1, ...]],
],
) -> WorkbenchGenerationAvailability:
if any(source.availability is WorkbenchGenerationAvailability.STALE for source in sources):
return WorkbenchGenerationAvailability.STALE
return WorkbenchGenerationAvailability.AVAILABLE
def _assemble_search(
*,
ledger: WorkbenchGenerationProjectionResultV1[LedgerWorkspaceProjectionV1],
declarations: WorkbenchGenerationProjectionResultV1[DeclarationsWorkspaceProjectionV1],
aeat_sync: WorkbenchGenerationProjectionResultV1[AeatSyncWorkspaceProjectionV1],
modelo: WorkbenchGenerationProjectionResultV1[tuple[ModeloWorkspaceProjectionV1, ...]],
ledger_admission: WorkbenchDestinationAdmission,
declarations_admission: WorkbenchDestinationAdmission,
aeat_sync_admission: WorkbenchDestinationAdmission,
) -> WorkbenchGenerationProjectionResultV1[InstalledWorkbenchSearchSnapshotV1]:
"""Derive search from the same source generation or preserve refusal."""
sources = (ledger, declarations, aeat_sync, modelo)
projections = _search_projections(sources)
if projections is None:
return _missing_search(sources)
ledger_projection, declarations_projection, aeat_projection, modelo_projection = projections
availability = _search_availability(sources)
observed_at = min(source.observed_at for source in sources if source.observed_at is not None)
refusal = next(
(source.refusal for source in sources if source.availability is WorkbenchGenerationAvailability.STALE),
None,
)
snapshot = assemble_installed_workbench_search_snapshot(
ledger=ledger_projection,
declarations=declarations_projection,
aeat_sync=aeat_projection,
modelo=modelo_projection,
ledger_admission=ledger_admission,
declarations_admission=declarations_admission,
aeat_sync_admission=aeat_sync_admission,
)
return WorkbenchGenerationProjectionResultV1(
availability=availability,
observed_at=observed_at,
refusal=refusal,
projection=snapshot,
)
def _missing_search(
sources: tuple[
WorkbenchGenerationProjectionResultV1[LedgerWorkspaceProjectionV1],
WorkbenchGenerationProjectionResultV1[DeclarationsWorkspaceProjectionV1],
WorkbenchGenerationProjectionResultV1[AeatSyncWorkspaceProjectionV1],
WorkbenchGenerationProjectionResultV1[tuple[ModeloWorkspaceProjectionV1, ...]],
],
) -> WorkbenchGenerationProjectionResultV1[InstalledWorkbenchSearchSnapshotV1]:
"""Collapse missing search dependencies without claiming an empty index."""
missing = tuple(source for source in sources if source.projection is None)
states = tuple(source.availability for source in missing)
if all(state is WorkbenchGenerationAvailability.NEVER_CAPTURED for state in states):
availability = WorkbenchGenerationAvailability.NEVER_CAPTURED
elif all(
state in {WorkbenchGenerationAvailability.LOCKED, WorkbenchGenerationAvailability.NEVER_CAPTURED}
for state in states
):
availability = WorkbenchGenerationAvailability.LOCKED
else:
availability = WorkbenchGenerationAvailability.UNAVAILABLE
refusal = next(source.refusal for source in missing if source.refusal is not None)
return WorkbenchGenerationProjectionResultV1(availability=availability, refusal=refusal)
__all__ = [
"InstalledWorkbenchGenerationProviderV1",
"ProfileRecordReadRepositoryV1",
"SecureProfileWorkbenchGenerationReadDoorV1",
"WorkbenchGenerationAvailability",
"WorkbenchGenerationInputsV1",
"WorkbenchGenerationProjectionResultV1",
"WorkbenchGenerationReadDoorV1",
"WorkbenchGenerationSourceResultV1",
"WorkbenchGenerationV1",
"assemble_workbench_generation",
"assemble_workbench_generation_from",
]