"""Authority-fenced durable single-flight Monday owner drafts.""" from __future__ import annotations from checkin_cli.weekly_operations_grounding import ( GroundedWeeklyFacts, GroundedWeeklyReviewSource, WeeklyGroundingError, build_grounded_weekly_source, deterministic_weekly_fallback, lock_weekly_facts, ) from checkin_cli.weekly_operations_knowledge import ( VerifiedWeeklyPublicKnowledge, load_shipped_weekly_public_knowledge, ) from .nutrition_weekly_operations_authority import weekly_operations_is_authorized from .nutrition_weekly_owner_contract import WeeklyDraftLifecycle, WeeklyLifecycleResult from .nutrition_weekly_owner_ledger import ( WeeklyGenerationLedger, WeeklyGenerationLedgerError, WeeklyGenerationOutcome, WeeklyGenerationRow, WeeklyGenerationState, WeeklyGenerationTransaction, require_exact_request_row, ) from .nutrition_weekly_owner_model import ( WeeklyExplanationModel, parse_weekly_model_selection, weekly_model_request, weekly_model_request_digest, weekly_model_request_is_safe, ) from .nutrition_weekly_owner_storage import WeeklyOwnerStorageAuthority from .nutrition_weekly_owner_provider import call_weekly_provider from .nutrition_weekly_owner_request import ( WeeklyProviderOutcomeUnknown, WeeklyOwnerDraftRequest, WeeklyOwnerDraftResult, customer_authority_digest, ) class WeeklyOwnerDraftService: def __init__(self, lifecycle: WeeklyDraftLifecycle, model: WeeklyExplanationModel) -> None: self._lifecycle: WeeklyDraftLifecycle = lifecycle self._model: WeeklyExplanationModel = model self._storage: WeeklyOwnerStorageAuthority = lifecycle.weekly_owner_storage_authority() self._ledger: WeeklyGenerationLedger = WeeklyGenerationLedger(self._storage) def create(self, request: WeeklyOwnerDraftRequest) -> WeeklyOwnerDraftResult: try: facts = request.facts except WeeklyGroundingError: return WeeklyOwnerDraftResult(False, None, 0, False, "authority_rejected") if not self._authorized(request, facts): return WeeklyOwnerDraftResult(False, None, 0, False, "authority_rejected") if not self._current(request, facts): return WeeklyOwnerDraftResult(False, None, 0, False, "grounding_stale") knowledge = load_shipped_weekly_public_knowledge() base = weekly_generation_base(request, facts) calls = 0 with self._ledger.locked() as transaction: latest = transaction.latest(base.logical_key) if latest is not None: require_exact_request_row(base, latest) for row in transaction.rows: if row.logical_key == base.logical_key or row.request_base_digest == base.request_base_digest: require_exact_request_row(base, row) if latest is not None and latest.state is WeeklyGenerationState.BOUND: draft = self._lifecycle.grounded_weekly_review_result(latest.draft_id, request.owner_key) if not draft.accepted or draft.draft_id != latest.draft_id: raise WeeklyGenerationLedgerError("replay draft binding") return self._result(draft, 0, latest.fallback) if latest is not None and latest.state in {WeeklyGenerationState.SENDING, WeeklyGenerationState.UNKNOWN}: if latest.state is WeeklyGenerationState.SENDING: latest = transaction.append( base, state=WeeklyGenerationState.UNKNOWN, outcome=WeeklyGenerationOutcome.INTERRUPTED_UNKNOWN, ordinal=latest.ordinal, request_digest=latest.request_digest, ) source = deterministic_weekly_fallback(facts) latest = self._append_final( transaction, base, latest, source, WeeklyGenerationOutcome.UNKNOWN_FALLBACK, ) elif latest is not None and latest.state is WeeklyGenerationState.FINAL: source = self._source_from_row(facts, latest) else: ordinal = 2 if latest is not None and latest.state is WeeklyGenerationState.INVALID else 1 model_request = weekly_model_request(facts, knowledge, correction=ordinal == 2) if not weekly_model_request_is_safe(model_request): return WeeklyOwnerDraftResult(False, None, 0, False, "privacy_rejected") if not self._current(request, facts): return WeeklyOwnerDraftResult(False, None, 0, False, "grounding_stale") request_digest = weekly_model_request_digest(model_request) _ = transaction.append( base, state=WeeklyGenerationState.SENDING, outcome=WeeklyGenerationOutcome.PROVIDER_AUTHORIZED, ordinal=ordinal, request_digest=request_digest, ) calls += 1 raw, provider_failed, unknown = call_weekly_provider(self._model, model_request) if unknown: uncertain = transaction.append( base, state=WeeklyGenerationState.UNKNOWN, outcome=WeeklyGenerationOutcome.INTERRUPTED_UNKNOWN, ordinal=ordinal, request_digest=request_digest, ) source = deterministic_weekly_fallback(facts) latest = self._append_final( transaction, base, uncertain, source, WeeklyGenerationOutcome.UNKNOWN_FALLBACK, ) else: selected = parse_weekly_model_selection(raw, facts, knowledge) if selected is None and not provider_failed and ordinal == 1: _ = transaction.append( base, state=WeeklyGenerationState.INVALID, outcome=WeeklyGenerationOutcome.SCHEMA_INVALID, ordinal=1, request_digest=request_digest, ) return self._correct(transaction, request, facts, base, knowledge, calls) source = deterministic_weekly_fallback(facts) if selected is None else build_grounded_weekly_source( facts, tuple(selected.principle_ids), selected.emphasis, ) outcome = WeeklyGenerationOutcome.SELECTED if selected is None: outcome = WeeklyGenerationOutcome.PROVIDER_FAILURE_FALLBACK if provider_failed else WeeklyGenerationOutcome.INVALID_FALLBACK latest = self._append_final_values( transaction, base, ordinal, request_digest, source, outcome, ) return self._persist(transaction, request, facts, base, latest, calls) def _correct( self, transaction: WeeklyGenerationTransaction, request: WeeklyOwnerDraftRequest, facts: GroundedWeeklyFacts, base: WeeklyGenerationRow, knowledge: VerifiedWeeklyPublicKnowledge, calls: int, ) -> WeeklyOwnerDraftResult: model_request = weekly_model_request(facts, knowledge, correction=True) if not self._current(request, facts): return WeeklyOwnerDraftResult(False, None, calls, False, "grounding_stale") request_digest = weekly_model_request_digest(model_request) _ = transaction.append( base, state=WeeklyGenerationState.SENDING, outcome=WeeklyGenerationOutcome.PROVIDER_AUTHORIZED, ordinal=2, request_digest=request_digest, ) calls += 1 raw, failed, unknown = call_weekly_provider(self._model, model_request) if unknown: uncertain = transaction.append( base, state=WeeklyGenerationState.UNKNOWN, outcome=WeeklyGenerationOutcome.INTERRUPTED_UNKNOWN, ordinal=2, request_digest=request_digest, ) source = deterministic_weekly_fallback(facts) latest = self._append_final(transaction, base, uncertain, source, WeeklyGenerationOutcome.UNKNOWN_FALLBACK) else: selected = parse_weekly_model_selection(raw, facts, knowledge) source = deterministic_weekly_fallback(facts) if selected is None else build_grounded_weekly_source(facts, tuple(selected.principle_ids), selected.emphasis) outcome = WeeklyGenerationOutcome.SELECTED if selected is not None else (WeeklyGenerationOutcome.PROVIDER_FAILURE_FALLBACK if failed else WeeklyGenerationOutcome.INVALID_FALLBACK) latest = self._append_final_values(transaction, base, 2, request_digest, source, outcome) return self._persist(transaction, request, facts, base, latest, calls) def _persist( self, transaction: WeeklyGenerationTransaction, request: WeeklyOwnerDraftRequest, facts: GroundedWeeklyFacts, base: WeeklyGenerationRow, latest: WeeklyGenerationRow, calls: int, ) -> WeeklyOwnerDraftResult: require_exact_request_row(base, latest) if not self._current(request, facts): return WeeklyOwnerDraftResult(False, None, calls, latest.fallback, "grounding_stale") source = self._source_from_row(facts, latest) draft = self._lifecycle.persist_grounded_weekly_review(request.customer_key, request.owner_key, source) if not draft.accepted or draft.draft_id is None: return self._result(draft, calls, source.fallback_used) bound = transaction.append( base, state=WeeklyGenerationState.BOUND, outcome=WeeklyGenerationOutcome.DRAFT_BOUND, ordinal=latest.ordinal, request_digest=latest.request_digest, principle_ids=latest.principle_ids, emphasis=latest.emphasis, fallback=latest.fallback, source_digest=latest.source_digest, draft_id=draft.draft_id, ) require_exact_request_row(base, bound) return self._result(draft, calls, source.fallback_used) def _authorized(self, request: WeeklyOwnerDraftRequest, facts: GroundedWeeklyFacts) -> bool: return ( request.bound_summary.verify() and facts.verify() and request.storage_binding_digest == self._storage.binding_digest and request.owner_key == self._lifecycle.weekly_owner_key() and request.owner_key[:2] == (request.runtime.owner_user_id, request.runtime.owner_chat_id) and request.customer_key == request.runtime.customer_key and request.runtime.now.weekday() == request.config.weekly_weekday and facts.candidate_digest == request.runtime.candidate_digest and facts.config_digest == request.config.digest and facts.customer_authority_digest == customer_authority_digest(request.runtime) and weekly_operations_is_authorized(request.config, request.receipt, request.runtime) ) @staticmethod def _current(request: WeeklyOwnerDraftRequest, facts: GroundedWeeklyFacts) -> bool: current = request.current_bound_summary() if not current.verify() or current.authority_digest != request.bound_summary.authority_digest or current.summary_digest != request.bound_summary.summary_digest: return False try: return lock_weekly_facts(current, request.bindings) == facts except WeeklyGroundingError: return False @staticmethod def _source_from_row(facts: GroundedWeeklyFacts, row: WeeklyGenerationRow) -> GroundedWeeklyReviewSource: if row.fallback: source = deterministic_weekly_fallback(facts) elif row.emphasis is None: raise WeeklyGenerationLedgerError("source emphasis") else: source = build_grounded_weekly_source(facts, row.principle_ids, row.emphasis) if source.source_digest != row.source_digest: raise WeeklyGenerationLedgerError("source binding") return source @staticmethod def _append_final( transaction: WeeklyGenerationTransaction, base: WeeklyGenerationRow, predecessor: WeeklyGenerationRow, source: GroundedWeeklyReviewSource, outcome: WeeklyGenerationOutcome, ) -> WeeklyGenerationRow: return WeeklyOwnerDraftService._append_final_values( transaction, base, predecessor.ordinal, predecessor.request_digest, source, outcome, ) @staticmethod def _append_final_values( transaction: WeeklyGenerationTransaction, base: WeeklyGenerationRow, ordinal: int, request_digest: str, source: GroundedWeeklyReviewSource, outcome: WeeklyGenerationOutcome, ) -> WeeklyGenerationRow: return transaction.append( base, state=WeeklyGenerationState.FINAL, outcome=outcome, ordinal=ordinal, request_digest=request_digest, principle_ids=source.principle_ids, emphasis=source.emphasis, fallback=source.fallback_used, source_digest=source.source_digest, ) @staticmethod def _result(draft: WeeklyLifecycleResult, calls: int, fallback: bool) -> WeeklyOwnerDraftResult: return WeeklyOwnerDraftResult(draft.accepted, draft, calls, fallback, None if draft.accepted else draft.error or "draft_rejected") from .nutrition_weekly_owner_authority_binding import ( grounded_weekly_source_is_reserved, weekly_generation_base, ) __all__ = ( "WeeklyOwnerDraftRequest", "WeeklyOwnerDraftResult", "WeeklyOwnerDraftService", "WeeklyProviderOutcomeUnknown", "customer_authority_digest", "grounded_weekly_source_is_reserved", "weekly_generation_base", )