334 lines
12 KiB
Python
334 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
|
|
from govoplan_core.core.institutional import (
|
|
CAPABILITY_DECISION_REGISTRY,
|
|
CAPABILITY_MANDATE_RESOLVER,
|
|
ActorRepresentationReference,
|
|
DecisionAssuranceLevel,
|
|
DecisionEffectReference,
|
|
DecisionRegistry,
|
|
EvidenceReference,
|
|
FormalDecision,
|
|
GovernedContextEnvelope,
|
|
InformationGovernanceReference,
|
|
InstitutionalContextError,
|
|
InstitutionalReference,
|
|
LegalBasisReference,
|
|
MandateDefinition,
|
|
MandateResolution,
|
|
MandateResolutionRequest,
|
|
MandateResolver,
|
|
TemporalRevision,
|
|
resolve_mandate_candidates,
|
|
)
|
|
from govoplan_committee.backend.workspace import CAPABILITY_COMMITTEE_WORKSPACE
|
|
|
|
|
|
CAPABILITY_COMMITTEE_DECISION_PATH = "committee.decision_path"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CommitteeDecisionProposal:
|
|
tenant_id: str
|
|
decision_id: str
|
|
revision: str
|
|
effective_at: datetime
|
|
meeting_ref: str
|
|
agenda_item_ref: str
|
|
decision_type: str
|
|
subject_refs: tuple[InstitutionalReference, ...]
|
|
organization_unit_ref: InstitutionalReference
|
|
function_ref: InstitutionalReference
|
|
actor: ActorRepresentationReference
|
|
approval_refs: tuple[InstitutionalReference, ...]
|
|
fact_evidence: tuple[EvidenceReference, ...]
|
|
legal_bases: tuple[LegalBasisReference, ...]
|
|
operative_result: str
|
|
reasoning: str
|
|
case_ref: InstitutionalReference | None = None
|
|
jurisdiction_refs: tuple[InstitutionalReference, ...] = ()
|
|
party_refs: tuple[InstitutionalReference, ...] = ()
|
|
record_refs: tuple[InstitutionalReference, ...] = ()
|
|
conditions: tuple[str, ...] = ()
|
|
requested_effects: tuple[DecisionEffectReference, ...] = ()
|
|
remedy_refs: tuple[str, ...] = ()
|
|
review_refs: tuple[str, ...] = ()
|
|
information_governance: InformationGovernanceReference | None = None
|
|
assurance_level: DecisionAssuranceLevel = "human"
|
|
automation_preparation_refs: tuple[str, ...] = ()
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.effective_at.tzinfo is None or self.effective_at.utcoffset() is None:
|
|
raise InstitutionalContextError(
|
|
"Committee decision effective_at must include a timezone."
|
|
)
|
|
if not self.meeting_ref.strip() or not self.agenda_item_ref.strip():
|
|
raise InstitutionalContextError(
|
|
"Committee meeting and agenda item references are required."
|
|
)
|
|
if not self.subject_refs:
|
|
raise InstitutionalContextError(
|
|
"Committee decisions require at least one subject."
|
|
)
|
|
if not self.approval_refs:
|
|
raise InstitutionalContextError(
|
|
"Committee decisions require an accepted approval reference."
|
|
)
|
|
if not self.fact_evidence or not self.legal_bases:
|
|
raise InstitutionalContextError(
|
|
"Committee decisions require fact evidence and legal basis versions."
|
|
)
|
|
if not self.operative_result.strip() or not self.reasoning.strip():
|
|
raise InstitutionalContextError(
|
|
"Committee decisions require an operative result and reasoning."
|
|
)
|
|
references = (
|
|
*self.subject_refs,
|
|
self.organization_unit_ref,
|
|
self.function_ref,
|
|
*self.approval_refs,
|
|
self.case_ref,
|
|
*self.jurisdiction_refs,
|
|
*self.party_refs,
|
|
*self.record_refs,
|
|
)
|
|
if any(
|
|
item is not None and item.tenant_id != self.tenant_id
|
|
for item in references
|
|
):
|
|
raise InstitutionalContextError(
|
|
"Committee decision references cannot cross tenants."
|
|
)
|
|
if self.organization_unit_ref.kind != "organization_unit":
|
|
raise InstitutionalContextError(
|
|
"Committee decision organization reference must identify an organization unit."
|
|
)
|
|
if self.function_ref.kind != "function":
|
|
raise InstitutionalContextError(
|
|
"Committee decision function reference must identify a function."
|
|
)
|
|
if any(item.kind != "approval" for item in self.approval_refs):
|
|
raise InstitutionalContextError(
|
|
"Committee decision approval references must have approval kind."
|
|
)
|
|
if any(item.kind != "jurisdiction" for item in self.jurisdiction_refs):
|
|
raise InstitutionalContextError(
|
|
"Committee decision jurisdiction references must have jurisdiction kind."
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CommitteeDecisionPathResult:
|
|
decision: FormalDecision
|
|
mandate: MandateDefinition
|
|
persisted_by_decision_registry: bool
|
|
persisted_by_committee_projection: bool = False
|
|
|
|
def reconstruction_payload(self) -> dict[str, object]:
|
|
return {
|
|
"decision": self.decision.to_dict(include_protected=True),
|
|
"mandate_ref": self.mandate.reference.to_dict(),
|
|
"mandate_revision": self.mandate.temporal.to_dict(),
|
|
"persisted_by_decision_registry": self.persisted_by_decision_registry,
|
|
"persisted_by_committee_projection": self.persisted_by_committee_projection,
|
|
}
|
|
|
|
|
|
class CommitteeDecisionPath:
|
|
"""Build one reconstructable formal outcome without owning Decision storage."""
|
|
|
|
def __init__(self, registry: object | None = None) -> None:
|
|
self._registry = registry
|
|
|
|
def decide(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
proposal: CommitteeDecisionProposal,
|
|
mandate_resolution: MandateResolution | None = None,
|
|
observed_effects: tuple[DecisionEffectReference, ...] = (),
|
|
expected_revision: str | None = None,
|
|
) -> CommitteeDecisionPathResult:
|
|
resolution = mandate_resolution or self._resolve_mandate(
|
|
session,
|
|
principal,
|
|
proposal=proposal,
|
|
)
|
|
mandate = _accepted_mandate(proposal, resolution)
|
|
decision_registry = _capability(
|
|
self._registry,
|
|
CAPABILITY_DECISION_REGISTRY,
|
|
)
|
|
decision_ref = InstitutionalReference(
|
|
kind="decision",
|
|
owner_module=(
|
|
"decisions"
|
|
if isinstance(decision_registry, DecisionRegistry)
|
|
else "committee"
|
|
),
|
|
object_id=proposal.decision_id,
|
|
tenant_id=proposal.tenant_id,
|
|
version=proposal.revision,
|
|
valid_at=proposal.effective_at,
|
|
)
|
|
temporal = TemporalRevision(
|
|
revision=proposal.revision,
|
|
valid_from=proposal.effective_at,
|
|
recorded_at=proposal.effective_at,
|
|
change_reason="Committee decision accepted",
|
|
)
|
|
evidence = tuple(
|
|
dict.fromkeys((*proposal.fact_evidence, *mandate.evidence, *resolution.evidence))
|
|
)
|
|
legal_bases = tuple(
|
|
dict.fromkeys((*proposal.legal_bases, *mandate.legal_bases))
|
|
)
|
|
authority_context = GovernedContextEnvelope(
|
|
tenant_id=proposal.tenant_id,
|
|
temporal=temporal,
|
|
actor=proposal.actor,
|
|
organization_unit_ref=proposal.organization_unit_ref,
|
|
function_ref=proposal.function_ref,
|
|
mandate_ref=mandate.reference,
|
|
jurisdiction_refs=proposal.jurisdiction_refs,
|
|
case_ref=proposal.case_ref,
|
|
party_refs=proposal.party_refs,
|
|
approval_refs=proposal.approval_refs,
|
|
decision_ref=decision_ref,
|
|
record_refs=proposal.record_refs,
|
|
legal_bases=legal_bases,
|
|
evidence=evidence,
|
|
information_governance=proposal.information_governance,
|
|
)
|
|
decision = FormalDecision(
|
|
reference=decision_ref,
|
|
temporal=temporal,
|
|
decision_type=proposal.decision_type,
|
|
subject_refs=proposal.subject_refs,
|
|
state="decided",
|
|
authority_context=authority_context,
|
|
fact_evidence=evidence,
|
|
legal_bases=legal_bases,
|
|
assurance_level=proposal.assurance_level,
|
|
automation_preparation_refs=proposal.automation_preparation_refs,
|
|
operative_result=proposal.operative_result,
|
|
reasoning=proposal.reasoning,
|
|
conditions=proposal.conditions,
|
|
requested_effects=proposal.requested_effects,
|
|
observed_effects=observed_effects,
|
|
publication_refs=(
|
|
f"committee-meeting:{proposal.meeting_ref}",
|
|
f"committee-agenda-item:{proposal.agenda_item_ref}",
|
|
),
|
|
remedy_refs=proposal.remedy_refs,
|
|
review_refs=proposal.review_refs,
|
|
)
|
|
persisted = False
|
|
projected = False
|
|
if isinstance(decision_registry, DecisionRegistry):
|
|
decision = decision_registry.record_decision(
|
|
session,
|
|
principal,
|
|
decision=decision,
|
|
expected_revision=expected_revision,
|
|
)
|
|
persisted = True
|
|
else:
|
|
workspace = _capability(self._registry, CAPABILITY_COMMITTEE_WORKSPACE)
|
|
if workspace is not None and hasattr(workspace, "record_local_decision"):
|
|
decision = workspace.record_local_decision(
|
|
session,
|
|
principal,
|
|
decision=decision,
|
|
meeting_id=proposal.meeting_ref,
|
|
agenda_item_id=proposal.agenda_item_ref,
|
|
expected_revision=expected_revision,
|
|
)
|
|
projected = True
|
|
return CommitteeDecisionPathResult(
|
|
decision=decision,
|
|
mandate=mandate,
|
|
persisted_by_decision_registry=persisted,
|
|
persisted_by_committee_projection=projected,
|
|
)
|
|
|
|
def _resolve_mandate(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
proposal: CommitteeDecisionProposal,
|
|
) -> MandateResolution:
|
|
resolver = _capability(self._registry, CAPABILITY_MANDATE_RESOLVER)
|
|
if not isinstance(resolver, MandateResolver):
|
|
raise InstitutionalContextError(
|
|
"Committee decision requires a Mandate resolver or an explicit governed resolution."
|
|
)
|
|
return resolver.resolve_mandate(
|
|
session,
|
|
principal,
|
|
request=_mandate_request(proposal),
|
|
)
|
|
|
|
|
|
def _accepted_mandate(
|
|
proposal: CommitteeDecisionProposal,
|
|
resolution: MandateResolution,
|
|
) -> MandateDefinition:
|
|
if not resolution.competent:
|
|
raise InstitutionalContextError(
|
|
resolution.explanation or "The acting function is not competent to decide."
|
|
)
|
|
if resolution.conflict_refs:
|
|
raise InstitutionalContextError(
|
|
"Mandate resolution has unresolved conflicts: "
|
|
+ ", ".join(resolution.conflict_refs)
|
|
)
|
|
verified = resolve_mandate_candidates(
|
|
_mandate_request(proposal),
|
|
resolution.mandates,
|
|
)
|
|
if not verified.competent:
|
|
raise InstitutionalContextError(
|
|
verified.explanation
|
|
or "Committee decision requires exactly one effective active mandate."
|
|
)
|
|
return verified.mandates[0]
|
|
|
|
|
|
def _mandate_request(
|
|
proposal: CommitteeDecisionProposal,
|
|
) -> MandateResolutionRequest:
|
|
return MandateResolutionRequest(
|
|
tenant_id=proposal.tenant_id,
|
|
effective_at=proposal.effective_at,
|
|
task_type="committee.formal_decision",
|
|
authority_type=proposal.decision_type,
|
|
organization_unit_ref=proposal.organization_unit_ref,
|
|
function_ref=proposal.function_ref,
|
|
jurisdiction_refs=proposal.jurisdiction_refs,
|
|
)
|
|
|
|
|
|
def _capability(registry: object | None, name: str) -> object | None:
|
|
if (
|
|
registry is None
|
|
or not hasattr(registry, "has_capability")
|
|
or not hasattr(registry, "capability")
|
|
or not registry.has_capability(name)
|
|
):
|
|
return None
|
|
return registry.capability(name)
|
|
|
|
|
|
__all__ = [
|
|
"CAPABILITY_COMMITTEE_DECISION_PATH",
|
|
"CommitteeDecisionPath",
|
|
"CommitteeDecisionPathResult",
|
|
"CommitteeDecisionProposal",
|
|
]
|