fix: resolve Sonar quality gate findings

implementations/python/packages/raes/semantics/participant_behavior.py

@@ -559,66 +559,96 @@ def _autonomous_clock_binding_issues(
     return issues, clock, progression
 
 
+def _autonomous_constraint_refs(
+    context: _AutonomousExecutionReferenceContext,
+    *,
+    activity_policy: bool,
+) -> list[str]:
+    if activity_policy:
+        return [*context.policy.work_window_refs, *context.policy.pause_window_refs]
+    return list(context.policy.temporal_constraint_refs)
+
+
+def _autonomous_window_subject_issues(
+    context: _AutonomousExecutionReferenceContext,
+    constraint_ref: str,
+    constraint: object,
+) -> list[ParticipantBehaviorIssue]:
+    subjects = {str(ref) for ref in getattr(constraint, "subject_refs", ())}
+    if context.spec_name in subjects or context.participants.issubset(subjects):
+        return []
+    return [
+        _autonomous_issue(
+            context,
+            "participant.autonomous-activity-window-subject-mismatch",
+            constraint_ref,
+        )
+    ]
+
+
+def _autonomous_constraint_reference_issues(
+    context: _AutonomousExecutionReferenceContext,
+    constraint_ref: str,
+    *,
+    activity_policy: bool,
+) -> tuple[list[ParticipantBehaviorIssue], object | None]:
+    if context.is_unresolved(constraint_ref):
+        return [], None
+    constraint_name = _resolve_section_ref(
+        constraint_ref,
+        "temporal_constraints",
+        context.references.temporal_constraints,
+    )
+    constraint = context.references.temporal_constraints.get(constraint_name) if constraint_name is not None else None
+    if constraint is None:
+        return [_autonomous_issue(context, "participant.autonomous-constraint-unbound", constraint_ref)], None
+
+    issues: list[ParticipantBehaviorIssue] = []
+    kind = getattr(getattr(constraint, "constraint_kind", None), "value", "")
+    if activity_policy and kind != "window":
+        issues.append(
+            _autonomous_issue(
+                context,
+                "participant.autonomous-activity-window-kind-invalid",
+                constraint_ref,
+            )
+        )
+    if activity_policy and kind == "window":
+        issues.extend(_autonomous_window_subject_issues(context, constraint_ref, constraint))
+    if getattr(constraint, "clock_ref", None) != context.policy.clock_ref:
+        issues.append(
+            _autonomous_issue(
+                context,
+                "participant.autonomous-constraint-clock-mismatch",
+                constraint_ref,
+            )
+        )
+    return issues, constraint
+
+
 def _autonomous_constraint_issues(
     context: _AutonomousExecutionReferenceContext,
 ) -> tuple[list[ParticipantBehaviorIssue], object | None, int]:
     issues: list[ParticipantBehaviorIssue] = []
     cadence = None
     cadence_count = 0
-    profile = getattr(context.policy, "profile", "participant-autonomous-execution/v1")
-    activity_policy = profile in {
+    activity_policy = getattr(context.policy, "profile", "participant-autonomous-execution/v1") in {
         "participant-autonomous-execution/v2",
         "participant-autonomous-execution/v3",
     }
-    constraint_refs = (
-        [*context.policy.work_window_refs, *context.policy.pause_window_refs]
-        if activity_policy
-        else list(context.policy.temporal_constraint_refs)
-    )
-    for constraint_ref in constraint_refs:
-        if context.is_unresolved(constraint_ref):
-            continue
-        constraint_name = _resolve_section_ref(
+    for constraint_ref in _autonomous_constraint_refs(context, activity_policy=activity_policy):
+        reference_issues, constraint = _autonomous_constraint_reference_issues(
+            context,
             constraint_ref,
-            "temporal_constraints",
-            context.references.temporal_constraints,
-        )
-        constraint = (
-            context.references.temporal_constraints.get(constraint_name) if constraint_name is not None else None
+            activity_policy=activity_policy,
         )
+        issues.extend(reference_issues)
         if constraint is None:
-            issues.append(_autonomous_issue(context, "participant.autonomous-constraint-unbound", constraint_ref))
             continue
         kind = getattr(getattr(constraint, "constraint_kind", None), "value", "")
         cadence_count += int(kind == "cadence")
         if kind == "cadence":
             cadence = constraint
-        if activity_policy and kind != "window":
-            issues.append(
-                _autonomous_issue(
-                    context,
-                    "participant.autonomous-activity-window-kind-invalid",
-                    constraint_ref,
-                )
-            )
-        if activity_policy and kind == "window":
-            subjects = {str(ref) for ref in getattr(constraint, "subject_refs", ())}
-            if context.spec_name not in subjects and not context.participants.issubset(subjects):
-                issues.append(
-                    _autonomous_issue(
-                        context,
-                        "participant.autonomous-activity-window-subject-mismatch",
-                        constraint_ref,
-                    )
-                )
-        if getattr(constraint, "clock_ref", None) != context.policy.clock_ref:
-            issues.append(
-                _autonomous_issue(
-                    context,
-                    "participant.autonomous-constraint-clock-mismatch",
-                    constraint_ref,
-                )
-            )
     if not activity_policy and cadence_count != 1:
         issues.append(_autonomous_issue(context, "participant.autonomous-cadence-missing", context.policy.clock_ref))
     return issues, cadence, cadence_count
@@ -673,50 +703,58 @@ def _autonomous_progression_issues(
     return issues
 
 
-def _autonomous_stepped_cadence_issues(
+def _autonomous_stepped_issue_code(
     context: _AutonomousExecutionReferenceContext,
     bindings: _AutonomousTimeBindings,
-    progression_mode: str,
-) -> list[ParticipantBehaviorIssue]:
-    if progression_mode != "stepped":
-        return []
+) -> str | None:
+    issue_code = None
     step_ticks = getattr(bindings.progression, "step_ticks", None)
-    if getattr(context.policy, "profile", "participant-autonomous-execution/v1") in {
+    activity_policy = getattr(context.policy, "profile", "participant-autonomous-execution/v1") in {
         "participant-autonomous-execution/v2",
         "participant-autonomous-execution/v3",
-    }:
+    }
+    if activity_policy:
         minimum_ticks = context.policy.timing.minimum_ticks
         maximum_ticks = context.policy.timing.maximum_ticks
-        if isinstance(step_ticks, int) and not minimum_ticks % step_ticks and not maximum_ticks % step_ticks:
-            return []
-        return [
-            _autonomous_issue(
-                context,
-                "participant.autonomous-activity-timing-unreachable",
-                context.policy.progression_policy_ref,
-            )
-        ]
-    if bindings.cadence_count != 1 or bindings.cadence is None:
-        return []
-    cadence_ticks = getattr(bindings.cadence, "cadence_ticks", None)
-    start = getattr(bindings.cadence, "start", None)
-    start_tick = getattr(start, "tick", 0) if start is not None else 0
-    reachable = (
-        isinstance(step_ticks, int)
-        and isinstance(cadence_ticks, int)
-        and start_tick >= 0
-        and not start_tick % step_ticks
-        and not cadence_ticks % step_ticks
-    )
-    if reachable:
-        return []
-    return [
-        _autonomous_issue(
+        reachable = isinstance(step_ticks, int) and not minimum_ticks % step_ticks and not maximum_ticks % step_ticks
+        if not reachable:
+            issue_code = "participant.autonomous-activity-timing-unreachable"
+    elif bindings.cadence_count == 1 and bindings.cadence is not None:
+        cadence_ticks = getattr(bindings.cadence, "cadence_ticks", None)
+        start = getattr(bindings.cadence, "start", None)
+        start_tick = getattr(start, "tick", 0) if start is not None else 0
+        reachable = (
+            isinstance(step_ticks, int)
+            and isinstance(cadence_ticks, int)
+            and start_tick >= 0
+            and not start_tick % step_ticks
+            and not cadence_ticks % step_ticks
+        )
+        if not reachable:
+            issue_code = "participant.autonomous-cadence-unreachable"
+    return issue_code
+
+
+def _autonomous_stepped_cadence_issues(
+    context: _AutonomousExecutionReferenceContext,
+    bindings: _AutonomousTimeBindings,
+    progression_mode: str,
+) -> list[ParticipantBehaviorIssue]:
+    issues: list[ParticipantBehaviorIssue] = []
+    if progression_mode == "stepped":
+        issue_code = _autonomous_stepped_issue_code(
             context,
-            "participant.autonomous-cadence-unreachable",
-            context.policy.progression_policy_ref,
+            bindings,
         )
-    ]
+        if issue_code is not None:
+            issues.append(
+                _autonomous_issue(
+                    context,
+                    issue_code,
+                    context.policy.progression_policy_ref,
+                )
+            )
+    return issues
 
 
 def _autonomous_non_evaluated_issues(

implementations/python/packages/raes_backend_protocols/capability_admission.py

@@ -12,9 +12,9 @@
     PARTICIPANT_RUNTIME_INTERACTION_FEATURE_SCOPE,
     PARTICIPANT_RUNTIME_ROLE_SCOPE,
 )
-from .participant_feature_admission import participant_feature_support_gaps as participant_feature_support_gaps
 from .participant_feature_admission import (
-    resolve_participant_feature_support as resolve_participant_feature_support,
+    participant_feature_support_gaps,
+    resolve_participant_feature_support,
 )
 from .participant_resource_admission import (
     ResourceGovernedPolicy,
@@ -430,3 +430,9 @@ def require_cleanup_plan_capability(manifest: BackendManifest, plan: TrialCleanu
     required_cleanup = any(obligation.requirement == "required" for obligation in plan.cleanup_obligations.values())
     if required_cleanup and not cleanup.supports_residual_state_disclosure:
         raise ValueError("required cleanup needs backend residual-state disclosure")
+
+
+__all__ = [
+    "participant_feature_support_gaps",
+    "resolve_participant_feature_support",
+]

implementations/python/packages/raes_backend_protocols/participant_capabilities.py

@@ -41,6 +41,46 @@ def _validate_participant_feature_support_term(feature: str) -> None:
     )
 
 
+def _participant_feature_support_level(
+    value: ParticipantFeatureSupportLevel | str,
+) -> ParticipantFeatureSupportLevel:
+    try:
+        if isinstance(value, ParticipantFeatureSupportLevel):
+            return value
+        return ParticipantFeatureSupportLevel(str(value))
+    except ValueError as exc:
+        raise ValueError("ParticipantFeatureSupport.support_level must be a valid support level") from exc
+
+
+def _participant_feature_refs(field_name: str, values: tuple[str, ...]) -> tuple[str, ...]:
+    normalized = tuple(values)
+    _validate_unique_non_empty_strings(f"ParticipantFeatureSupport.{field_name}", normalized)
+    return normalized
+
+
+def _validate_participant_feature_evidence(
+    *,
+    feature: str,
+    support_level: ParticipantFeatureSupportLevel,
+    constraint_refs: tuple[str, ...],
+    limitation_refs: tuple[str, ...],
+    disclosure_refs: tuple[str, ...],
+    evidence_refs: tuple[str, ...],
+) -> None:
+    if support_level != ParticipantFeatureSupportLevel.EXACT and not disclosure_refs:
+        raise ValueError(
+            "ParticipantFeatureSupport disclosure_refs must be non-empty when support_level is below exact"
+        )
+    if feature not in PARTICIPANT_RUNTIME_POLICY_FEATURES:
+        return
+    if support_level != ParticipantFeatureSupportLevel.EXACT and not limitation_refs:
+        raise ValueError("ParticipantFeatureSupport limitation_refs must be non-empty for below-exact policy support")
+    if support_level == ParticipantFeatureSupportLevel.BOUNDED and not constraint_refs:
+        raise ValueError("ParticipantFeatureSupport constraint_refs must be non-empty for bounded policy support")
+    if support_level != ParticipantFeatureSupportLevel.UNSUPPORTED and not evidence_refs:
+        raise ValueError("ParticipantFeatureSupport evidence_refs must be non-empty for positive policy support")
+
+
 @dataclass(frozen=True)
 class ParticipantFeatureSupport:
     """API-407 per-feature participant runtime support declaration."""
@@ -56,39 +96,19 @@ def __post_init__(self) -> None:
         if not self.feature.strip():
             raise ValueError("ParticipantFeatureSupport.feature must be non-empty")
         _validate_participant_feature_support_term(self.feature)
-        try:
-            support_level = (
-                self.support_level
-                if isinstance(self.support_level, ParticipantFeatureSupportLevel)
-                else ParticipantFeatureSupportLevel(str(self.support_level))
-            )
-        except ValueError as exc:
-            raise ValueError("ParticipantFeatureSupport.support_level must be a valid support level") from exc
-        constraint_refs = tuple(self.constraint_refs)
-        limitation_refs = tuple(self.limitation_refs)
-        disclosure_refs = tuple(self.disclosure_refs)
-        evidence_refs = tuple(self.evidence_refs)
-        _validate_unique_non_empty_strings("ParticipantFeatureSupport.constraint_refs", constraint_refs)
-        _validate_unique_non_empty_strings("ParticipantFeatureSupport.limitation_refs", limitation_refs)
-        _validate_unique_non_empty_strings("ParticipantFeatureSupport.disclosure_refs", disclosure_refs)
-        _validate_unique_non_empty_strings("ParticipantFeatureSupport.evidence_refs", evidence_refs)
-        if support_level != ParticipantFeatureSupportLevel.EXACT and not disclosure_refs:
-            raise ValueError(
-                "ParticipantFeatureSupport disclosure_refs must be non-empty when support_level is below exact"
-            )
-        if self.feature in PARTICIPANT_RUNTIME_POLICY_FEATURES:
-            if support_level != ParticipantFeatureSupportLevel.EXACT and not limitation_refs:
-                raise ValueError(
-                    "ParticipantFeatureSupport limitation_refs must be non-empty for below-exact policy support"
-                )
-            if support_level == ParticipantFeatureSupportLevel.BOUNDED and not constraint_refs:
-                raise ValueError(
-                    "ParticipantFeatureSupport constraint_refs must be non-empty for bounded policy support"
-                )
-            if support_level != ParticipantFeatureSupportLevel.UNSUPPORTED and not evidence_refs:
-                raise ValueError(
-                    "ParticipantFeatureSupport evidence_refs must be non-empty for positive policy support"
-                )
+        support_level = _participant_feature_support_level(self.support_level)
+        constraint_refs = _participant_feature_refs("constraint_refs", self.constraint_refs)
+        limitation_refs = _participant_feature_refs("limitation_refs", self.limitation_refs)
+        disclosure_refs = _participant_feature_refs("disclosure_refs", self.disclosure_refs)
+        evidence_refs = _participant_feature_refs("evidence_refs", self.evidence_refs)
+        _validate_participant_feature_evidence(
+            feature=self.feature,
+            support_level=support_level,
+            constraint_refs=constraint_refs,
+            limitation_refs=limitation_refs,
+            disclosure_refs=disclosure_refs,
+            evidence_refs=evidence_refs,
+        )
         object.__setattr__(self, "support_level", support_level)
         object.__setattr__(self, "constraint_refs", constraint_refs)
         object.__setattr__(self, "limitation_refs", limitation_refs)
@@ -355,22 +375,25 @@ def _autonomous_limits(self) -> tuple[tuple[str, int | None], ...]:
         )
 
     def _has_autonomous_configuration(self) -> bool:
-        return bool(
-            self.supported_autonomous_selection_strategies
-            or self.supported_autonomous_action_contracts
-            or self.supported_autonomous_observation_boundaries
-            or self.supported_autonomous_target_addresses
-            or self.supported_autonomous_policy_profiles
-            or self.supported_autonomous_activity_features
-            or self.supported_autonomous_random_stream_profiles
-            or self.execution_bindings
-            or self.supports_execution_control
-            or self.supported_execution_control_actions
-            or self.supports_bounded_concurrency
-            or self.max_execution_services is not None
-            or self.max_concurrent_actions is not None
-            or self.resource_budgets is not None
-            or any(value is not None for _, value in self._autonomous_limits())
+        limits_configured = any(value is not None for _, value in self._autonomous_limits())
+        return any(
+            (
+                self.supported_autonomous_selection_strategies,
+                self.supported_autonomous_action_contracts,
+                self.supported_autonomous_observation_boundaries,
+                self.supported_autonomous_target_addresses,
+                self.supported_autonomous_policy_profiles,
+                self.supported_autonomous_activity_features,
+                self.supported_autonomous_random_stream_profiles,
+                self.execution_bindings,
+                self.supports_execution_control,
+                self.supported_execution_control_actions,
+                self.supports_bounded_concurrency,
+                self.max_execution_services is not None,
+                self.max_concurrent_actions is not None,
+                self.resource_budgets is not None,
+                limits_configured,
+            )
         )