cadrumo.application.workbench_generation module

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: CalculationRevision, InvoiceCatalogue, ModeloRecord, TaxpayerProfile, TransactionCatalogue, UserProfileRecord.

class WorkbenchGenerationAvailability(*values)[source]

Bases: StrEnum

Truthful state of one already-captured source result.

AVAILABLE
STALE
LOCKED
NEVER_CAPTURED
UNAVAILABLE
class WorkbenchGenerationSourceResultV1(**data)[source]

Bases: BaseModel, Generic

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 WorkbenchGenerationV1.

Parameters:
availability: WorkbenchGenerationAvailability
observed_at: UtcInstant | None
refusal: NamespacedId | None
value: SourceT | None
classmethod available(value, *, observed_at)[source]

Construct a source result with a measured value.

Return type:

Self

Parameters:

  • value (SourceT)

  • observed_at (UtcInstant)

classmethod stale(value, *, observed_at, refusal)[source]

Construct a source result retaining a known but stale value.

Return type:

Self

Parameters:

  • value (SourceT)

  • observed_at (UtcInstant)

  • refusal (NamespacedId)

classmethod locked(*, refusal)[source]

Construct a source result blocked by local custody.

Return type:

Self

Parameters:

refusal (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=3, max_length=160, pattern=^[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)+$, ascii_only=None)])

classmethod never_captured(*, refusal)[source]

Construct a source result for a source not read in this session.

Return type:

Self

Parameters:

refusal (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=3, max_length=160, pattern=^[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)+$, ascii_only=None)])

classmethod unavailable(*, refusal)[source]

Construct a source result whose reader cannot currently answer.

Return type:

Self

Parameters:

refusal (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=3, max_length=160, pattern=^[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)+$, ascii_only=None)])

class WorkbenchGenerationProjectionResultV1(**data)[source]

Bases: BaseModel, Generic

One immutable safe projection plus its source availability evidence.

Parameters:
availability: WorkbenchGenerationAvailability
observed_at: UtcInstant | None
refusal: NamespacedId | None
projection: ProjectionT | None
class WorkbenchGenerationInputsV1(**data)[source]

Bases: BaseModel

One coherent set of preloaded public inputs for child composition.

Ledger, Declarations, AEAT Sync, and Modelo inputs are already safe projection results. The Home input is the existing safe composer input so this boundary can reuse compose_home_projection without accepting any repository or source adapter.

Parameters:

data (Any)

assembled_at: UtcInstant
home: WorkbenchGenerationSourceResultV1[HomeProjectionInput]
ledger: WorkbenchGenerationSourceResultV1[LedgerWorkspaceProjectionV1]
declarations: WorkbenchGenerationSourceResultV1[DeclarationsWorkspaceProjectionV1]
declarations_calendar: WorkbenchGenerationSourceResultV1[DeclarationsCalendarProjectionV1]
aeat_sync: WorkbenchGenerationSourceResultV1[AeatSyncWorkspaceProjectionV1]
modelo: WorkbenchGenerationSourceResultV1[tuple[ModeloWorkspaceProjectionV1, ...]]
ledger_admission: WorkbenchDestinationAdmission
declarations_admission: WorkbenchDestinationAdmission
aeat_sync_admission: WorkbenchDestinationAdmission
class WorkbenchGenerationV1(**data)[source]

Bases: BaseModel

Immutable public generation consumed by an installed workbench root.

Parameters:

data (Any)

contract_version: Literal[1]
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
class WorkbenchGenerationReadDoorV1(*args, **kwargs)[source]

Bases: Protocol

Injected child-composition door returning one already-loaded generation.

read_workbench_generation_inputs()[source]

Return one coherent, preloaded input set without frontend work.

Return type:

WorkbenchGenerationInputsV1

class ProfileRecordReadRepositoryV1(*args, **kwargs)[source]

Bases: Protocol

Narrow read-only door for the current encrypted profile record.

load(profile_id)[source]

Load the record served by the already-open custody session.

Return type:

UserProfileRecord

Parameters:

profile_id (str)

class SecureProfileWorkbenchGenerationReadDoorV1(profile_id, operation, profile_repository, work_unit_repository, calculation_repository, filing_repository, clock, account_session_reader, transaction_repository=None, invoice_repository=None, bucket_event_repository=None, verification_repository=None, notification_custody_reader=None, result_casilla_reader=None, operation_contracts=None, modelo_projection_reader=None, ledger_action_ports=None, capture_memory=None)[source]

Bases: object

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.

Parameters:
profile_id: str
operation: PinnedAuthorityOperation
profile_repository: ProfileRecordReadRepositoryV1
work_unit_repository: WorkUnitCatalogueRepositoryProtocol
calculation_repository: CalculationRevisionCatalogueRepositoryProtocol
filing_repository: ModeloRecordCatalogueRepositoryProtocol
clock: Callable[[], Annotated[datetime]]
account_session_reader: Callable[[], HomeAccountSession]
transaction_repository: TransactionCatalogueRepositoryProtocol | None
invoice_repository: InvoiceCatalogueRepositoryProtocol | None
bucket_event_repository: BucketEventHistoryRepositoryProtocol | None
verification_repository: VerificationReportCatalogueRepositoryProtocol | None
notification_custody_reader: Callable[[], int] | 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: Callable[[ModeloCode, int, Period], str | 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
modelo_projection_reader: Callable[[WorkUnit], ModeloWorkspaceProjectionV1] | None
ledger_action_ports: LedgerActionPorts | None

Outer-composed ledger ports for this profile, when the ledger is bound.

capture_memory: WorkbenchCaptureMemory | None

The session’s reusable capture work; every capture recomputes everything without one.

read_workbench_generation_inputs()[source]

Capture secure local facts once and build installed projections.

Return type:

WorkbenchGenerationInputsV1

class RevisionedCatalogueStore(*args, **kwargs)[source]

Bases: 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.

load_revision()[source]

Return the current catalogue revision, or None when it cannot be stated.

Return type:

str | None

class InstalledWorkbenchGenerationProviderV1(read_door)[source]

Bases: object

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.

Parameters:

read_door (WorkbenchGenerationReadDoorV1)

read_door: WorkbenchGenerationReadDoorV1
assemble_workbench_generation(inputs)[source]

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.

Return type:

WorkbenchGenerationV1

Parameters:

inputs (WorkbenchGenerationInputsV1)

assemble_workbench_generation_from(read_door)[source]

Assemble one generation by invoking an injected read door exactly once.

Return type:

WorkbenchGenerationV1

Parameters:

read_door (WorkbenchGenerationReadDoorV1)