from __future__ import annotations from collections.abc import Mapping, Sequence from datetime import timedelta import hashlib import json from sqlalchemy import func, or_, select from sqlalchemy.orm import Session from govoplan_core.auth import ApiPrincipal from govoplan_core.core.concurrency import claim_revision from govoplan_core.core.events import ( EventActorRef, EventObjectRef, EventTenantRef, PlatformEvent, emit_platform_event, ) from govoplan_core.core.notifications import ( NotificationDispatchRequest, notification_dispatch_provider, ) from govoplan_core.core.organizations import OrganizationFunctionRef from govoplan_core.core.policy import ( FunctionAssignmentGovernanceAction, FunctionAssignmentGovernanceDecision, FunctionAssignmentGovernanceRequest, function_assignment_governance_policy, ) from govoplan_core.core.principal_cache import invalidate_auth_principals from govoplan_core.core.workflows import ( WorkflowCurrentStepResolution, WorkflowInstanceRef, WorkflowStandardStartRequest, workflow_orchestration_provider, ) from govoplan_core.db.base import utcnow from govoplan_core.security.time import utc_now from govoplan_idm.backend.assignment_events import emit_assignment_event from govoplan_idm.backend.assignment_transitions import ( AssignmentTransitionError, validate_assignment_shape, ) from govoplan_idm.backend.db.models import ( IdmFunctionAssignmentChange, IdmFunctionAssignmentChangeEvent, IdmOrganizationFunctionAssignment, IdmTenantSettings, new_uuid, ) OPEN_STATES = { "submitted", "awaiting_holder", "awaiting_authority", "awaiting_recipient", "changes_requested", "blocked", "failed_manual_review", } TERMINAL_STATES = { "applied", "rejected", "withdrawn", "expired", "cancelled", } STEP_STATE = { "holder": "awaiting_holder", "authority": "awaiting_authority", "recipient": "awaiting_recipient", } NODE_STEP = { "holder_review": "holder", "authority_review": "authority", "recipient_review": "recipient", } class FunctionAssignmentChangeError(ValueError): pass class FunctionAssignmentChangeUnavailable(FunctionAssignmentChangeError): pass class FunctionAssignmentChangeConflict(FunctionAssignmentChangeError): pass def resolve_submission_capability( session: Session, *, principal: ApiPrincipal, registry: object | None, kind: str, function: OrganizationFunctionRef, candidate_identity_id: str, candidate_account_id: str | None, has_evidence: bool, ) -> tuple[FunctionAssignmentGovernanceDecision | None, str | None]: workflow = workflow_orchestration_provider(registry) if workflow is None: return None, "Workflow Engine is not installed or enabled." policy = function_assignment_governance_policy(registry) if policy is None: return ( None, "Function assignment governance Policy is not installed or enabled.", ) settings = _effective_function_settings( session, tenant_id=principal.tenant_id, function=function, ) authority_function_id = _authority_function_id(settings) context = _actor_context( session, principal=principal, function_id=function.id, authority_function_id=authority_function_id, candidate_identity_id=candidate_identity_id, candidate_account_id=candidate_account_id, initiator_account_id=principal.account_id, has_evidence=has_evidence, ) decision = policy.resolve_function_assignment_action( session, request=FunctionAssignmentGovernanceRequest( tenant_id=principal.tenant_id, kind=kind, # type: ignore[arg-type] action="submit", function_id=function.id, actor=principal.to_platform_principal(), candidate_identity_id=candidate_identity_id, candidate_account_id=candidate_account_id, function_settings=settings, context=context, ), ) return decision, decision.reason def create_function_assignment_change( session: Session, *, principal: ApiPrincipal, registry: object | None, function: OrganizationFunctionRef, payload: object, ) -> tuple[IdmFunctionAssignmentChange, bool]: kind = str(getattr(payload, "kind")) candidate_identity_id = str(getattr(payload, "candidate_identity_id")) candidate_account_id = _text(getattr(payload, "candidate_account_id", None)) idempotency_key = str(getattr(payload, "idempotency_key")).strip() fingerprint = _request_fingerprint(payload) existing = session.scalar( select(IdmFunctionAssignmentChange).where( IdmFunctionAssignmentChange.tenant_id == principal.tenant_id, IdmFunctionAssignmentChange.kind == kind, IdmFunctionAssignmentChange.initiator_account_id == principal.account_id, IdmFunctionAssignmentChange.idempotency_key == idempotency_key, ) ) if existing is not None: if existing.metadata_.get("request_fingerprint") != fingerprint: raise FunctionAssignmentChangeConflict( "The idempotency key was already used with different request data." ) return existing, True decision, unavailable_reason = resolve_submission_capability( session, principal=principal, registry=registry, kind=kind, function=function, candidate_identity_id=candidate_identity_id, candidate_account_id=candidate_account_id, has_evidence=bool(getattr(payload, "evidence", ())), ) if decision is None: raise FunctionAssignmentChangeUnavailable( unavailable_reason or "Governed function assignment changes are unavailable." ) if not decision.allowed: raise FunctionAssignmentChangeUnavailable( decision.reason or "The function assignment change is not allowed." ) _validate_requested_validity(payload, decision) represented_assignment_id = _text( getattr(payload, "represented_assignment_id", None) ) if represented_assignment_id is not None: _require_actor_assignment( session, principal=principal, assignment_id=represented_assignment_id, function_id=function.id, ) now = utc_now() change = IdmFunctionAssignmentChange( id=new_uuid(), tenant_id=principal.tenant_id, kind=kind, state="submitted", profile=decision.profile, function_id=function.id, organization_unit_id=function.organization_unit_id, candidate_identity_id=candidate_identity_id, candidate_account_id=candidate_account_id, initiator_account_id=principal.account_id, initiator_identity_id=principal.identity_id, represented_assignment_id=represented_assignment_id, justification=str(getattr(payload, "justification")).strip(), evidence=list(getattr(payload, "evidence", ())), requested_valid_from=getattr(payload, "requested_valid_from", None), requested_valid_until=getattr(payload, "requested_valid_until", None), applies_to_subunits=bool(getattr(payload, "applies_to_subunits", False)), assignment_source=str(getattr(payload, "assignment_source", "governance")), required_steps=list(decision.required_steps), completed_steps=[], policy_decision=decision.to_dict(), idempotency_key=idempotency_key, expires_at=now + timedelta(hours=decision.request_expiry_hours), resource_revision=1, metadata_={ **dict(getattr(payload, "metadata", {})), "request_fingerprint": fingerprint, "step_approvals": {}, }, ) workflow = workflow_orchestration_provider(registry) assert workflow is not None try: workflow_ref = workflow.start_standard( session, principal, request=WorkflowStandardStartRequest( tenant_id=principal.tenant_id, origin_module_id="idm", definition_key=f"function-assignment-{kind}", idempotency_key=f"idm-function-change:{change.id}", input={ "change_id": change.id, "kind": kind, "function_id": function.id, "candidate_identity_id": candidate_identity_id, "profile": decision.profile, "required_steps": list(decision.required_steps), }, actor_id=principal.account_id, correlation_id=change.id, ), ) workflow_ref = _align_workflow_to_required_step( session, workflow=workflow, principal=principal, change=change, reference=workflow_ref, ) except (TypeError, ValueError) as exc: raise FunctionAssignmentChangeUnavailable(str(exc)) from exc _pin_workflow(change, workflow_ref) next_step = _next_required_step(change) blocked_reason = _missing_reviewer_reason(session, change) if blocked_reason is not None: change.state = "blocked" change.outcome_reason = blocked_reason elif next_step is None: _apply_assignment( session, change=change, principal=principal, registry=registry, ) else: change.state = STEP_STATE[next_step] session.add(change) session.flush() _append_event( session, change=change, action="submitted", from_state=None, actor=principal, decision=decision, details={"replayed": workflow_ref.replayed}, ) _emit_change_event( session, change, event_type="idm.function_change.submitted.v1", principal=principal, ) _notify_current_state(session, change=change, registry=registry) return change, False def transition_function_assignment_change( session: Session, *, principal: ApiPrincipal, registry: object | None, change: IdmFunctionAssignmentChange, function: OrganizationFunctionRef, action: str, base_revision: int, comment: str | None, evidence: Sequence[str], ) -> IdmFunctionAssignmentChange: if change.state in TERMINAL_STATES: raise FunctionAssignmentChangeConflict( f"Function assignment change is already {change.state}." ) governance_action = _governance_action(change, action) decision = _resolve_transition_decision( session, principal=principal, registry=registry, change=change, function=function, action=governance_action, has_evidence=bool(evidence or change.evidence), ) if not decision.allowed: raise FunctionAssignmentChangeUnavailable( decision.reason or "This transition is not allowed." ) if ( decision.separation_of_duties and action in {"approve", "accept"} and principal.account_id == change.initiator_account_id and not ( action == "accept" and principal.identity_id == change.candidate_identity_id ) ): raise FunctionAssignmentChangeUnavailable( "The initiator cannot perform this approval because separation of duties is enabled." ) next_revision = claim_revision( session, model=IdmFunctionAssignmentChange, filters=( IdmFunctionAssignmentChange.id == change.id, IdmFunctionAssignmentChange.tenant_id == change.tenant_id, ), revision_attribute="resource_revision", expected_revision=base_revision, resource_type="idm_function_assignment_change", resource_id=change.id, refresh_path=f"/api/v1/idm/function-assignment-changes/{change.id}", ) change.resource_revision = next_revision previous_state = change.state if action in {"reject", "withdraw"}: _finish_negative_transition( session, principal=principal, registry=registry, change=change, action=action, comment=comment, evidence=evidence, ) elif action in {"approve", "accept"}: _record_step_approval( change, step=_current_required_step(change), actor_id=principal.account_id, ) if _step_approval_count(change, _current_required_step(change)) >= ( 1 if action == "accept" else decision.quorum ): _complete_current_step( session, principal=principal, registry=registry, change=change, comment=comment, evidence=evidence, ) else: change.policy_decision = decision.to_dict() elif action == "request_changes": _request_changes( session, principal=principal, registry=registry, change=change, comment=comment, evidence=evidence, ) elif action == "respond": if not comment or not comment.strip(): raise FunctionAssignmentChangeConflict( "A response to requested changes requires a comment." ) resume_state = _text(change.metadata_.get("resume_state")) if resume_state not in STEP_STATE.values(): raise FunctionAssignmentChangeConflict( "The previous review state is unavailable." ) change.state = resume_state change.outcome_reason = None change.metadata_ = { **dict(change.metadata_), "last_response": comment.strip(), } elif action == "recover": change.outcome_reason = None _resume_change( session, principal=principal, registry=registry, change=change, ) else: raise FunctionAssignmentChangeConflict( f"Unsupported function assignment change action: {action}." ) _append_event( session, change=change, action=action, from_state=previous_state, actor=principal, decision=decision, comment=comment, evidence=evidence, ) _emit_change_event( session, change, event_type=f"idm.function_change.{action}.v1", principal=principal, ) _notify_current_state(session, change=change, registry=registry) session.flush() return change def available_change_actions( session: Session, *, principal: ApiPrincipal, registry: object | None, change: IdmFunctionAssignmentChange, function: OrganizationFunctionRef, ) -> tuple[list[str], str | None]: if change.state in TERMINAL_STATES: return [], f"The change is {change.state}." candidates = ( ["approve", "request_changes", "reject", "withdraw"] if change.state in {"awaiting_holder", "awaiting_authority"} else ["accept", "request_changes", "reject", "withdraw"] if change.state == "awaiting_recipient" else ["respond", "withdraw"] if change.state == "changes_requested" else ["recover", "withdraw"] ) available: list[str] = [] reasons: list[str] = [] for action in candidates: decision = _resolve_transition_decision( session, principal=principal, registry=registry, change=change, function=function, action=_governance_action(change, action), has_evidence=bool(change.evidence), ) if decision.allowed: available.append(action) elif decision.reason: reasons.append(decision.reason) return available, reasons[0] if not available and reasons else None def change_events( session: Session, *, change_id: str, ) -> list[IdmFunctionAssignmentChangeEvent]: return list( session.scalars( select(IdmFunctionAssignmentChangeEvent) .where(IdmFunctionAssignmentChangeEvent.change_id == change_id) .order_by(IdmFunctionAssignmentChangeEvent.sequence.asc()) ) ) def visible_change_filter( session: Session, principal: ApiPrincipal, ): if principal.has("idm:function_change:admin") or principal.has( "idm:organization_assignment:write" ): return None actor_function_ids = tuple( sorted({item.function_id for item in _actor_assignments(session, principal)}) ) clauses = [ IdmFunctionAssignmentChange.initiator_account_id == principal.account_id, IdmFunctionAssignmentChange.candidate_identity_id == principal.identity_id, IdmFunctionAssignmentChange.candidate_account_id == principal.account_id, ] if actor_function_ids: clauses.extend( ( IdmFunctionAssignmentChange.function_id.in_(actor_function_ids), IdmFunctionAssignmentChange.policy_decision["authority_function_id"] .as_string() .in_(actor_function_ids), ) ) return or_(*clauses) def _resolve_transition_decision( session: Session, *, principal: ApiPrincipal, registry: object | None, change: IdmFunctionAssignmentChange, function: OrganizationFunctionRef, action: FunctionAssignmentGovernanceAction, has_evidence: bool, ) -> FunctionAssignmentGovernanceDecision: policy = function_assignment_governance_policy(registry) if policy is None: raise FunctionAssignmentChangeUnavailable( "Function assignment governance Policy is unavailable." ) settings = _effective_function_settings( session, tenant_id=change.tenant_id, function=function, ) authority_function_id = _authority_function_id(settings) context = _actor_context( session, principal=principal, function_id=change.function_id, authority_function_id=authority_function_id, candidate_identity_id=change.candidate_identity_id, candidate_account_id=change.candidate_account_id, initiator_account_id=change.initiator_account_id, has_evidence=has_evidence, ) context["approvals_complete"] = _next_required_step(change) is None return policy.resolve_function_assignment_action( session, request=FunctionAssignmentGovernanceRequest( tenant_id=change.tenant_id, kind=change.kind, # type: ignore[arg-type] action=action, function_id=change.function_id, actor=principal.to_platform_principal(), candidate_identity_id=change.candidate_identity_id, candidate_account_id=change.candidate_account_id, current_state=change.state, function_settings=settings, context=context, ), ) def _complete_current_step( session: Session, *, principal: ApiPrincipal, registry: object | None, change: IdmFunctionAssignmentChange, comment: str | None, evidence: Sequence[str], ) -> None: step = _current_required_step(change) workflow = workflow_orchestration_provider(registry) if workflow is None or change.workflow_instance_id is None: raise FunctionAssignmentChangeUnavailable( "The pinned Workflow instance is unavailable." ) reference = workflow.resolve_current_step( session, principal, tenant_id=change.tenant_id, instance_id=change.workflow_instance_id, resolution=WorkflowCurrentStepResolution( action="approve", expected_step_id=change.workflow_current_step_id, actor_id=principal.account_id, output={"change_id": change.id, "completed_step": step}, evidence=tuple(evidence), comment=comment, ), ) change.completed_steps = [*change.completed_steps, step] reference = _align_workflow_to_required_step( session, workflow=workflow, principal=principal, change=change, reference=reference, ) _pin_workflow(change, reference) next_step = _next_required_step(change) if next_step is None: try: _apply_assignment( session, change=change, principal=principal, registry=registry, ) except FunctionAssignmentChangeConflict as exc: change.state = "failed_manual_review" change.outcome_reason = str(exc) else: blocked_reason = _missing_reviewer_reason(session, change) if blocked_reason is not None: change.state = "blocked" change.outcome_reason = blocked_reason else: change.state = STEP_STATE[next_step] change.outcome_reason = None def _finish_negative_transition( session: Session, *, principal: ApiPrincipal, registry: object | None, change: IdmFunctionAssignmentChange, action: str, comment: str | None, evidence: Sequence[str], ) -> None: workflow = workflow_orchestration_provider(registry) if workflow is None or change.workflow_instance_id is None: raise FunctionAssignmentChangeUnavailable( "The pinned Workflow instance is unavailable." ) resolution_action = "reject" if action == "reject" else "cancel" reference = workflow.resolve_current_step( session, principal, tenant_id=change.tenant_id, instance_id=change.workflow_instance_id, resolution=WorkflowCurrentStepResolution( action=resolution_action, expected_step_id=change.workflow_current_step_id, actor_id=principal.account_id, output={"change_id": change.id}, evidence=tuple(evidence), comment=comment, ), ) _pin_workflow(change, reference) change.state = "rejected" if action == "reject" else "withdrawn" change.outcome_reason = comment def _request_changes( session: Session, *, principal: ApiPrincipal, registry: object | None, change: IdmFunctionAssignmentChange, comment: str | None, evidence: Sequence[str], ) -> None: if not comment or not comment.strip(): raise FunctionAssignmentChangeConflict("Requesting changes requires a comment.") workflow = workflow_orchestration_provider(registry) if workflow is None or change.workflow_instance_id is None: raise FunctionAssignmentChangeUnavailable( "The pinned Workflow instance is unavailable." ) previous_state = change.state reference = workflow.resolve_current_step( session, principal, tenant_id=change.tenant_id, instance_id=change.workflow_instance_id, resolution=WorkflowCurrentStepResolution( action="changes", expected_step_id=change.workflow_current_step_id, actor_id=principal.account_id, output={"change_id": change.id}, evidence=tuple(evidence), comment=comment, ), ) _pin_workflow(change, reference) change.state = "changes_requested" change.outcome_reason = comment.strip() change.metadata_ = { **dict(change.metadata_), "resume_state": previous_state, } def _resume_change( session: Session, *, principal: ApiPrincipal, registry: object | None, change: IdmFunctionAssignmentChange, ) -> None: workflow = workflow_orchestration_provider(registry) if workflow is None or change.workflow_instance_id is None: raise FunctionAssignmentChangeUnavailable( "The pinned Workflow instance is unavailable." ) reference = workflow.get_instance( session, tenant_id=change.tenant_id, instance_id=change.workflow_instance_id, ) reference = _align_workflow_to_required_step( session, workflow=workflow, principal=principal, change=change, reference=reference, ) _pin_workflow(change, reference) step = _next_required_step(change) blocked_reason = _missing_reviewer_reason(session, change) if blocked_reason is not None: change.state = "blocked" change.outcome_reason = blocked_reason elif step is not None: change.state = STEP_STATE[step] change.outcome_reason = None else: try: _apply_assignment( session, change=change, principal=principal, registry=registry, ) except FunctionAssignmentChangeConflict as exc: change.state = "failed_manual_review" change.outcome_reason = str(exc) def _align_workflow_to_required_step( session: Session, *, workflow: object, principal: ApiPrincipal, change: IdmFunctionAssignmentChange, reference: WorkflowInstanceRef, ) -> WorkflowInstanceRef: required = set(change.required_steps) completed = set(change.completed_steps) current = reference for _ in range(4): step = NODE_STEP.get(current.current_node_id or "") if step is None or (step in required and step not in completed): return current current = workflow.resolve_current_step( # type: ignore[attr-defined] session, principal, tenant_id=change.tenant_id, instance_id=current.id, resolution=WorkflowCurrentStepResolution( action="approve", expected_step_id=current.current_step_id, actor_id=principal.account_id, output={"skipped": True, "step": step}, comment="Skipped by the effective assignment governance profile.", ), ) raise FunctionAssignmentChangeConflict( "Workflow baseline did not converge on the required governance step." ) def _apply_assignment( session: Session, *, change: IdmFunctionAssignmentChange, principal: ApiPrincipal, registry: object | None, ) -> None: if change.resulting_assignment_id: change.state = "applied" return existing = session.scalar( select(IdmOrganizationFunctionAssignment).where( IdmOrganizationFunctionAssignment.tenant_id == change.tenant_id, IdmOrganizationFunctionAssignment.identity_id == change.candidate_identity_id, IdmOrganizationFunctionAssignment.function_id == change.function_id, IdmOrganizationFunctionAssignment.organization_unit_id == change.organization_unit_id, ) ) if existing is not None and existing.is_active: raise FunctionAssignmentChangeConflict( "The candidate already holds this organization function." ) assignment = existing or IdmOrganizationFunctionAssignment( tenant_id=change.tenant_id, identity_id=change.candidate_identity_id, function_id=change.function_id, organization_unit_id=change.organization_unit_id, ) assignment.account_id = change.candidate_account_id assignment.applies_to_subunits = change.applies_to_subunits assignment.source = change.assignment_source assignment.delegated_from_assignment_id = change.represented_assignment_id assignment.valid_from = change.requested_valid_from assignment.valid_until = change.requested_valid_until assignment.is_active = True assignment.settings = { **dict(assignment.settings or {}), "governance": { "change_id": change.id, "workflow_instance_id": change.workflow_instance_id, "profile": change.profile, "applied_by": principal.account_id, "applied_at": utc_now().isoformat(), }, } try: validate_assignment_shape(assignment) except AssignmentTransitionError as exc: raise FunctionAssignmentChangeConflict(str(exc)) from exc if existing is None: session.add(assignment) session.flush() change.resulting_assignment_id = assignment.id change.state = "applied" change.outcome_reason = None emit_assignment_event( session, assignment, event_type="idm.function_assignment.created.v1", actor_type="account", actor_id=principal.account_id, ) invalidate_auth_principals( session, tenant_id=change.tenant_id, source_module="idm", resource_type="organization_function_assignment", resource_id=assignment.id, ) _emit_change_event( session, change, event_type="idm.function_change.applied.v1", principal=principal, ) del registry def _actor_context( session: Session, *, principal: ApiPrincipal, function_id: str, authority_function_id: str | None, candidate_identity_id: str, candidate_account_id: str | None, initiator_account_id: str, has_evidence: bool, ) -> dict[str, object]: actor_assignments = _actor_assignments(session, principal) return { "actor_is_holder": any( item.function_id == function_id for item in actor_assignments ), "actor_is_authority": bool(authority_function_id) and any( item.function_id == authority_function_id for item in actor_assignments ), "candidate_is_actor": ( principal.identity_id == candidate_identity_id or ( candidate_account_id is not None and principal.account_id == candidate_account_id ) ), "actor_is_initiator": principal.account_id == initiator_account_id, "has_evidence": has_evidence, } def _actor_assignments( session: Session, principal: ApiPrincipal, ) -> list[IdmOrganizationFunctionAssignment]: now = utc_now() return list( session.scalars( select(IdmOrganizationFunctionAssignment).where( IdmOrganizationFunctionAssignment.tenant_id == principal.tenant_id, IdmOrganizationFunctionAssignment.is_active.is_(True), or_( IdmOrganizationFunctionAssignment.account_id == principal.account_id, IdmOrganizationFunctionAssignment.identity_id == principal.identity_id, ), or_( IdmOrganizationFunctionAssignment.valid_from.is_(None), IdmOrganizationFunctionAssignment.valid_from <= now, ), or_( IdmOrganizationFunctionAssignment.valid_until.is_(None), IdmOrganizationFunctionAssignment.valid_until > now, ), ) ) ) def _effective_function_settings( session: Session, *, tenant_id: str, function: OrganizationFunctionRef, ) -> dict[str, object]: tenant_settings = session.get(IdmTenantSettings, tenant_id) defaults: dict[str, object] = {} if tenant_settings is not None: raw_defaults = tenant_settings.settings.get( "function_assignment_governance_defaults" ) if isinstance(raw_defaults, Mapping): defaults = dict(raw_defaults) function_policy = function.settings.get("assignment_governance") return { **dict(function.settings), "assignment_governance": { **defaults, **(dict(function_policy) if isinstance(function_policy, Mapping) else {}), }, } def _authority_function_id(settings: Mapping[str, object]) -> str | None: raw = settings.get("assignment_governance") policy = raw if isinstance(raw, Mapping) else {} return _text(policy.get("authority_function_id")) def _validate_requested_validity( payload: object, decision: FunctionAssignmentGovernanceDecision, ) -> None: valid_from = getattr(payload, "requested_valid_from", None) valid_until = getattr(payload, "requested_valid_until", None) if valid_from is not None and valid_until is not None and valid_until <= valid_from: raise FunctionAssignmentChangeConflict( "Requested valid until must be after valid from." ) if decision.maximum_validity_days is not None and valid_until is not None: start = valid_from or utc_now() if valid_until > start + timedelta(days=decision.maximum_validity_days): raise FunctionAssignmentChangeUnavailable( f"Requested validity exceeds the Policy limit of " f"{decision.maximum_validity_days} days." ) def _require_actor_assignment( session: Session, *, principal: ApiPrincipal, assignment_id: str, function_id: str, ) -> None: assignment = session.get(IdmOrganizationFunctionAssignment, assignment_id) if ( assignment is None or assignment.tenant_id != principal.tenant_id or assignment.function_id != function_id or not assignment.is_active or ( assignment.account_id != principal.account_id and assignment.identity_id != principal.identity_id ) ): raise FunctionAssignmentChangeUnavailable( "The represented function assignment is not an effective assignment of the actor." ) def _next_required_step(change: IdmFunctionAssignmentChange) -> str | None: completed = set(change.completed_steps) return next( (step for step in change.required_steps if step not in completed), None, ) def _current_required_step(change: IdmFunctionAssignmentChange) -> str: step = _next_required_step(change) if step is None: raise FunctionAssignmentChangeConflict( "All required governance steps are already complete." ) return step def _governance_action( change: IdmFunctionAssignmentChange, action: str, ) -> FunctionAssignmentGovernanceAction: if action == "approve": step = _current_required_step(change) if step == "holder": return "approve_holder" if step == "authority": return "approve_authority" raise FunctionAssignmentChangeConflict( "Recipient approval must use the accept action." ) if action == "accept": if _current_required_step(change) != "recipient": raise FunctionAssignmentChangeConflict( "The change is not awaiting recipient acceptance." ) return "accept_recipient" if action in { "reject", "request_changes", "respond", "withdraw", "recover", }: return action # type: ignore[return-value] raise FunctionAssignmentChangeConflict( f"Unsupported function assignment change action: {action}." ) def _record_step_approval( change: IdmFunctionAssignmentChange, *, step: str, actor_id: str, ) -> None: approvals = dict(change.metadata_.get("step_approvals") or {}) actors = [str(item) for item in approvals.get(step, ())] if actor_id in actors: raise FunctionAssignmentChangeConflict( "This actor already approved the current governance step." ) approvals[step] = [*actors, actor_id] change.metadata_ = {**dict(change.metadata_), "step_approvals": approvals} def _step_approval_count( change: IdmFunctionAssignmentChange, step: str, ) -> int: approvals = change.metadata_.get("step_approvals") or {} return len(approvals.get(step, ())) if isinstance(approvals, Mapping) else 0 def _pin_workflow( change: IdmFunctionAssignmentChange, reference: WorkflowInstanceRef, ) -> None: change.workflow_definition_id = reference.definition_id change.workflow_definition_revision_id = reference.definition_revision_id change.workflow_definition_revision = reference.definition_revision change.workflow_definition_hash = reference.definition_hash change.workflow_instance_id = reference.id change.workflow_current_step_id = reference.current_step_id def _append_event( session: Session, *, change: IdmFunctionAssignmentChange, action: str, from_state: str | None, actor: ApiPrincipal, decision: FunctionAssignmentGovernanceDecision, comment: str | None = None, evidence: Sequence[str] = (), details: Mapping[str, object] | None = None, ) -> None: sequence = ( int( session.scalar( select(func.max(IdmFunctionAssignmentChangeEvent.sequence)).where( IdmFunctionAssignmentChangeEvent.change_id == change.id ) ) or 0 ) + 1 ) session.add( IdmFunctionAssignmentChangeEvent( tenant_id=change.tenant_id, change_id=change.id, sequence=sequence, action=action, from_state=from_state, to_state=change.state, actor_account_id=actor.account_id, actor_identity_id=actor.identity_id, actor_assignment_id=( change.represented_assignment_id if actor.account_id == change.initiator_account_id else None ), comment=comment, evidence=list(evidence), policy_decision=decision.to_dict(), workflow_step_id=change.workflow_current_step_id, details=dict(details or {}), created_at=utcnow(), ) ) def _emit_change_event( session: Session, change: IdmFunctionAssignmentChange, *, event_type: str, principal: ApiPrincipal, ) -> None: emit_platform_event( session, PlatformEvent( type=event_type, module_id="idm", payload={ "kind": change.kind, "state": change.state, "profile": change.profile, "function_id": change.function_id, "candidate_identity_id": change.candidate_identity_id, "workflow_instance_id": change.workflow_instance_id, "resulting_assignment_id": change.resulting_assignment_id, "resource_revision": change.resource_revision, }, actor=EventActorRef(type="account", id=principal.account_id), tenant=EventTenantRef(id=change.tenant_id), subject=EventObjectRef( type="organization_function", id=change.function_id, ), resource=EventObjectRef( type="function_assignment_change", id=change.id, ), classification="internal", ), ) def _notify_current_state( session: Session, *, change: IdmFunctionAssignmentChange, registry: object | None, ) -> None: provider = notification_dispatch_provider(registry) if provider is None: return recipients = _notification_recipients(session, change) for account_id in recipients: provider.enqueue_notification( session, NotificationDispatchRequest( tenant_id=change.tenant_id, source_module="idm", source_resource_type="function_assignment_change", source_resource_id=change.id, event_kind=f"function_assignment_change.{change.state}", recipient_type="account", recipient_id=account_id, subject=f"Function assignment change: {change.state.replace('_', ' ')}", body_text=( f"A {change.kind} for organization function " f"{change.function_id} is {change.state.replace('_', ' ')}." ), action_url=f"/idm?change={change.id}", payload={"change_id": change.id, "state": change.state}, ), ) def _notification_recipients( session: Session, change: IdmFunctionAssignmentChange, ) -> tuple[str, ...]: recipients = {change.initiator_account_id} if change.candidate_account_id: recipients.add(change.candidate_account_id) if change.state == "awaiting_holder": recipients.update( _function_holder_accounts( session, tenant_id=change.tenant_id, function_id=change.function_id, ) ) elif change.state == "awaiting_authority": authority_id = change.policy_decision.get("authority_function_id") if authority_id: recipients.update( _function_holder_accounts( session, tenant_id=change.tenant_id, function_id=str(authority_id), ) ) return tuple(sorted(item for item in recipients if item)) def _function_holder_accounts( session: Session, *, tenant_id: str, function_id: str, ) -> tuple[str, ...]: now = utc_now() return tuple( item for item in session.scalars( select(IdmOrganizationFunctionAssignment.account_id).where( IdmOrganizationFunctionAssignment.tenant_id == tenant_id, IdmOrganizationFunctionAssignment.function_id == function_id, IdmOrganizationFunctionAssignment.account_id.is_not(None), IdmOrganizationFunctionAssignment.is_active.is_(True), or_( IdmOrganizationFunctionAssignment.valid_from.is_(None), IdmOrganizationFunctionAssignment.valid_from <= now, ), or_( IdmOrganizationFunctionAssignment.valid_until.is_(None), IdmOrganizationFunctionAssignment.valid_until > now, ), ) ) if item is not None ) def _missing_reviewer_reason( session: Session, change: IdmFunctionAssignmentChange, ) -> str | None: step = _next_required_step(change) if step == "holder" and not _function_has_incumbent( session, tenant_id=change.tenant_id, function_id=change.function_id, ): return "The function is vacant; no effective holder can review this change." if step == "authority": authority_id = _text(change.policy_decision.get("authority_function_id")) if authority_id is None: return "The effective Policy does not designate an authority function." if not _function_has_incumbent( session, tenant_id=change.tenant_id, function_id=authority_id, ): return "The designated authority function is vacant." if step == "recipient" and not ( change.candidate_account_id or change.candidate_identity_id ): return "Recipient acceptance has no resolvable candidate." return None def _function_has_incumbent( session: Session, *, tenant_id: str, function_id: str, ) -> bool: now = utc_now() return bool( session.scalar( select(func.count(IdmOrganizationFunctionAssignment.id)).where( IdmOrganizationFunctionAssignment.tenant_id == tenant_id, IdmOrganizationFunctionAssignment.function_id == function_id, IdmOrganizationFunctionAssignment.is_active.is_(True), or_( IdmOrganizationFunctionAssignment.valid_from.is_(None), IdmOrganizationFunctionAssignment.valid_from <= now, ), or_( IdmOrganizationFunctionAssignment.valid_until.is_(None), IdmOrganizationFunctionAssignment.valid_until > now, ), ) ) ) def _request_fingerprint(payload: object) -> str: if hasattr(payload, "model_dump"): value = payload.model_dump(mode="json", exclude={"idempotency_key"}) else: value = vars(payload) encoded = json.dumps(value, sort_keys=True, separators=(",", ":")) return hashlib.sha256(encoded.encode("utf-8")).hexdigest() def _text(value: object) -> str | None: text = str(value).strip() if value is not None else "" return text or None __all__ = [ "FunctionAssignmentChangeConflict", "FunctionAssignmentChangeError", "FunctionAssignmentChangeUnavailable", "OPEN_STATES", "TERMINAL_STATES", "available_change_actions", "change_events", "create_function_assignment_change", "resolve_submission_capability", "transition_function_assignment_change", "visible_change_filter", ]