"""Resumable censo acquisition, exact review, and cotejo application."""
from __future__ import annotations
import secrets
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from enum import StrEnum
from typing import TYPE_CHECKING, Final, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from ...core.async_cleanup import AsyncCloseable
from ...core.bucket_pointer import require_active_bucket_id
from ...core.errors.hierarchy import pydantic_validation_boundary
from ...core.hashing import content_hash_hex
from ...core.identity.digest import ContentDigest, ContentDigestOrAbsent
from ...core.identity.profile import ProfileId
from ...core.models import STRICT_FROZEN_CONFIG
from ...core.operations import (
EFFECTS_WITHOUT_PARTIAL_COMMIT,
OperationCancellation,
OperationClosePolicy,
OperationDeadline,
OperationDurability,
OperationEffect,
OperationInteractionKind,
)
from ...domain.user_profile.values import UserProfileRecord
from ..auth.certificate_secret_backend import CertificateSecretBackendFactory
from ..auth.operator_scope_ports import OperatorScopePorts
from ..auth.protocols import BrowserSessionFactoryPort
from ..live.censo_ports import CensalFetchPort
from ..operations.capabilities import (
OperationBaselinePolicy,
OperationCapabilities,
OperationConflictScope,
OperationOwnedResource,
OperationReplayPolicy,
OperationRequestStoragePolicy,
OperationSensitiveInputPolicy,
)
from ..operations.interactions import OperationResponseIntentValue
from ..operations.models import OperationRequest
from ..operations.owner import OperationExecutorContext, OperationResumeCheckpoint
from ..operations.registry import (
OperationDefinition,
OperationExecutorFactory,
OperationFrontendProjection,
OperationPublicDefinitionRegistrationV1,
OperationReconciliationPolicy,
OperationSchemaBindingV1,
operation_public_schema_reference,
)
from .capsule_record import ProfileRecordConflictError
from .censal_observation import CensalObservation
from .censo_sync import (
CENSAL_ADOPTABLE_PATHS,
CENSO_SOURCE_TAG,
censal_facts_from_read,
reconcile_censal_read,
)
from .cotejo_apply import apply_cotejo
from .profile_record_repository import ProfileRecordRepository
from .projections import record_to_effective_facts
if TYPE_CHECKING:
from ...domain.calculations.registry.authority_artifact import ProfileDecodeContext
CENSAL_OPERATION_DEFINITION_ID = "user-profile.censo-review"
CENSAL_PHASE_PREFLIGHT = "censo.preflight"
CENSAL_PHASE_CLAVE_DEVICE_WAIT = "censo.clave-device-wait"
CENSAL_PHASE_REMOTE_READ = "censo.remote-read"
CENSAL_PHASE_PROPOSAL = "censo.proposal"
CENSAL_PHASE_INTERACTION_WAIT = "censo.interaction-wait"
CENSAL_PHASE_APPLY = "censo.apply"
CENSAL_PHASE_REJECT = "censo.reject"
CENSAL_PHASE_SETTLEMENT = "censo.settlement"
_CENSAL_PHASES = (
CENSAL_PHASE_PREFLIGHT,
CENSAL_PHASE_CLAVE_DEVICE_WAIT,
CENSAL_PHASE_REMOTE_READ,
CENSAL_PHASE_PROPOSAL,
CENSAL_PHASE_INTERACTION_WAIT,
CENSAL_PHASE_APPLY,
CENSAL_PHASE_REJECT,
CENSAL_PHASE_SETTLEMENT,
)
[docs]
class CensalFieldIntent(StrEnum):
"""Closed operator intent for one reviewed censo-derived profile field."""
ADOPT = "adopt"
PRESERVE = "preserve"
[docs]
class CensalReviewedFieldIntent(BaseModel):
"""One exact field decision retained with the reviewed observation."""
model_config = STRICT_FROZEN_CONFIG
path: str = Field(min_length=3, max_length=160)
intent: CensalFieldIntent
@field_validator("path")
@classmethod
@pydantic_validation_boundary
def _require_adoptable_path(cls, value: str) -> str:
if value not in CENSAL_ADOPTABLE_PATHS:
raise ValueError("censal field intent must target a canonical adoptable profile path")
return value
[docs]
class CensalProfileBaseline(BaseModel):
"""Exact immutable profile revision against which the proposal was reviewed."""
model_config = STRICT_FROZEN_CONFIG
profile_id: ProfileId
record_revision: int = Field(ge=1)
content_digest: ContentDigest
[docs]
@classmethod
def from_record(cls, record: UserProfileRecord) -> CensalProfileBaseline:
"""Capture the canonical revision and self-verifying content digest.
Core types:
:class:`~cadrumo.domain.user_profile.values.UserProfileRecord`.
"""
return cls(
profile_id=record.profile_id,
record_revision=record.record_revision,
content_digest=record.content_digest,
)
#: Wire version of the reviewed-operand record. Named rather than written as a
#: bare default so the number has one home: a reader can find every site bound
#: to this shape, and a bump cannot land on the model while a writer stamping
#: the old number silently disagrees with it.
CENSAL_REVIEWED_OPERAND_SCHEMA_VERSION: Final[Literal[1]] = 1
[docs]
class CensalReviewedOperand(BaseModel):
"""Encrypted exact preimage approved or rejected by the operator."""
model_config = ConfigDict(strict=True, frozen=True, extra="forbid", validate_default=True)
schema_version: Literal[1] = CENSAL_REVIEWED_OPERAND_SCHEMA_VERSION
observation: CensalObservation
baseline: CensalProfileBaseline
field_intents: tuple[CensalReviewedFieldIntent, ...]
proposed_effect_digest: ContentDigestOrAbsent = ""
@field_validator("field_intents")
@classmethod
@pydantic_validation_boundary
def _require_unique_field_intents(
cls,
value: tuple[CensalReviewedFieldIntent, ...],
) -> tuple[CensalReviewedFieldIntent, ...]:
paths = tuple(item.path for item in value)
if paths != CENSAL_ADOPTABLE_PATHS:
raise ValueError(
"censal reviewed operand field intents must cover every canonical "
"adoptable path exactly once and in canonical order"
)
return value
@model_validator(mode="after")
@pydantic_validation_boundary
def _bind_proposed_effect(self) -> CensalReviewedOperand:
expected = self._expected_proposed_effect_digest()
if not self.proposed_effect_digest:
object.__setattr__(self, "proposed_effect_digest", expected)
elif self.proposed_effect_digest != expected:
raise ValueError("censal proposed-effect digest does not match the reviewed operand")
return self
def _expected_proposed_effect_digest(self) -> ContentDigest:
return content_hash_hex(
self.model_dump(
mode="json",
exclude={"proposed_effect_digest"},
exclude_defaults=False,
exclude_none=False,
exclude_unset=False,
)
)
[docs]
class CensalOperationRequest(BaseModel):
"""Exact preflight baseline and review choices with no bearer material."""
model_config = STRICT_FROZEN_CONFIG
baseline: CensalProfileBaseline
field_intents: tuple[CensalReviewedFieldIntent, ...]
@field_validator("field_intents")
@classmethod
@pydantic_validation_boundary
def _require_complete_intents(
cls, value: tuple[CensalReviewedFieldIntent, ...]
) -> tuple[CensalReviewedFieldIntent, ...]:
if tuple(item.path for item in value) != CENSAL_ADOPTABLE_PATHS:
raise ValueError("censal operation request must decide every adoptable path in canonical order")
return value
[docs]
def build_censal_operation_request(record: UserProfileRecord) -> CensalOperationRequest:
"""Bind the exact baseline and safe per-field defaults for frontend review.
Core types:
:class:`~cadrumo.domain.user_profile.values.UserProfileRecord`.
"""
effective = record_to_effective_facts(record)
return CensalOperationRequest(
baseline=CensalProfileBaseline.from_record(record),
field_intents=tuple(
CensalReviewedFieldIntent(
path=path,
intent=(
CensalFieldIntent.ADOPT
if (current := effective.get(path)) is None or current.source == CENSO_SOURCE_TAG
else CensalFieldIntent.PRESERVE
),
)
for path in CENSAL_ADOPTABLE_PATHS
),
)
[docs]
class CensalOperationOutcome(StrEnum):
"""Settled domain outcome represented by the executor result reference."""
APPLIED = "applied"
REJECTED = "rejected"
[docs]
class CensalOperationResult(BaseModel):
"""Typed schema for the safe result referenced by a completed continuation."""
model_config = STRICT_FROZEN_CONFIG
outcome: CensalOperationOutcome
reviewed_proposal_digest: ContentDigest
[docs]
class CensalReviewResponse(BaseModel):
"""Authority-free public decision submitted for a censal REVIEW."""
model_config = STRICT_FROZEN_CONFIG
response_version: Literal[1]
intent: OperationResponseIntentValue
[docs]
class CensalReviewFieldProjectionV1(BaseModel):
"""One safe censo field displayed for an exact REVIEW decision."""
model_config = STRICT_FROZEN_CONFIG
path: str = Field(min_length=3, max_length=160)
intent: CensalFieldIntent
observed_value: str | None
[docs]
class CensalReviewProjectionV1(BaseModel):
"""Bearer-free public projection of the encrypted censal proposal."""
model_config = STRICT_FROZEN_CONFIG
projection_version: Literal[1]
fields: tuple[CensalReviewFieldProjectionV1, ...]
CENSAL_REVIEW_RESPONSE_SCHEMA_BINDING = OperationSchemaBindingV1.bind(
schema_id="user-profile.censo-review.response",
schema_version=1,
model_type=CensalReviewResponse,
)
CENSAL_REVIEW_PROJECTION_SCHEMA_BINDING = OperationSchemaBindingV1.bind(
schema_id="user-profile.censo-review.projection",
schema_version=1,
model_type=CensalReviewProjectionV1,
)
def _project_censal_review(
operand: BaseModel,
interaction: object,
) -> BaseModel:
"""Project only reviewed field values; discard custody and bearer facts."""
del interaction
reviewed = CensalReviewedOperand.model_validate(operand, strict=True)
observed = {fact.path: str(fact.value) for fact in censal_facts_from_read(reviewed.observation)}
return CensalReviewProjectionV1(
projection_version=1,
fields=tuple(
CensalReviewFieldProjectionV1(
path=item.path,
intent=item.intent,
observed_value=observed.get(item.path),
)
for item in reviewed.field_intents
),
)
[docs]
def build_censal_operation_registration(
definition: OperationDefinition,
) -> OperationPublicDefinitionRegistrationV1:
"""Bind the canonical censal definition to its public REVIEW contract."""
return OperationPublicDefinitionRegistrationV1.compose(
definition=definition,
request_schema=OperationSchemaBindingV1.bind(
schema_id="user-profile.censo-review.request",
schema_version=1,
model_type=CensalOperationRequest,
),
result_schema=OperationSchemaBindingV1.bind(
schema_id="user-profile.censo-review.result",
schema_version=1,
model_type=CensalOperationResult,
),
review_projection_schema=CENSAL_REVIEW_PROJECTION_SCHEMA_BINDING,
interaction_response_schema=CENSAL_REVIEW_RESPONSE_SCHEMA_BINDING,
reviewed_operand_type=CensalReviewedOperand,
review_projector=_project_censal_review,
)
[docs]
@dataclass(frozen=True, slots=True)
class CensalOperationAcquisition:
"""Completed read plus its idempotently closeable acquisition resource."""
observation: CensalObservation
resource: AsyncCloseable | None = None
def _load_exact_baseline(
request: OperationRequest[CensalOperationRequest],
*,
profile_decode_context: ProfileDecodeContext,
) -> UserProfileRecord:
profile_id = require_active_bucket_id()
baseline = request.payload.baseline
if request.subject_ref != str(baseline.profile_id) or profile_id != str(baseline.profile_id):
raise ProfileRecordConflictError("censal operation baseline does not identify the active profile")
record = ProfileRecordRepository.for_current_session(
profile_id,
profile_decode_context=profile_decode_context,
).load(profile_id)
if (
record.profile_id != baseline.profile_id
or record.record_revision != baseline.record_revision
or record.content_digest != baseline.content_digest
):
raise ProfileRecordConflictError("censal operation baseline is stale")
return record
async def _acknowledge_if_cancelled(context: OperationExecutorContext) -> bool:
if not context.cancellation.cancellation_requested:
return False
await context.cancellation.acknowledge_cancellation()
return True
[docs]
class CensalOperationExecutor:
"""Acquire once, publish one durable review, and resume only from it."""
def __init__(
self,
*,
certificate_secret_backend_factory: CertificateSecretBackendFactory,
browser_session_factory: BrowserSessionFactoryPort,
operator_scope_ports: OperatorScopePorts,
censal_fetch_port: CensalFetchPort,
acquire: Callable[[], Awaitable[CensalObservation | CensalOperationAcquisition]] | None = None,
apply: Callable[[CensalReviewedOperand], None] | None = None,
before_irreversible_section: Callable[[], Awaitable[None]] | None = None,
) -> None:
"""Initialize the executor with its acquisition, apply, and boundary hooks."""
self._acquire = acquire or (
lambda: _pull_censal_datos(
certificate_secret_backend_factory=certificate_secret_backend_factory,
browser_session_factory=browser_session_factory,
operator_scope_ports=operator_scope_ports,
censal_fetch_port=censal_fetch_port,
)
)
self._apply = apply
self._before_irreversible_section = before_irreversible_section or _ready_for_irreversible_section
[docs]
async def execute(
self,
request: OperationRequest[CensalOperationRequest],
context: OperationExecutorContext,
) -> str | None:
"""Acquire the observation and publish one durable review checkpoint."""
if await _acknowledge_if_cancelled(context):
return None
await context.events.phase(CENSAL_PHASE_PREFLIGHT)
record = _load_exact_baseline(
request,
profile_decode_context=context.authority_operation.profile_decode_context(),
)
if await _acknowledge_if_cancelled(context):
return None
await context.events.phase(CENSAL_PHASE_CLAVE_DEVICE_WAIT)
await context.events.phase(CENSAL_PHASE_REMOTE_READ)
acquired = await self._acquire()
if isinstance(acquired, CensalOperationAcquisition):
observation = acquired.observation
if acquired.resource is not None:
context.cleanup.own(acquired.resource, family=OperationOwnedResource.ASYNC_TASK)
# Resume consumes only the durable operand, so a process-local
# acquisition handle must be released before the detachable
# review is published. Supervisor ownership remains the retry
# path when this idempotent close does not complete.
await acquired.resource.close()
else:
observation = acquired
if await _acknowledge_if_cancelled(context):
return None
await context.events.phase(CENSAL_PHASE_PROPOSAL)
current = _load_exact_baseline(
request,
profile_decode_context=context.authority_operation.profile_decode_context(),
)
reconcile_censal_read(
current,
censal_facts_from_read(observation),
incoming_identity=observation.identity.nif,
)
operand = CensalReviewedOperand(
observation=observation,
baseline=CensalProfileBaseline.from_record(record),
field_intents=request.payload.field_intents,
)
await context.events.phase(CENSAL_PHASE_INTERACTION_WAIT)
continuation_digest = content_hash_hex(
{
"schema_version": 1,
"operation_id": context.identity.operation_id,
"proposed_effect_digest": operand.proposed_effect_digest,
}
)
await context.interactions.publish_review(
interaction_id=secrets.token_hex(32),
identity=context.identity,
revision=context.revision + 1,
presentation_code="censo.review.ready",
response_schema_ref=operation_public_schema_reference(CENSAL_REVIEW_RESPONSE_SCHEMA_BINDING.identity),
continuation_digest=continuation_digest,
expires_at=None,
reviewed_operand=operand,
baseline_digest=content_hash_hex(operand.baseline.model_dump(mode="json")),
proposed_effect_digest=operand.proposed_effect_digest,
)
return None
[docs]
async def resume(
self,
request: OperationRequest[CensalOperationRequest],
checkpoint: OperationResumeCheckpoint,
context: OperationExecutorContext,
) -> str | None:
"""Resume a consumed review checkpoint and apply or reject its operand."""
del request
if await _acknowledge_if_cancelled(context):
return None
if not checkpoint.consumed:
return None
proposal_digest = checkpoint.reviewed_proposal_digest
operand = await context.operands.resolve(proposal_digest, CensalReviewedOperand)
if checkpoint.response_action == "reject":
await context.events.phase(CENSAL_PHASE_REJECT)
await context.events.effect(OperationEffect.NONE)
await context.events.phase(CENSAL_PHASE_SETTLEMENT)
return f"censo-review:{proposal_digest}:{CensalOperationOutcome.REJECTED.value}"
await context.events.phase(CENSAL_PHASE_APPLY)
if await _acknowledge_if_cancelled(context):
return None
_require_current_operand_baseline(
operand,
profile_decode_context=context.authority_operation.profile_decode_context(),
)
await self._before_irreversible_section()
entered_irreversible_section = False
stale_conflict: ProfileRecordConflictError | None = None
try:
async with context.cancellation.irreversible_section():
entered_irreversible_section = True
await context.events.effect(OperationEffect.UNKNOWN)
try:
if self._apply is None:
_apply_reviewed_cotejo(
operand,
profile_decode_context=context.authority_operation.profile_decode_context(),
)
else:
self._apply(operand)
except ProfileRecordConflictError as exc:
stale_conflict = exc
except ValueError:
if not entered_irreversible_section and context.cancellation.cancellation_requested:
await context.cancellation.acknowledge_cancellation()
return None
raise
if stale_conflict is not None:
await context.events.effect(OperationEffect.NONE)
raise stale_conflict
await context.events.effect(OperationEffect.UPDATED)
await context.events.phase(CENSAL_PHASE_SETTLEMENT)
return f"censo-review:{proposal_digest}:{CensalOperationOutcome.APPLIED.value}"
async def _pull_censal_datos(
*,
certificate_secret_backend_factory: CertificateSecretBackendFactory,
browser_session_factory: BrowserSessionFactoryPort,
operator_scope_ports: OperatorScopePorts,
censal_fetch_port: CensalFetchPort,
) -> CensalObservation:
"""Acquire through the sole public live application door."""
from ..live.censo import pull_censal_datos
return await pull_censal_datos(
certificate_secret_backend_factory=certificate_secret_backend_factory,
browser_session_factory=browser_session_factory,
operator_scope_ports=operator_scope_ports,
censal_fetch_port=censal_fetch_port,
)
def _require_current_operand_baseline(
operand: CensalReviewedOperand,
*,
profile_decode_context: ProfileDecodeContext,
) -> None:
"""Keep proven stale state at NONE before entering the ambiguous write window."""
profile_id = require_active_bucket_id()
baseline = operand.baseline
record = ProfileRecordRepository.for_current_session(
profile_id,
profile_decode_context=profile_decode_context,
).load(profile_id)
if (
profile_id != str(baseline.profile_id)
or record.profile_id != baseline.profile_id
or record.record_revision != baseline.record_revision
or record.content_digest != baseline.content_digest
):
raise ProfileRecordConflictError("reviewed censal proposal baseline is stale")
def _apply_reviewed_cotejo(
operand: CensalReviewedOperand,
*,
profile_decode_context: ProfileDecodeContext,
) -> None:
"""Delegate to the sole exact censal mutation authority."""
apply_cotejo(
None,
reviewed_proposal=operand,
profile_decode_context=profile_decode_context,
)
async def _ready_for_irreversible_section() -> None:
"""Default non-blocking boundary before irreversible profile apply."""
[docs]
def build_censal_operation_definition(
*,
certificate_secret_backend_factory: CertificateSecretBackendFactory,
browser_session_factory: BrowserSessionFactoryPort,
operator_scope_ports: OperatorScopePorts,
censal_fetch_port: CensalFetchPort,
acquire: Callable[[], Awaitable[CensalObservation | CensalOperationAcquisition]] | None = None,
apply: Callable[[CensalReviewedOperand], None] | None = None,
before_irreversible_section: Callable[[], Awaitable[None]] | None = None,
) -> OperationDefinition:
"""Build the one CENSO registration with its outer live authorities bound."""
def build() -> CensalOperationExecutor:
return CensalOperationExecutor(
certificate_secret_backend_factory=certificate_secret_backend_factory,
browser_session_factory=browser_session_factory,
operator_scope_ports=operator_scope_ports,
censal_fetch_port=censal_fetch_port,
acquire=acquire,
apply=apply,
before_irreversible_section=before_irreversible_section,
)
return OperationDefinition(
definition_id=CENSAL_OPERATION_DEFINITION_ID,
request_type=CensalOperationRequest,
result_type=CensalOperationResult,
executor_factory=OperationExecutorFactory(
request_type=CensalOperationRequest,
executor_type=CensalOperationExecutor,
build=build,
),
phase_codes=_CENSAL_PHASES,
interaction_kinds=frozenset({OperationInteractionKind.REVIEW}),
capabilities=OperationCapabilities(
durability=OperationDurability.RESUMABLE,
cancellation=OperationCancellation.COOPERATIVE,
deadline=OperationDeadline.COOPERATIVE,
replay=OperationReplayPolicy.RESUMABLE,
baseline=OperationBaselinePolicy.EXACT_APPROVAL,
request_storage=OperationRequestStoragePolicy.SECURE_REFERENCE,
sensitive_input=OperationSensitiveInputPolicy.SECURE_REFERENCE,
conflict_scope=OperationConflictScope.DEFINITION_SUBJECT,
owned_resources=frozenset({OperationOwnedResource.ASYNC_TASK}),
permitted_effects=EFFECTS_WITHOUT_PARTIAL_COMMIT,
close_policy=OperationClosePolicy.DETACH_ALLOWED,
),
reconciliation_policy=OperationReconciliationPolicy.RESUME_FROM_CHECKPOINT,
permitted_frontends=frozenset(
{OperationFrontendProjection.CLI, OperationFrontendProjection.MCP, OperationFrontendProjection.TUI}
),
)
__all__ = [
"CENSAL_OPERATION_DEFINITION_ID",
"CENSAL_REVIEW_PROJECTION_SCHEMA_BINDING",
"CENSAL_REVIEW_RESPONSE_SCHEMA_BINDING",
"CensalFieldIntent",
"CensalOperationAcquisition",
"CensalOperationExecutor",
"CensalOperationOutcome",
"CensalOperationRequest",
"CensalOperationResult",
"CensalProfileBaseline",
"CensalReviewFieldProjectionV1",
"CensalReviewProjectionV1",
"CensalReviewResponse",
"CensalReviewedFieldIntent",
"CensalReviewedOperand",
"build_censal_operation_definition",
"build_censal_operation_registration",
"build_censal_operation_request",
]