"""Public frontend driver for the canonical reviewed censal operation."""
from __future__ import annotations
import asyncio
from collections.abc import Callable
from contextlib import AsyncExitStack
from dataclasses import dataclass
from typing import TYPE_CHECKING
from ..application.operations.composition import OperationComposedServices, OperationSubmission
from ..application.operations.frontend_contracts import OperationReviewProjectionResultV1
from ..application.operations.frontend_projection import OperationReviewAvailableInteractionV1
from ..application.operations.frontend_requests import (
OperationObservationRequestV1,
OperationObservationSuccessV1,
OperationResponseApplyRequestV1,
OperationResponseControlRequestV1,
OperationResponseMutationSuccessV1,
OperationResponseRejectRequestV1,
OperationReviewProjectionRequestV1,
OperationReviewProjectionSuccessV1,
)
from ..application.operations.models import OperationRequest
from ..application.user_profile.censal_operation import (
CENSAL_OPERATION_DEFINITION_ID,
CensalOperationOutcome,
CensalOperationRequest,
CensalOperationResult,
CensalReviewProjectionV1,
build_censal_operation_request,
)
from ..application.user_profile.profile_record_repository import ProfileRecordRepository
from ..core.bucket_pointer import require_active_bucket_id
from ..core.errors.hierarchy import InternalInvariantError
from ..core.operations import (
OperationEffect,
OperationLifecycle,
OperationTerminalCondition,
)
from ..core.time.clock import now
from .operation_composition import compose_operation_dependencies
if TYPE_CHECKING:
from ..domain.calculations.registry.authority import PinnedAuthorityOperation
_OBSERVATION_LIMIT = 256
[docs]
@dataclass(frozen=True, slots=True)
class CensalReviewedFrontendResult:
"""Safe terminal result retained by a frontend after exact review."""
operation_id: str
applied: bool
projection: CensalReviewProjectionV1
async def _observe(services: OperationComposedServices, operation_id: str) -> OperationObservationSuccessV1:
observed = await services.observation.observe(
OperationObservationRequestV1(operation_id=operation_id, after_cursor=0, page_limit=_OBSERVATION_LIMIT)
)
if not isinstance(observed, OperationObservationSuccessV1):
raise InternalInvariantError("censal operation observation was refused")
return observed
def _active_censal_operation_request(
operation: PinnedAuthorityOperation,
) -> OperationRequest[CensalOperationRequest]:
"""Build the registered request from the one authenticated profile record."""
profile_id = require_active_bucket_id()
record = ProfileRecordRepository.for_current_session(
profile_id,
profile_decode_context=operation.profile_decode_context(),
).load(profile_id)
payload = build_censal_operation_request(record)
return OperationRequest(
definition_id=CENSAL_OPERATION_DEFINITION_ID,
subject_ref=profile_id,
payload=payload,
)
def _require_review_interaction(observed: OperationObservationSuccessV1) -> OperationReviewAvailableInteractionV1:
"""Require the operation to expose the reviewed interaction it promised."""
pending = observed.projection.pending_interaction
if not isinstance(pending, OperationReviewAvailableInteractionV1):
raise InternalInvariantError("censal operation did not publish its reviewed proposal")
return pending
async def _resolve_censal_projection(
services: OperationComposedServices,
pending: OperationReviewAvailableInteractionV1,
) -> CensalReviewProjectionV1:
"""Resolve and type-check the safe projection shown to the reviewer."""
projected: OperationReviewProjectionResultV1[CensalReviewProjectionV1] = await services.review.resolve(
OperationReviewProjectionRequestV1(reference=pending.review_reference)
)
if not isinstance(projected, OperationReviewProjectionSuccessV1) or not isinstance(
projected.projection, CensalReviewProjectionV1
):
raise InternalInvariantError("censal reviewed projection was unavailable")
return projected.projection
async def _answer_censal_review(
services: OperationComposedServices,
submission: OperationSubmission,
pending: OperationReviewAvailableInteractionV1,
*,
actor_ref: str,
apply: bool,
) -> None:
"""Submit exactly one operator answer through the held response authority."""
operation_id = submission.receipt.operation_id
control = await services.response(
OperationResponseControlRequestV1(
operation_id=operation_id,
interaction_id=pending.interaction_id,
revision=pending.revision,
actor_ref=actor_ref,
),
submission.response_capability,
)
if apply:
accepted = await control.apply(
OperationResponseApplyRequestV1(
operation_id=operation_id,
interaction_id=pending.interaction_id,
revision=pending.revision,
actor_ref=actor_ref,
responded_at=now(),
)
)
else:
accepted = await control.reject(
OperationResponseRejectRequestV1(
operation_id=operation_id,
interaction_id=pending.interaction_id,
revision=pending.revision,
actor_ref=actor_ref,
responded_at=now(),
reason_code="censo.review.operator-rejected",
)
)
if not isinstance(accepted, OperationResponseMutationSuccessV1):
raise InternalInvariantError("censal reviewed response was refused")
def _parse_censal_result_reference(
result_ref: str,
*,
expected_outcome: CensalOperationOutcome,
) -> CensalOperationResult:
"""Validate and type the backend result reference after settlement."""
prefix, separator, outcome = result_ref.rpartition(":")
family, digest_separator, reviewed_digest = prefix.partition(":")
if not separator or not digest_separator or family != "censo-review":
raise InternalInvariantError("censal reviewed operation returned an invalid result reference")
typed_result = CensalOperationResult(
outcome=CensalOperationOutcome(outcome),
reviewed_proposal_digest=reviewed_digest,
)
if typed_result.outcome is not expected_outcome:
raise InternalInvariantError("censal reviewed operation returned a mismatched outcome")
return typed_result
def _assert_censal_terminal_success(
observed: OperationObservationSuccessV1,
*,
apply: bool,
) -> None:
"""Require a successful terminal projection with the declared effect."""
expected_outcome = CensalOperationOutcome.APPLIED if apply else CensalOperationOutcome.REJECTED
expected_effect = OperationEffect.UPDATED if apply else OperationEffect.NONE
projection = observed.projection
result_ref = projection.result_ref
if (
projection.terminal_condition is not OperationTerminalCondition.SUCCEEDED
or projection.effect is not expected_effect
or result_ref is None
):
raise InternalInvariantError("censal reviewed operation did not succeed with its declared effect")
_parse_censal_result_reference(result_ref, expected_outcome=expected_outcome)
async def _await_censal_settlement(
services: OperationComposedServices,
operation_id: str,
*,
apply: bool,
) -> None:
"""Wait for the reviewed operation to settle, then check its public outcome."""
await services.submission.settled(operation_id)
observed = await _observe(services, operation_id)
if observed.projection.lifecycle is not OperationLifecycle.TERMINAL:
raise InternalInvariantError("censal reviewed operation did not settle")
_assert_censal_terminal_success(observed, apply=apply)
[docs]
async def review_censal_with_services(
services: OperationComposedServices,
*,
operation: PinnedAuthorityOperation,
actor_ref: str,
decide: Callable[[CensalReviewProjectionV1], bool],
) -> CensalReviewedFrontendResult:
"""Run one reviewed censal acquisition through already composed services."""
request = _active_censal_operation_request(operation)
submitted = await services.submission.submit(
request,
actor_ref=actor_ref,
)
operation_id = submitted.receipt.operation_id
await services.submission.start(operation_id)
# The executor stops at its review checkpoint; that is what settles first.
await services.submission.settled(operation_id)
waiting = await _observe(services, operation_id)
pending = _require_review_interaction(waiting)
projection = await _resolve_censal_projection(services, pending)
apply = decide(projection)
await _answer_censal_review(services, submitted, pending, actor_ref=actor_ref, apply=apply)
await _await_censal_settlement(services, operation_id, apply=apply)
return CensalReviewedFrontendResult(operation_id=operation_id, applied=apply, projection=projection)
async def _run(
*,
actor_ref: str,
decide: Callable[[CensalReviewProjectionV1], bool],
) -> CensalReviewedFrontendResult:
from ..adapters.persistence.storage.operator_scope import build_operator_scope_ports
from ..domain.calculations.registry.authority import bundled_indexed_authority
async with AsyncExitStack() as owned:
authority_operation = owned.enter_context(bundled_indexed_authority().operation())
composed = compose_operation_dependencies(
authority_operation=authority_operation,
operator_scope_ports=build_operator_scope_ports(),
)
owned.push_async_callback(composed.shutdown)
return await review_censal_with_services(
composed,
operation=authority_operation,
actor_ref=actor_ref,
decide=decide,
)
[docs]
def run_censal_review(
*,
actor_ref: str,
decide: Callable[[CensalReviewProjectionV1], bool],
) -> CensalReviewedFrontendResult:
"""Acquire once, show the exact safe projection, and answer its review."""
return asyncio.run(_run(actor_ref=actor_ref, decide=decide))
__all__ = ["CensalReviewedFrontendResult", "review_censal_with_services", "run_censal_review"]