refactor(runtime): satisfy Sonar maintainability limits

implementations/python/packages/raes_runtime/manager.py

@@ -28,8 +28,7 @@
 
 _RUNTIME_APPLY_ADDRESS, _APPLY_EVALUATOR_ADDRESS = "runtime.apply", "runtime.apply.evaluator"
 _APPLY_ORCHESTRATOR_ADDRESS = "runtime.apply.orchestrator"
-_APPLY_PHASE_FAILED = "runtime.apply-phase-failed"
-_DESTROY_PHASE_FAILED = "runtime.destroy-phase-failed"
+_APPLY_PHASE_FAILED, _DESTROY_PHASE_FAILED = "runtime.apply-phase-failed", "runtime.destroy-phase-failed"
 _ROLLBACK_EVALUATOR_ADDRESS = "runtime.rollback.evaluator"
 _ROLLBACK_ORCHESTRATOR_ADDRESS = "runtime.rollback.orchestrator"
 

implementations/python/packages/raes_runtime/participant_crossing_mediation.py

@@ -28,6 +28,7 @@
 )
 from raes_contracts.contracts.participant_runtime import ParticipantRuntimeOrderingBasis
 from raes_contracts.diagnostics import Diagnostic
+from raes_contracts.participant_opacity_runtime import validate_participant_opacity_runtime_enforcement
 from raes_contracts.planning import RuntimeDomain
 from raes_contracts.runtime_state import OperationReceipt, OperationState, OperationStatus, RuntimeSnapshot
 from raes_contracts.vocabulary import ParticipantFeatureSupportLevel
@@ -177,6 +178,47 @@ class PreparedParticipantCrossing:
     existing_receipt: OperationReceipt | None = None
 
 
