from __future__ import annotations from collections.abc import Mapping, Sequence from datetime import datetime, 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, ) from govoplan_idm.backend.delegation_routes import ( DelegationRoute, resolve_actor_function_route, resolve_function_route_availability, validate_delegation_chain, ) OPEN_STATES = { "submitted", "awaiting_holder", "awaiting_authority", "awaiting_recipient", "changes_requested", "blocked", "failed_manual_review", "escalated", } 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, ) base_context = _base_actor_context( principal=principal, candidate_identity_id=candidate_identity_id, candidate_account_id=candidate_account_id, initiator_account_id=principal.account_id, has_evidence=has_evidence, ) preliminary = 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=base_context, ), ) context = _actor_context( session, principal=principal, function_id=function.id, authority_function_id=preliminary.authority_function_id, escalation_target_function_id=None, candidate_identity_id=candidate_identity_id, candidate_account_id=candidate_account_id, initiator_account_id=principal.account_id, has_evidence=has_evidence, decision=preliminary, ) 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) assignment_source = str(getattr(payload, "assignment_source", "governance")) represented_assignment_id = _text( getattr(payload, "represented_assignment_id", None) ) if assignment_source == "delegated" and represented_assignment_id is None: raise FunctionAssignmentChangeUnavailable( "A delegated assignment requires the represented source assignment." ) if assignment_source != "delegated" and represented_assignment_id is not None: raise FunctionAssignmentChangeUnavailable( "Only a delegated assignment may reference a represented source assignment." ) if represented_assignment_id is not None: _require_actor_assignment( session, principal=principal, assignment_id=represented_assignment_id, function_id=function.id, decision=decision, ) 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=assignment_source, 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, decision=decision) if blocked_reason is not None: change.state = "blocked" change.outcome_reason = blocked_reason _clear_review_route(change) elif next_step is None: _apply_assignment( session, change=change, principal=principal, registry=registry, function=function, ) else: change.state = STEP_STATE[next_step] _set_review_route(change, decision=decision, step=next_step, now=now) 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." ) change.policy_decision = decision.to_dict() 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), principal=principal, decision=decision, ) 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, function=function, decision=decision, comment=comment, evidence=evidence, ) 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(), } _set_review_route( change, decision=decision, step=_current_required_step(change), now=utc_now(), ) elif action == "recover": change.outcome_reason = None _resume_change( session, principal=principal, registry=registry, change=change, function=function, decision=decision, ) 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", "escalated"} 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), IdmFunctionAssignmentChange.escalation_target_function_id.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, ) base_context = _base_actor_context( principal=principal, 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, ) preliminary = 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=base_context, ), ) context = _actor_context( session, principal=principal, function_id=change.function_id, authority_function_id=preliminary.authority_function_id, escalation_target_function_id=change.escalation_target_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, decision=preliminary, ) 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, function: OrganizationFunctionRef, decision: FunctionAssignmentGovernanceDecision, 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, function=function, ) except FunctionAssignmentChangeConflict as exc: change.state = "failed_manual_review" change.outcome_reason = str(exc) else: blocked_reason = _missing_reviewer_reason( session, change, decision=decision, ) if blocked_reason is not None: change.state = "blocked" change.outcome_reason = blocked_reason _clear_review_route(change) else: change.state = STEP_STATE[next_step] change.outcome_reason = None _set_review_route( change, decision=decision, step=next_step, now=utc_now(), ) 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 _clear_review_route(change) 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.escalation_from_state if change.state == "escalated" and change.escalation_from_state else 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, } _clear_review_route(change) def _resume_change( session: Session, *, principal: ApiPrincipal, registry: object | None, change: IdmFunctionAssignmentChange, function: OrganizationFunctionRef, decision: FunctionAssignmentGovernanceDecision, ) -> 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, decision=decision, ) if blocked_reason is not None: change.state = "blocked" change.outcome_reason = blocked_reason _clear_review_route(change) elif step is not None: change.state = STEP_STATE[step] change.outcome_reason = None _set_review_route( change, decision=decision, step=step, now=utc_now(), ) else: try: _apply_assignment( session, change=change, principal=principal, registry=registry, function=function, ) 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 _recheck_application( session: Session, *, change: IdmFunctionAssignmentChange, principal: ApiPrincipal, registry: object | None, function: OrganizationFunctionRef, ) -> FunctionAssignmentGovernanceDecision: policy = function_assignment_governance_policy(registry) if policy is None: raise FunctionAssignmentChangeConflict( "Function assignment governance Policy is unavailable at application." ) settings = _effective_function_settings( session, tenant_id=change.tenant_id, function=function, ) context = _base_actor_context( principal=principal, candidate_identity_id=change.candidate_identity_id, candidate_account_id=change.candidate_account_id, initiator_account_id=change.initiator_account_id, has_evidence=bool(change.evidence), ) preliminary = policy.resolve_function_assignment_action( session, request=FunctionAssignmentGovernanceRequest( tenant_id=change.tenant_id, kind=change.kind, # type: ignore[arg-type] action="apply", 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, ), ) if tuple(change.required_steps) != preliminary.required_steps: raise FunctionAssignmentChangeConflict( "The effective Policy review steps changed after submission; " "administrative recovery must re-plan the pinned workflow." ) _recheck_completed_approvals( session, change=change, decision=preliminary, ) _recheck_delegated_assignment_source( session, change=change, function=function, decision=preliminary, ) context["approvals_complete"] = True decision = policy.resolve_function_assignment_action( session, request=FunctionAssignmentGovernanceRequest( tenant_id=change.tenant_id, kind=change.kind, # type: ignore[arg-type] action="apply", 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, ), ) if not decision.allowed: raise FunctionAssignmentChangeConflict( decision.reason or "The current Policy blocks application." ) change.policy_decision = decision.to_dict() return decision def _recheck_completed_approvals( session: Session, *, change: IdmFunctionAssignmentChange, decision: FunctionAssignmentGovernanceDecision, ) -> None: approvals = change.metadata_.get("step_approvals") approval_map = approvals if isinstance(approvals, Mapping) else {} for step in decision.required_steps: records = list(approval_map.get(step, ())) required = 1 if step == "recipient" else decision.quorum valid = 0 failure_reason: str | None = None for raw in records: record = raw if isinstance(raw, Mapping) else {"actor_account_id": raw} account_id = _text(record.get("actor_account_id")) identity_id = _text(record.get("actor_identity_id")) if step == "recipient": if ( (account_id and account_id == change.candidate_account_id) or (identity_id and identity_id == change.candidate_identity_id) ): valid += 1 else: failure_reason = "Recipient acceptance no longer resolves to the candidate." continue route_kind = _text(record.get("route_kind")) or step expected_function_id = ( change.function_id if step == "holder" and route_kind != "escalation" else decision.authority_function_id if step == "authority" and route_kind != "escalation" else _text(record.get("expected_function_id")) ) if route_kind == "escalation": current_rule = decision.escalation_rule(step) # type: ignore[arg-type] if ( current_rule is None or expected_function_id != current_rule.target_function_id ): failure_reason = ( "The effective Policy no longer permits the recorded " f"{step} escalation route." ) continue elif step == "authority" and expected_function_id != _text( record.get("expected_function_id") ): failure_reason = ( "The designated authority changed after its approval was recorded." ) continue if expected_function_id is None: failure_reason = f"The {step} approval route is unavailable." continue route = resolve_actor_function_route( session, tenant_id=change.tenant_id, function_id=expected_function_id, account_id=account_id, identity_id=identity_id, decision=decision, ) if route.effective: valid += 1 else: failure_reason = route.reason if valid < required: raise FunctionAssignmentChangeConflict( failure_reason or f"The current {step} approval quorum is no longer effective." ) def _recheck_delegated_assignment_source( session: Session, *, change: IdmFunctionAssignmentChange, function: OrganizationFunctionRef, decision: FunctionAssignmentGovernanceDecision, ) -> None: if change.assignment_source != "delegated": if change.represented_assignment_id is not None: raise FunctionAssignmentChangeConflict( "A non-delegated change cannot retain a represented assignment." ) return if not function.delegable or not decision.delegation_allowed: raise FunctionAssignmentChangeConflict( "The organization function or current Policy no longer permits delegation." ) source = ( session.get( IdmOrganizationFunctionAssignment, change.represented_assignment_id, ) if change.represented_assignment_id else None ) if source is None: raise FunctionAssignmentChangeConflict( "The represented source assignment is unavailable." ) if ( source.account_id != change.initiator_account_id and source.identity_id != change.initiator_identity_id ): raise FunctionAssignmentChangeConflict( "The represented source assignment no longer belongs to the initiator." ) route = validate_delegation_chain( session, assignment=source, tenant_id=change.tenant_id, function_id=change.function_id, decision=decision, ) if not route.effective: raise FunctionAssignmentChangeConflict( route.reason or "The represented delegation route is no longer effective." ) if ( source.valid_from is not None and ( change.requested_valid_from is None or change.requested_valid_from < source.valid_from ) ): raise FunctionAssignmentChangeConflict( "The delegated assignment would start before its source assignment." ) if ( source.valid_until is not None and ( change.requested_valid_until is None or change.requested_valid_until > source.valid_until ) ): raise FunctionAssignmentChangeConflict( "The delegated assignment would outlive its source assignment." ) if ( decision.maximum_delegated_validity_days is not None and change.requested_valid_until is not None ): start = change.requested_valid_from or utc_now() if change.requested_valid_until > start + timedelta( days=decision.maximum_delegated_validity_days ): raise FunctionAssignmentChangeConflict( "The delegated validity window exceeds the current Policy ceiling." ) def _apply_assignment( session: Session, *, change: IdmFunctionAssignmentChange, principal: ApiPrincipal, registry: object | None, function: OrganizationFunctionRef, ) -> None: if change.resulting_assignment_id: change.state = "applied" return _recheck_application( session, change=change, principal=principal, registry=registry, function=function, ) 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 _clear_review_route(change) 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, escalation_target_function_id: str | None, candidate_identity_id: str, candidate_account_id: str | None, initiator_account_id: str, has_evidence: bool, decision: FunctionAssignmentGovernanceDecision, ) -> dict[str, object]: holder_route = resolve_actor_function_route( session, tenant_id=principal.tenant_id, function_id=function_id, account_id=principal.account_id, identity_id=principal.identity_id, decision=decision, ) authority_route = ( resolve_actor_function_route( session, tenant_id=principal.tenant_id, function_id=authority_function_id, account_id=principal.account_id, identity_id=principal.identity_id, decision=decision, ) if authority_function_id else DelegationRoute( False, "unavailable", "The effective Policy does not designate an authority function.", ) ) escalation_route = ( resolve_actor_function_route( session, tenant_id=principal.tenant_id, function_id=escalation_target_function_id, account_id=principal.account_id, identity_id=principal.identity_id, decision=decision, ) if escalation_target_function_id else DelegationRoute( False, "unavailable", "This review is not currently escalated to a target function.", ) ) return { **_base_actor_context( principal=principal, candidate_identity_id=candidate_identity_id, candidate_account_id=candidate_account_id, initiator_account_id=initiator_account_id, has_evidence=has_evidence, ), "actor_is_holder": holder_route.effective, "actor_is_authority": authority_route.effective, "actor_is_escalation_target": escalation_route.effective, "actor_routes": { "holder": holder_route.to_dict(), "authority": authority_route.to_dict(), "escalation": escalation_route.to_dict(), }, } def _base_actor_context( *, principal: ApiPrincipal, candidate_identity_id: str, candidate_account_id: str | None, initiator_account_id: str, has_evidence: bool, ) -> dict[str, object]: return { "actor_is_holder": False, "actor_is_authority": False, "actor_is_escalation_target": False, "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, "actor_routes": {}, } 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 _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." ) if str(getattr(payload, "assignment_source", "governance")) != "delegated": return if not decision.delegation_allowed: raise FunctionAssignmentChangeUnavailable( "The current Policy does not permit delegated assignments." ) if ( decision.maximum_delegated_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_delegated_validity_days ): raise FunctionAssignmentChangeUnavailable( "Requested delegated validity exceeds the Policy limit of " f"{decision.maximum_delegated_validity_days} days." ) def _require_actor_assignment( session: Session, *, principal: ApiPrincipal, assignment_id: str, function_id: str, decision: FunctionAssignmentGovernanceDecision, ) -> 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 ( 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." ) route = validate_delegation_chain( session, assignment=assignment, tenant_id=principal.tenant_id, function_id=function_id, decision=decision, ) if not route.effective: raise FunctionAssignmentChangeUnavailable( route.reason or "The represented function assignment route is not effective." ) 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 _set_review_route( change: IdmFunctionAssignmentChange, *, decision: FunctionAssignmentGovernanceDecision, step: str, now: datetime, ) -> None: rule = decision.escalation_rule(step) # type: ignore[arg-type] change.review_deadline_at = ( now + timedelta(hours=rule.timeout_hours) if rule is not None else None ) change.escalation_target_function_id = ( rule.target_function_id if rule is not None else None ) change.escalated_at = None change.escalation_from_state = None def _clear_review_route(change: IdmFunctionAssignmentChange) -> None: change.review_deadline_at = None change.escalated_at = None change.escalation_from_state = None change.escalation_target_function_id = None def _governance_action( change: IdmFunctionAssignmentChange, action: str, ) -> FunctionAssignmentGovernanceAction: if action == "approve": if change.state == "escalated": return "approve_escalation" 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, principal: ApiPrincipal, decision: FunctionAssignmentGovernanceDecision, ) -> None: approvals = dict(change.metadata_.get("step_approvals") or {}) records = list(approvals.get(step, ())) actors = [ str(item.get("actor_account_id")) if isinstance(item, Mapping) else str(item) for item in records ] if principal.account_id in actors: raise FunctionAssignmentChangeConflict( "This actor already approved the current governance step." ) route_key = "escalation" if change.state == "escalated" else step routes = decision.details.get("actor_routes") route = routes.get(route_key) if isinstance(routes, Mapping) else None route_data = dict(route) if isinstance(route, Mapping) else {} approvals[step] = [ *records, { "actor_account_id": principal.account_id, "actor_identity_id": principal.identity_id, "approved_at": utc_now().isoformat(), "route_kind": route_key, "expected_function_id": ( change.escalation_target_function_id if route_key == "escalation" else change.function_id if step == "holder" else decision.authority_function_id if step == "authority" else None ), "assignment_id": route_data.get("assignment_id"), "chain_assignment_ids": list( route_data.get("chain_assignment_ids") or () ), "delegation_depth": route_data.get("delegation_depth", 0), }, ] 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), ) ) elif change.state == "escalated" and change.escalation_target_function_id: recipients.update( _function_holder_accounts( session, tenant_id=change.tenant_id, function_id=change.escalation_target_function_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, *, decision: FunctionAssignmentGovernanceDecision, ) -> str | None: step = _next_required_step(change) if step == "holder": route = resolve_function_route_availability( session, tenant_id=change.tenant_id, function_id=change.function_id, decision=decision, ) if not route.effective: return route.reason or "No effective holder can review this change." if step == "authority": authority_id = decision.authority_function_id if authority_id is None: return "The effective Policy does not designate an authority function." route = resolve_function_route_availability( session, tenant_id=change.tenant_id, function_id=authority_id, decision=decision, ) if not route.effective: return route.reason or "The designated authority function is unavailable." if step == "recipient" and not ( change.candidate_account_id or change.candidate_identity_id ): return "Recipient acceptance has no resolvable candidate." return None 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", ]