+def _resolve_crossing_policy(
+    control_plane: object,
+    intent: ParticipantCrossingIntent,
+    resolver: ParticipantCrossingPolicyResolver,
+    incumbent_carrier: object | None,
+) -> ParticipantCrossingPolicyResolution:
+    operation_resolver = getattr(resolver, "resolve_operation", None)
+    resolution = (
+        operation_resolver(intent, control_plane._snapshot, incumbent_carrier)
+        if callable(operation_resolver)
+        else resolver.resolve(intent, control_plane._snapshot)
+    )
+    context = resolver.validation_context(
+        control_plane._snapshot,
+        intent.participant_address,
+    )
+    resolution = bind_active_participant_opacity_support(
+        resolution,
+        context.opacity_enforcement_supports,
+    )
+    if resolution.opacity_enforcement is None:
+        return resolution
+    support = next(
+        (
+            candidate
+            for candidate in context.opacity_enforcement_supports
+            if candidate.binding == resolution.opacity_enforcement
+        ),
+        None,
+    )
+    if support is None:
+        raise ValueError("participant opacity runtime binding is not admitted by the resolver context")
+    validate_participant_opacity_runtime_enforcement(
+        resolution.opacity_enforcement,
+        support=support,
+        participant_address=intent.participant_address,
+        audience_scope_ref=intent.audience_scope_ref,
+    )
+    return normalize_participant_opacity_resolution(resolution)
+
+
 def prepare_participant_crossing(
     control_plane: object,
     intent: ParticipantCrossingIntent,
@@ -219,42 +261,12 @@ def prepare_participant_crossing(
         _require_replay_state_cut(existing, expected_heads)
         fingerprint_heads = existing.decision_history_heads
     try:
-        operation_resolver = getattr(resolver, "resolve_operation", None)
-        resolution = (
-            operation_resolver(intent, control_plane._snapshot, incumbent_carrier)
-            if callable(operation_resolver)
-            else resolver.resolve(intent, control_plane._snapshot)
-        )
-        context = resolver.validation_context(
-            control_plane._snapshot,
-            intent.participant_address,
-        )
-        resolution = bind_active_participant_opacity_support(
-            resolution,
-            context.opacity_enforcement_supports,
+        resolution = _resolve_crossing_policy(
+            control_plane,
+            intent,
+            resolver,
+            incumbent_carrier,
         )
-        if resolution.opacity_enforcement is not None:
-            from raes_contracts.participant_opacity_runtime import (
-                validate_participant_opacity_runtime_enforcement,
-            )
-
-            support = next(
-                (
-                    candidate
-                    for candidate in context.opacity_enforcement_supports
-                    if candidate.binding == resolution.opacity_enforcement
-                ),
-                None,
-            )
-            if support is None:
-                raise ValueError("participant opacity runtime binding is not admitted by the resolver context")
-            validate_participant_opacity_runtime_enforcement(
-                resolution.opacity_enforcement,
-                support=support,
-                participant_address=intent.participant_address,
-                audience_scope_ref=intent.audience_scope_ref,
-            )
-            resolution = normalize_participant_opacity_resolution(resolution)
     except (TypeError, ValueError):
         return _prepare_policy_unresolved(
             control_plane,

implementations/python/packages/raes_runtime/participant_crossing_records.py

@@ -21,7 +21,7 @@
 )
 from raes_contracts.diagnostics import Diagnostic
 from raes_contracts.planning import RuntimeDomain
-from raes_contracts.runtime_state import OperationReceipt, OperationState, OperationStatus
+from raes_contracts.runtime_state import OperationReceipt, OperationState, OperationStatus, RuntimeSnapshot
 
 from .control_plane_security import ControlPlaneIdentity
 from .control_plane_store import AuditEvent, ControlPlaneOperationRecord
@@ -57,6 +57,37 @@ class _CrossingDecisionPreparation:
     scoped_key: str
 
 
+def _next_crossing_snapshot(
+    control_plane: object,
+    intent: ParticipantCrossingIntent,
+    records: list[ParticipantCrossingOccurrenceModel],
+) -> RuntimeSnapshot:
+    history = list(control_plane._snapshot.participant_crossing_history.get(intent.participant_address, ()))
+    candidate_history = [
+        *history,
+        *(record.model_dump(mode="json", exclude_none=True) for record in records),
+    ]
+    next_snapshot = control_plane._snapshot.with_entries(
+        dict(control_plane._snapshot.entries),
+        participant_crossing_history={
+            **control_plane._snapshot.participant_crossing_history,
+            intent.participant_address: candidate_history,
+        },
+    )
+    context = control_plane._crossing_policy_resolver.validation_context(
+        control_plane._snapshot,
+        intent.participant_address,
+    )
+    validate_participant_crossing_occurrence_context(
+        [ParticipantCrossingOccurrenceModel.model_validate(item) for item in candidate_history],
+        known_subjects=context.known_subjects,
+        policies=context.policies,
+        known_evidence_refs=context.known_evidence_refs,
+        known_authority_basis_refs=context.known_authority_basis_refs,
+    )
+    return next_snapshot
+
+
 def _prepare_crossing_decision(
     control_plane: object,
     intent: ParticipantCrossingIntent,
@@ -67,7 +98,6 @@ def _prepare_crossing_decision(
     support = preparation.support
     gates = preparation.gates
     disposition = preparation.disposition
-    history = list(control_plane._snapshot.participant_crossing_history.get(intent.participant_address, ()))
     request, decision = _crossing_records(intent, identity, resolution, support, gates, disposition)
     records = [request, decision]
     final_decision = decision
@@ -108,28 +138,7 @@ def _prepare_crossing_decision(
             records.extend((fresh_request, fresh_decision))
             final_decision = fresh_decision
             final_disposition = fresh_disposition
-    candidate_history = [
-        *history,
-        *(record.model_dump(mode="json", exclude_none=True) for record in records),
-    ]
-    next_snapshot = control_plane._snapshot.with_entries(
-        dict(control_plane._snapshot.entries),
-        participant_crossing_history={
-            **control_plane._snapshot.participant_crossing_history,
-            intent.participant_address: candidate_history,
-        },
-    )
-    context = control_plane._crossing_policy_resolver.validation_context(
-        control_plane._snapshot,
-        intent.participant_address,
-    )
-    validate_participant_crossing_occurrence_context(
-        [ParticipantCrossingOccurrenceModel.model_validate(item) for item in candidate_history],
-        known_subjects=context.known_subjects,
-        policies=context.policies,
-        known_evidence_refs=context.known_evidence_refs,
-        known_authority_basis_refs=context.known_authority_basis_refs,
-    )
+    next_snapshot = _next_crossing_snapshot(control_plane, intent, records)
     record, audit_event = _operation_artifacts(
         intent,
         identity,