From 298937bd2817cfa402c4de269ea11f5e976f9baf Mon Sep 17 00:00:00 2001 From: Brad Edwards Date: Mon, 27 Jul 2026 08:04:21 +0200 Subject: [PATCH 1/5] fix: resolve Sonar quality gate findings --- .../raes/semantics/participant_behavior.py | 194 +++++---- .../capability_admission.py | 10 +- .../participant_capabilities.py | 121 +++--- .../raes_contracts/contracts/bundle.py | 11 +- .../raes_contracts/contracts/manifests.py | 69 +-- .../contracts/participant_runtime.py | 8 +- .../packages/raes_contracts/runtime_state.py | 226 +++++----- .../participant_autonomous_execution.py | 202 +++++---- .../raes_reference_backend/manifest.py | 112 ++--- .../raes_runtime/participant_activity.py | 102 +++-- .../participant_activity_support.py | 8 +- .../raes_runtime/participant_scheduler.py | 220 ++++++---- .../participant_scheduler_concurrency.py | 16 +- .../participant_scheduler_operations.py | 401 ++++++++++++------ .../participant_scheduler_reset.py | 19 +- ...st_dsl_437_benign_participant_execution.py | 12 +- .../tests/test_random_stream_profile.py | 5 +- 17 files changed, 1070 insertions(+), 666 deletions(-) diff --git a/implementations/python/packages/raes/semantics/participant_behavior.py b/implementations/python/packages/raes/semantics/participant_behavior.py index 76ae04795..7a7d19531 100644 --- a/implementations/python/packages/raes/semantics/participant_behavior.py +++ b/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( diff --git a/implementations/python/packages/raes_backend_protocols/capability_admission.py b/implementations/python/packages/raes_backend_protocols/capability_admission.py index ce1f99b61..a482ba5af 100644 --- a/implementations/python/packages/raes_backend_protocols/capability_admission.py +++ b/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", +] diff --git a/implementations/python/packages/raes_backend_protocols/participant_capabilities.py b/implementations/python/packages/raes_backend_protocols/participant_capabilities.py index 79aa177c8..253b62631 100644 --- a/implementations/python/packages/raes_backend_protocols/participant_capabilities.py +++ b/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, + ) ) diff --git a/implementations/python/packages/raes_contracts/contracts/bundle.py b/implementations/python/packages/raes_contracts/contracts/bundle.py index 2e6d1a2fa..41912f71d 100644 --- a/implementations/python/packages/raes_contracts/contracts/bundle.py +++ b/implementations/python/packages/raes_contracts/contracts/bundle.py @@ -115,7 +115,7 @@ ) -def _raw_schema_bundle() -> dict[str, dict[str, Any]]: +def _core_schema_bundle() -> dict[str, dict[str, Any]]: from raes_contracts.realization_envelope import BackendRealizationEnvelopeModel from ..behavioral_relations import BehavioralRelationCatalogModel @@ -185,6 +185,11 @@ def _raw_schema_bundle() -> dict[str, dict[str, Any]]: "scientific-completeness-assessment-v1": ScientificCompletenessAssessmentModel.model_json_schema(), "validation-profile-catalog-v1": ValidationProfileCatalogModel.model_json_schema(), "validation-basis-disclosure-v1": ValidationBasisDisclosureDocumentModel.model_json_schema(), + } + + +def _runtime_schema_bundle() -> dict[str, dict[str, Any]]: + return { "evaluation-history-event-stream-v1": _event_stream_schema( "EvaluationHistoryEventStream", EvaluationHistoryEventModel.model_json_schema(), @@ -227,6 +232,10 @@ def _raw_schema_bundle() -> dict[str, dict[str, Any]]: } +def _raw_schema_bundle() -> dict[str, dict[str, Any]]: + return {**_core_schema_bundle(), **_runtime_schema_bundle()} + + @cache def _schema_bundle_template() -> dict[str, dict[str, Any]]: """Build the immutable-in-practice template used by :func:`schema_bundle`.""" diff --git a/implementations/python/packages/raes_contracts/contracts/manifests.py b/implementations/python/packages/raes_contracts/contracts/manifests.py index a5f53a886..b44e99d4d 100644 --- a/implementations/python/packages/raes_contracts/contracts/manifests.py +++ b/implementations/python/packages/raes_contracts/contracts/manifests.py @@ -217,32 +217,36 @@ def _validate_autonomous_configuration(self) -> None: declares_autonomous = "autonomous_execution" in self.supported_behavior_features if declares_autonomous != self.supports_autonomous_execution: raise ValueError("autonomous_execution feature and support flag must agree") - if self.supports_autonomous_execution and not self._has_complete_autonomous_configuration(): + if self.supports_autonomous_execution: + self._validate_enabled_autonomous_configuration() + elif self._has_any_autonomous_configuration(): + raise ValueError("autonomous execution limits require autonomous execution support") + + def _validate_enabled_autonomous_configuration(self) -> None: + if not self._has_complete_autonomous_configuration(): raise ValueError( "autonomous execution requires selection strategies, exact action, observation, and policy-profile " "support, and finite limits" ) - if ( - self.supports_autonomous_execution - and { - "participant-autonomous-execution/v2", - "participant-autonomous-execution/v3", - }.intersection(self.supported_autonomous_policy_profiles) - and ( - not self.supported_autonomous_activity_features or not self.supported_autonomous_random_stream_profiles - ) - ): + self._validate_activity_profile_configuration() + + def _validate_activity_profile_configuration(self) -> None: + activity_profiles = { + "participant-autonomous-execution/v2", + "participant-autonomous-execution/v3", + }.intersection(self.supported_autonomous_policy_profiles) + missing_activity_support = ( + not self.supported_autonomous_activity_features or not self.supported_autonomous_random_stream_profiles + ) + if activity_profiles and missing_activity_support: raise ValueError( "autonomous execution v2 requires exact activity-feature and random-stream-profile support" ) if ( - self.supports_autonomous_execution - and "participant-autonomous-execution/v3" in self.supported_autonomous_policy_profiles + "participant-autonomous-execution/v3" in self.supported_autonomous_policy_profiles and self.resource_budgets is None ): raise ValueError("autonomous execution v3 requires participant resource-budget capabilities") - if not self.supports_autonomous_execution and self._has_any_autonomous_configuration(): - raise ValueError("autonomous execution limits require autonomous execution support") def _has_complete_autonomous_configuration(self) -> bool: return bool( @@ -259,22 +263,25 @@ def _has_complete_autonomous_configuration(self) -> bool: ) def _has_any_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 any(value is not None for value in self._autonomous_limits()) - 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 + 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, + limits_configured, + 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, + ) ) def _validate_execution_control(self) -> None: diff --git a/implementations/python/packages/raes_contracts/contracts/participant_runtime.py b/implementations/python/packages/raes_contracts/contracts/participant_runtime.py index 5fdf334bc..244c0f9c8 100644 --- a/implementations/python/packages/raes_contracts/contracts/participant_runtime.py +++ b/implementations/python/packages/raes_contracts/contracts/participant_runtime.py @@ -39,6 +39,8 @@ from .participant_resource_budgets import ParticipantResourceMeasurementModel from .random_stream import ParticipantStreamAddressModel +_AUTONOMOUS_EXECUTION_V1 = "participant-autonomous-execution/v1" + class ParticipantEpisodeStateModel(ContractModel): state_schema_version: Literal[PARTICIPANT_EPISODE_STATE_SCHEMA_VERSION] = PARTICIPANT_EPISODE_STATE_SCHEMA_VERSION @@ -287,7 +289,7 @@ class ParticipantAutonomousExecutionStateModel(ContractModel): "participant-autonomous-execution/v1", "participant-autonomous-execution/v2", "participant-autonomous-execution/v3", - ] = "participant-autonomous-execution/v1" + ] = _AUTONOMOUS_EXECUTION_V1 occurrence_ordinal: StrictInt = Field(default=0, ge=0) current_retry: StrictInt = Field(default=0, ge=0) burst_position: StrictInt = Field(default=0, ge=0) @@ -306,7 +308,7 @@ def _serialize_profile_state( handler: SerializerFunctionWrapHandler, ) -> dict[str, Any]: payload = handler(self) - if self.profile == "participant-autonomous-execution/v1": + if self.profile == _AUTONOMOUS_EXECUTION_V1: for field_name in ( "profile", "occurrence_ordinal", @@ -328,7 +330,7 @@ def _serialize_profile_state( def _validate_counters(self) -> ParticipantAutonomousExecutionStateModel: if self.succeeded_actions + self.failed_actions > self.attempted_actions: raise ValueError("terminal autonomous action counts cannot exceed attempted actions") - if self.profile == "participant-autonomous-execution/v1": + if self.profile == _AUTONOMOUS_EXECUTION_V1: if any((self.random_control_id, self.random_profile_id, self.random_namespace)): raise ValueError("v1 autonomous execution state cannot carry participant random-control identity") elif not all((self.random_control_id, self.random_profile_id, self.random_namespace)): diff --git a/implementations/python/packages/raes_contracts/runtime_state.py b/implementations/python/packages/raes_contracts/runtime_state.py index 93d635c02..bb67c3855 100644 --- a/implementations/python/packages/raes_contracts/runtime_state.py +++ b/implementations/python/packages/raes_contracts/runtime_state.py @@ -125,107 +125,131 @@ def with_entries( **updates: object, ) -> RuntimeSnapshot: _validate_snapshot_update_keys(updates) - return RuntimeSnapshot( - entries=entries, - orchestration_results=_mapping_update( - updates, - "orchestration_results", - self.orchestration_results, - ), - orchestration_history=_history_update( - updates, - "orchestration_history", - self.orchestration_history, - ), - evaluation_results=_mapping_update(updates, "evaluation_results", self.evaluation_results), - evaluation_history=_history_update(updates, "evaluation_history", self.evaluation_history), - proposition_truth_results=_mapping_update( - updates, - "proposition_truth_results", - self.proposition_truth_results, - ), - participant_episode_results=_mapping_update( - updates, - "participant_episode_results", - self.participant_episode_results, - ), - participant_episode_history=_history_update( - updates, - "participant_episode_history", - self.participant_episode_history, - ), - participant_behavior_history=_history_update( - updates, - "participant_behavior_history", - self.participant_behavior_history, - ), - participant_control_history=_history_update( - updates, - "participant_control_history", - self.participant_control_history, - ), - participant_autonomous_execution_states=_mapping_update( - updates, - "participant_autonomous_execution_states", - self.participant_autonomous_execution_states, - ), - participant_execution_services=_mapping_update( - updates, - "participant_execution_services", - self.participant_execution_services, - ), - participant_resource_budget_states=_mapping_update( - updates, - "participant_resource_budget_states", - self.participant_resource_budget_states, - ), - participant_resource_pool_states=_mapping_update( - updates, - "participant_resource_pool_states", - self.participant_resource_pool_states, - ), - participant_resource_budget_events=_mapping_update( - updates, - "participant_resource_budget_events", - self.participant_resource_budget_events, - ), - shared_state_records=_mapping_update( - updates, - "shared_state_records", - self.shared_state_records, - ), - shared_state_history=_history_update( - updates, - "shared_state_history", - self.shared_state_history, - ), - joint_action_records=_mapping_update( - updates, - "joint_action_records", - self.joint_action_records, - ), - time_management_contexts=_mapping_update( - updates, - "time_management_contexts", - self.time_management_contexts, - ), - time_model_state=_time_model_state_update( - updates, - "time_model_state", - self.time_model_state, - ), - realization_provenance=_provenance_update( - updates, - "realization_provenance", - self.realization_provenance, - ), - realization_envelope=_identity_update( - updates, - "realization_envelope", - self.realization_envelope, - ), - metadata=_mapping_update(updates, "metadata", self.metadata), - ) + return RuntimeSnapshot(entries=entries, **_snapshot_updates(self, updates)) + + +def _snapshot_result_updates( + snapshot: RuntimeSnapshot, + updates: Mapping[str, object], +) -> dict[str, Any]: + return { + "orchestration_results": _mapping_update( + updates, + "orchestration_results", + snapshot.orchestration_results, + ), + "orchestration_history": _history_update( + updates, + "orchestration_history", + snapshot.orchestration_history, + ), + "evaluation_results": _mapping_update(updates, "evaluation_results", snapshot.evaluation_results), + "evaluation_history": _history_update(updates, "evaluation_history", snapshot.evaluation_history), + "proposition_truth_results": _mapping_update( + updates, + "proposition_truth_results", + snapshot.proposition_truth_results, + ), + "participant_episode_results": _mapping_update( + updates, + "participant_episode_results", + snapshot.participant_episode_results, + ), + "participant_episode_history": _history_update( + updates, + "participant_episode_history", + snapshot.participant_episode_history, + ), + "participant_behavior_history": _history_update( + updates, + "participant_behavior_history", + snapshot.participant_behavior_history, + ), + "participant_control_history": _history_update( + updates, + "participant_control_history", + snapshot.participant_control_history, + ), + } + + +def _snapshot_participant_updates( + snapshot: RuntimeSnapshot, + updates: Mapping[str, object], +) -> dict[str, Any]: + return { + "participant_autonomous_execution_states": _mapping_update( + updates, + "participant_autonomous_execution_states", + snapshot.participant_autonomous_execution_states, + ), + "participant_execution_services": _mapping_update( + updates, + "participant_execution_services", + snapshot.participant_execution_services, + ), + "participant_resource_budget_states": _mapping_update( + updates, + "participant_resource_budget_states", + snapshot.participant_resource_budget_states, + ), + "participant_resource_pool_states": _mapping_update( + updates, + "participant_resource_pool_states", + snapshot.participant_resource_pool_states, + ), + "participant_resource_budget_events": _mapping_update( + updates, + "participant_resource_budget_events", + snapshot.participant_resource_budget_events, + ), + "shared_state_records": _mapping_update( + updates, + "shared_state_records", + snapshot.shared_state_records, + ), + "shared_state_history": _history_update( + updates, + "shared_state_history", + snapshot.shared_state_history, + ), + "joint_action_records": _mapping_update( + updates, + "joint_action_records", + snapshot.joint_action_records, + ), + "time_management_contexts": _mapping_update( + updates, + "time_management_contexts", + snapshot.time_management_contexts, + ), + } + + +def _snapshot_updates( + snapshot: RuntimeSnapshot, + updates: Mapping[str, object], +) -> dict[str, Any]: + return { + **_snapshot_result_updates(snapshot, updates), + **_snapshot_participant_updates(snapshot, updates), + "time_model_state": _time_model_state_update( + updates, + "time_model_state", + snapshot.time_model_state, + ), + "realization_provenance": _provenance_update( + updates, + "realization_provenance", + snapshot.realization_provenance, + ), + "realization_envelope": _identity_update( + updates, + "realization_envelope", + snapshot.realization_envelope, + ), + "metadata": _mapping_update(updates, "metadata", snapshot.metadata), + } _SNAPSHOT_UPDATE_KEYS = { diff --git a/implementations/python/packages/raes_processor/compiler/participant_autonomous_execution.py b/implementations/python/packages/raes_processor/compiler/participant_autonomous_execution.py index 4a498ec6c..11a7991f7 100644 --- a/implementations/python/packages/raes_processor/compiler/participant_autonomous_execution.py +++ b/implementations/python/packages/raes_processor/compiler/participant_autonomous_execution.py @@ -1,5 +1,7 @@ """Compilation of autonomous participant execution policies.""" +from typing import Any + from raes.scenario import InstantiatedScenario from ..models import ( @@ -159,35 +161,13 @@ def _compiled_resource_budget( ) -def _compile_autonomous_execution( - *, +def _compiled_execution_bindings( scenario: InstantiatedScenario, - spec_name: str, - participant_addresses: tuple[str, ...], - behavior_spec: object, -) -> ParticipantAutonomousExecutionRuntime | None: - policy = behavior_spec.autonomous_execution - if policy is None: - return None - address = _address("participant", "autonomous-execution", spec_name) - authority = policy.evaluation_authority - profile = getattr(policy, "profile", "participant-autonomous-execution/v1") - activity_candidates = getattr(policy, "action_candidates", None) - ordered_candidates = sorted(activity_candidates.items()) if activity_candidates is not None else [] - action_refs = ( - [candidate.action_ref for _, candidate in ordered_candidates] - if ordered_candidates - else list(policy.action_order) - ) - work_window_refs = list(getattr(policy, "work_window_refs", ())) - pause_window_refs = list(getattr(policy, "pause_window_refs", ())) - temporal_constraint_refs = ( - [*work_window_refs, *pause_window_refs] - if profile in {"participant-autonomous-execution/v2", "participant-autonomous-execution/v3"} - else list(policy.temporal_constraint_refs) - ) + policy: object, + action_refs: list[str], +) -> tuple[tuple[ParticipantExecutionBindingRuntime, ...], tuple[str, ...]]: addressable_ref_index = _runtime_addressable_ref_index(scenario) - execution_bindings_by_key: dict[tuple[str, tuple[str, ...]], ParticipantExecutionBindingRuntime] = {} + bindings_by_key: dict[tuple[str, tuple[str, ...]], ParticipantExecutionBindingRuntime] = {} for action_ref in action_refs: action_name = _section_ref_name( action_ref, @@ -204,7 +184,7 @@ def _compile_autonomous_execution( list(dict.fromkeys(target_refs)), addressable_ref_index=addressable_ref_index, ) - execution_bindings_by_key.setdefault( + bindings_by_key.setdefault( (action_contract_address, target_addresses), ParticipantExecutionBindingRuntime( action_contract_address=action_contract_address, @@ -214,10 +194,104 @@ def _compile_autonomous_execution( max_in_flight=policy.max_in_flight, ), ) - execution_bindings = tuple(execution_bindings_by_key.values()) - target_addresses = tuple( - dict.fromkeys(target for binding in execution_bindings for target in binding.target_addresses) + bindings = tuple(bindings_by_key.values()) + targets = tuple(dict.fromkeys(target for binding in bindings for target in binding.target_addresses)) + return bindings, targets + + +def _temporal_constraint_addresses( + scenario: InstantiatedScenario, + refs: list[str], +) -> tuple[str, ...]: + return tuple( + _address( + "time", + "constraint", + _section_ref_name(ref, "temporal_constraints", scenario.temporal_constraints), + ) + for ref in refs ) + + +def _activity_runtime_fields( + scenario: InstantiatedScenario, + policy: object, + *, + profile: str, + work_window_refs: list[str], + pause_window_refs: list[str], + ordered_candidates: list[tuple[str, Any]], +) -> dict[str, object]: + return { + "profile": profile, + "work_window_addresses": _temporal_constraint_addresses(scenario, work_window_refs), + "pause_window_addresses": _temporal_constraint_addresses(scenario, pause_window_refs), + "stochastic_control_ref": str(getattr(policy, "stochastic_control_ref", "")), + "timing_minimum_ticks": int(getattr(getattr(policy, "timing", None), "minimum_ticks", 0)), + "timing_maximum_ticks": int(getattr(getattr(policy, "timing", None), "maximum_ticks", 0)), + "outside_window_disposition": str(getattr(policy, "outside_window_disposition", "")), + "empty_eligible_disposition": str(getattr(policy, "empty_eligible_disposition", "")), + "action_candidate_ids": tuple(str(candidate_id) for candidate_id, _ in ordered_candidates), + "action_candidate_weights": tuple(candidate.weight for _, candidate in ordered_candidates), + "action_candidate_dependencies": tuple( + tuple(str(ref) for ref in candidate.depends_on) for _, candidate in ordered_candidates + ), + "action_candidate_retry_failure_classes": tuple( + tuple(value.value for value in candidate.retryable_failure_classes) for _, candidate in ordered_candidates + ), + "action_candidate_max_retries": tuple(candidate.max_retries for _, candidate in ordered_candidates), + "action_candidate_cooldown_ticks": tuple(candidate.cooldown_ticks for _, candidate in ordered_candidates), + "max_occurrences": int(getattr(policy, "max_occurrences", 0)), + "max_burst_size": int(getattr(policy, "max_burst_size", 1)), + } + + +def _runtime_refresh_dependencies( + scenario: InstantiatedScenario, + participant_addresses: tuple[str, ...], + action_refs: list[str], + objective_refs: tuple[str, ...], + target_addresses: tuple[str, ...], +) -> tuple[str, ...]: + return ( + *participant_addresses, + *tuple( + _action_contract_address(_section_ref_name(ref, "action_contracts", scenario.action_contracts)) + for ref in action_refs + ), + *tuple(_objective_address(_section_ref_name(ref, "objectives", scenario.objectives)) for ref in objective_refs), + *target_addresses, + ) + + +def _compile_autonomous_execution( + *, + scenario: InstantiatedScenario, + spec_name: str, + participant_addresses: tuple[str, ...], + behavior_spec: object, +) -> ParticipantAutonomousExecutionRuntime | None: + policy = behavior_spec.autonomous_execution + if policy is None: + return None + address = _address("participant", "autonomous-execution", spec_name) + authority = policy.evaluation_authority + profile = getattr(policy, "profile", "participant-autonomous-execution/v1") + activity_candidates = getattr(policy, "action_candidates", None) + ordered_candidates = sorted(activity_candidates.items()) if activity_candidates is not None else [] + action_refs = ( + [candidate.action_ref for _, candidate in ordered_candidates] + if ordered_candidates + else list(policy.action_order) + ) + work_window_refs = list(getattr(policy, "work_window_refs", ())) + pause_window_refs = list(getattr(policy, "pause_window_refs", ())) + temporal_constraint_refs = ( + [*work_window_refs, *pause_window_refs] + if profile in {"participant-autonomous-execution/v2", "participant-autonomous-execution/v3"} + else list(policy.temporal_constraint_refs) + ) + execution_bindings, target_addresses = _compiled_execution_bindings(scenario, policy, action_refs) resource_owners, resource_demands, resource_fairness = _compiled_resource_budget( scenario, policy, @@ -239,14 +313,7 @@ def _compile_autonomous_execution( scenario.time_progression_policies, ), ), - temporal_constraint_addresses=tuple( - _address( - "time", - "constraint", - _section_ref_name(ref, "temporal_constraints", scenario.temporal_constraints), - ) - for ref in temporal_constraint_refs - ), + temporal_constraint_addresses=_temporal_constraint_addresses(scenario, temporal_constraint_refs), action_contract_addresses=tuple( _action_contract_address(_section_ref_name(ref, "action_contracts", scenario.action_contracts)) for ref in action_refs @@ -272,54 +339,23 @@ def _compile_autonomous_execution( proof_producer_refs=tuple(authority.proof_producer_refs), score_authority_refs=tuple(authority.score_authority_refs), receipt_authority_refs=tuple(authority.receipt_authority_refs), - profile=profile, - work_window_addresses=tuple( - _address( - "time", - "constraint", - _section_ref_name(ref, "temporal_constraints", scenario.temporal_constraints), - ) - for ref in work_window_refs + **_activity_runtime_fields( + scenario, + policy, + profile=profile, + work_window_refs=work_window_refs, + pause_window_refs=pause_window_refs, + ordered_candidates=ordered_candidates, ), - pause_window_addresses=tuple( - _address( - "time", - "constraint", - _section_ref_name(ref, "temporal_constraints", scenario.temporal_constraints), - ) - for ref in pause_window_refs - ), - stochastic_control_ref=str(getattr(policy, "stochastic_control_ref", "")), - timing_minimum_ticks=int(getattr(getattr(policy, "timing", None), "minimum_ticks", 0)), - timing_maximum_ticks=int(getattr(getattr(policy, "timing", None), "maximum_ticks", 0)), - outside_window_disposition=str(getattr(policy, "outside_window_disposition", "")), - empty_eligible_disposition=str(getattr(policy, "empty_eligible_disposition", "")), - action_candidate_ids=tuple(str(candidate_id) for candidate_id, _ in ordered_candidates), - action_candidate_weights=tuple(candidate.weight for _, candidate in ordered_candidates), - action_candidate_dependencies=tuple( - tuple(str(ref) for ref in candidate.depends_on) for _, candidate in ordered_candidates - ), - action_candidate_retry_failure_classes=tuple( - tuple(value.value for value in candidate.retryable_failure_classes) for _, candidate in ordered_candidates - ), - action_candidate_max_retries=tuple(candidate.max_retries for _, candidate in ordered_candidates), - action_candidate_cooldown_ticks=tuple(candidate.cooldown_ticks for _, candidate in ordered_candidates), - max_occurrences=int(getattr(policy, "max_occurrences", 0)), - max_burst_size=int(getattr(policy, "max_burst_size", 1)), resource_owners=resource_owners, resource_demands=resource_demands, resource_fairness=resource_fairness, - refresh_dependencies=( - *participant_addresses, - *tuple( - _action_contract_address(_section_ref_name(ref, "action_contracts", scenario.action_contracts)) - for ref in action_refs - ), - *tuple( - _objective_address(_section_ref_name(ref, "objectives", scenario.objectives)) - for ref in authority.objective_refs - ), - *target_addresses, + refresh_dependencies=_runtime_refresh_dependencies( + scenario, + participant_addresses, + action_refs, + tuple(authority.objective_refs), + target_addresses, ), spec=_dump(policy), ) diff --git a/implementations/python/packages/raes_reference_backend/manifest.py b/implementations/python/packages/raes_reference_backend/manifest.py index 635d818d6..9c83a3617 100644 --- a/implementations/python/packages/raes_reference_backend/manifest.py +++ b/implementations/python/packages/raes_reference_backend/manifest.py @@ -164,6 +164,65 @@ def _time_capabilities(*, enabled: bool) -> TimeCapabilities | None: ) +def _participant_runtime_capabilities() -> ParticipantRuntimeCapabilities: + return ParticipantRuntimeCapabilities( + name="reference-emulation-participant-runtime", + supported_participant_roles=_PARTICIPANT_ROLES, + supported_behavior_features=_PARTICIPANT_BEHAVIOR_FEATURES, + supported_interaction_features=_PARTICIPANT_INTERACTION_FEATURES, + feature_support=tuple( + ParticipantFeatureSupport( + feature=feature, + support_level=ParticipantFeatureSupportLevel.UNSUPPORTED, + limitation_refs=(f"limitation:{feature}:not-realized",), + disclosure_refs=(f"disclosure:{feature}:unsupported",), + ) + for feature in sorted(PARTICIPANT_RUNTIME_POLICY_FEATURES) + ), + ) + + +def _observation_capabilities() -> ObservationCapabilities: + return ObservationCapabilities( + name="reference-emulation-observation", + supported_capture_kinds=frozenset({"artifact", "log", "observation", "telemetry", "trace"}), + supported_channel_kinds=frozenset( + { + "backend-log", + "evaluation-history", + "file-artifact", + "participant-observation", + "runtime-snapshot", + "workflow-history", + } + ), + supported_evidence_contracts=frozenset( + { + "experiment-capture-spec-v1", + "experiment-evidence-record-v1", + "experiment-derived-measure-v1", + "experiment-run-v1", + } + ), + supported_media_types=frozenset({"application/json", "text/plain"}), + supported_sealing_modes=frozenset({"digest", "immutable-store"}), + supports_redaction=True, + supports_loss_disclosure=True, + supports_chain_of_custody=False, + ) + + +def _cleanup_capabilities() -> CleanupCapabilities: + return CleanupCapabilities( + name="reference-emulation-cleanup", + supported_contract_versions=CLEANUP_CAPABILITY_REQUIRED_CONTRACTS, + supported_action_kinds=frozenset({"destroy", "reset", "restore", "compensate", "verify"}), + supported_verification_methods=frozenset({"probe", "receipt"}), + supports_reusable_state=True, + supports_residual_state_disclosure=True, + ) + + def _capabilities(*, with_time: bool) -> BackendCapabilitySet: return BackendCapabilitySet( provisioner=ProvisionerCapabilities( @@ -215,56 +274,9 @@ def _capabilities(*, with_time: bool) -> BackendCapabilitySet: supported_time_domains=frozenset({"scenario_time"}), preserves_binding_provenance=True, ), - participant_runtime=ParticipantRuntimeCapabilities( - name="reference-emulation-participant-runtime", - supported_participant_roles=_PARTICIPANT_ROLES, - supported_behavior_features=_PARTICIPANT_BEHAVIOR_FEATURES, - supported_interaction_features=_PARTICIPANT_INTERACTION_FEATURES, - feature_support=tuple( - ParticipantFeatureSupport( - feature=feature, - support_level=ParticipantFeatureSupportLevel.UNSUPPORTED, - limitation_refs=(f"limitation:{feature}:not-realized",), - disclosure_refs=(f"disclosure:{feature}:unsupported",), - ) - for feature in sorted(PARTICIPANT_RUNTIME_POLICY_FEATURES) - ), - ), - observation=ObservationCapabilities( - name="reference-emulation-observation", - supported_capture_kinds=frozenset({"artifact", "log", "observation", "telemetry", "trace"}), - supported_channel_kinds=frozenset( - { - "backend-log", - "evaluation-history", - "file-artifact", - "participant-observation", - "runtime-snapshot", - "workflow-history", - } - ), - supported_evidence_contracts=frozenset( - { - "experiment-capture-spec-v1", - "experiment-evidence-record-v1", - "experiment-derived-measure-v1", - "experiment-run-v1", - } - ), - supported_media_types=frozenset({"application/json", "text/plain"}), - supported_sealing_modes=frozenset({"digest", "immutable-store"}), - supports_redaction=True, - supports_loss_disclosure=True, - supports_chain_of_custody=False, - ), - cleanup=CleanupCapabilities( - name="reference-emulation-cleanup", - supported_contract_versions=CLEANUP_CAPABILITY_REQUIRED_CONTRACTS, - supported_action_kinds=frozenset({"destroy", "reset", "restore", "compensate", "verify"}), - supported_verification_methods=frozenset({"probe", "receipt"}), - supports_reusable_state=True, - supports_residual_state_disclosure=True, - ), + participant_runtime=_participant_runtime_capabilities(), + observation=_observation_capabilities(), + cleanup=_cleanup_capabilities(), time=_time_capabilities(enabled=with_time), ) diff --git a/implementations/python/packages/raes_runtime/participant_activity.py b/implementations/python/packages/raes_runtime/participant_activity.py index c34c46ec4..aeef0ab49 100644 --- a/implementations/python/packages/raes_runtime/participant_activity.py +++ b/implementations/python/packages/raes_runtime/participant_activity.py @@ -39,6 +39,17 @@ class ParticipantActivityTimingSelection: disposition: str +@dataclass(frozen=True) +class ParticipantActivityDrawContext: + """Stable address inputs shared by occurrence-local random draws.""" + + policy: ParticipantAutonomousExecutionRuntime + participant_address: str + time_segment: int + occurrence_ordinal: int + control: ParticipantActivityRandomControl + + def resolve_participant_activity_controls( controls: Iterable[ExperimentStochasticControlModel], ) -> dict[str, ParticipantActivityRandomControl]: @@ -103,12 +114,8 @@ def activity_draw_address( def draw_activity_integer( + context: ParticipantActivityDrawContext, *, - policy: ParticipantAutonomousExecutionRuntime, - participant_address: str, - time_segment: int, - occurrence_ordinal: int, - control: ParticipantActivityRandomControl, local_coordinate: int, minimum: int, maximum: int, @@ -116,14 +123,14 @@ def draw_activity_integer( """Draw one bounded value from a stable occurrence-local coordinate.""" draw = draw_bounded_integer( - profile_id=control.profile_id, - stream_key=control.stream_key, + profile_id=context.control.profile_id, + stream_key=context.control.stream_key, address=activity_draw_address( - policy=policy, - participant_address=participant_address, - time_segment=time_segment, - occurrence_ordinal=occurrence_ordinal, - control=control, + policy=context.policy, + participant_address=context.participant_address, + time_segment=context.time_segment, + occurrence_ordinal=context.occurrence_ordinal, + control=context.control, local_coordinate=local_coordinate, ), minimum=minimum, @@ -187,14 +194,16 @@ def next_activity_timing( item for item in time_model.progression_policies if item.address == policy.progression_policy_address ) step_ticks = progression.step_ticks if progression.advancement_mode == "stepped" else None - minimum = policy.timing_minimum_ticks // step_ticks if step_ticks is not None else policy.timing_minimum_ticks - maximum = policy.timing_maximum_ticks // step_ticks if step_ticks is not None else policy.timing_maximum_ticks + minimum = _timing_units(policy.timing_minimum_ticks, step_ticks) + maximum = _timing_units(policy.timing_maximum_ticks, step_ticks) interval_units = draw_activity_integer( - policy=policy, - participant_address=participant_address, - time_segment=time_segment, - occurrence_ordinal=occurrence_ordinal, - control=control, + ParticipantActivityDrawContext( + policy=policy, + participant_address=participant_address, + time_segment=time_segment, + occurrence_ordinal=occurrence_ordinal, + control=control, + ), local_coordinate=0, minimum=minimum, maximum=maximum, @@ -202,18 +211,40 @@ def next_activity_timing( interval = interval_units * step_ticks if step_ticks is not None else interval_units candidate = current_tick + interval if activity_tick_is_eligible(policy, time_model, candidate): - return ParticipantActivityTimingSelection(tick=candidate, disposition="drawn") - if policy.outside_window_disposition == "skip": - return ParticipantActivityTimingSelection(tick=None, disposition="drawn") + selection = ParticipantActivityTimingSelection(tick=candidate, disposition="drawn") + elif policy.outside_window_disposition == "skip": + selection = ParticipantActivityTimingSelection(tick=None, disposition="drawn") + else: + selection = ParticipantActivityTimingSelection( + tick=_next_activity_opening(policy, time_model, candidate, step_ticks), + disposition="next_opening", + ) + return selection + + +def _timing_units(ticks: int, step_ticks: int | None) -> int: + return ticks // step_ticks if step_ticks is not None else ticks + + +def _aligned_activity_tick(tick: int, step_ticks: int | None) -> int: + if step_ticks is not None and tick % step_ticks: + return tick + step_ticks - tick % step_ticks + return tick + + +def _next_activity_opening( + policy: ParticipantAutonomousExecutionRuntime, + time_model: CompiledTimeModel, + candidate: int, + step_ticks: int | None, +) -> int | None: work = _window_ranges(policy.work_window_addresses, time_model) for start, end in work: first_tick = start[0] + int(start[1] > 0) - normalized = max(candidate, first_tick) - if step_ticks is not None and normalized % step_ticks: - normalized += step_ticks - normalized % step_ticks + normalized = _aligned_activity_tick(max(candidate, first_tick), step_ticks) while (normalized, 0) < end: if activity_tick_is_eligible(policy, time_model, normalized): - return ParticipantActivityTimingSelection(tick=normalized, disposition="next_opening") + return normalized pause_end = max( ( pause_end[0] + int(pause_end[1] > 0) @@ -222,10 +253,8 @@ def next_activity_timing( ), default=normalized + 1, ) - normalized = pause_end - if step_ticks is not None and normalized % step_ticks: - normalized += step_ticks - normalized % step_ticks - return ParticipantActivityTimingSelection(tick=None, disposition="next_opening") + normalized = _aligned_activity_tick(pause_end, step_ticks) + return None def next_activity_tick( @@ -266,11 +295,13 @@ def select_activity_candidate( return None total = sum(policy.action_candidate_weights[index] for index in eligible_indices) selected = draw_activity_integer( - policy=policy, - participant_address=participant_address, - time_segment=time_segment, - occurrence_ordinal=occurrence_ordinal, - control=control, + ParticipantActivityDrawContext( + policy=policy, + participant_address=participant_address, + time_segment=time_segment, + occurrence_ordinal=occurrence_ordinal, + control=control, + ), local_coordinate=1, minimum=0, maximum=total - 1, @@ -284,6 +315,7 @@ def select_activity_candidate( __all__ = [ + "ParticipantActivityDrawContext", "ParticipantActivityRandomControl", "activity_draw_address", "activity_control_for", diff --git a/implementations/python/packages/raes_runtime/participant_activity_support.py b/implementations/python/packages/raes_runtime/participant_activity_support.py index 66be116da..b7355b89c 100644 --- a/implementations/python/packages/raes_runtime/participant_activity_support.py +++ b/implementations/python/packages/raes_runtime/participant_activity_support.py @@ -95,9 +95,11 @@ def _activity_provenance( occurrence_ordinal=state.occurrence_ordinal, retry_ordinal=state.current_retry - 1, ) - disposition = ( - "retry" if state.current_retry else ("burst" if state.burst_position else state.next_timing_disposition) - ) + disposition = state.next_timing_disposition + if state.burst_position: + disposition = "burst" + if state.current_retry: + disposition = "retry" return ParticipantActivityOccurrenceProvenanceModel( policy_address=context.policy.address, policy_profile=context.policy.profile, diff --git a/implementations/python/packages/raes_runtime/participant_scheduler.py b/implementations/python/packages/raes_runtime/participant_scheduler.py index 096c8cd62..eab1b58d2 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler.py @@ -16,6 +16,7 @@ from raes_processor.models import CompiledTimeModel, ParticipantAutonomousExecutionRuntime from .participant_activity import ( + ParticipantActivityDrawContext, ParticipantActivityRandomControl, activity_control_for, draw_activity_integer, @@ -29,6 +30,7 @@ from .participant_scheduler_lifecycle import reset_policy_at_clock from .participant_scheduler_operations import ( SchedulerRunState, + participant_due_context, run_participant_due, run_policy_due_concurrently, ) @@ -68,91 +70,105 @@ def _state_identity(state: ParticipantAutonomousExecutionStateModel) -> tuple[ob ) -def _initialize_participant( - policy: ParticipantAutonomousExecutionRuntime, - time_model: CompiledTimeModel, +def _ensure_participant_episode( participant_runtime: object, snapshot: RuntimeSnapshot, participant_address: str, - activity_controls: dict[str, ParticipantActivityRandomControl], ) -> ApplyResult: - working = snapshot - changed: list[str] = [] - if participant_address not in working.participant_episode_results: - result = participant_runtime.initialize( - ParticipantEpisodeInitializeRequest( - participant_address=participant_address, - episode_id=f"{participant_address}-autonomous-0", - ), - working, - ) - if not result.success: - return result - working = result.snapshot - changed.extend(result.changed_addresses) - key = _state_key(policy.address, participant_address) - segment, _ = clock_coordinate(working, policy.clock_address) - activity_control = activity_control_for(policy, activity_controls) - if ( - policy.profile - in { - "participant-autonomous-execution/v2", - _RESOURCE_GOVERNED_PROFILE, - } - and activity_control is None - ): - return ApplyResult( - success=False, - snapshot=working, - diagnostics=[ - Diagnostic( - code="runtime.participant-activity-control-unbound", - domain="participant", - address=policy.address, - message=( - f"Participant activity policy requires admitted stochastic control " - f"{policy.stochastic_control_ref!r}." - ), - ) - ], - ) + if participant_address in snapshot.participant_episode_results: + return ApplyResult(success=True, snapshot=snapshot) + return participant_runtime.initialize( + ParticipantEpisodeInitializeRequest( + participant_address=participant_address, + episode_id=f"{participant_address}-autonomous-0", + ), + snapshot, + ) + + +def _activity_control_unbound_result( + policy: ParticipantAutonomousExecutionRuntime, + snapshot: RuntimeSnapshot, +) -> ApplyResult: + return ApplyResult( + success=False, + snapshot=snapshot, + diagnostics=[ + Diagnostic( + code="runtime.participant-activity-control-unbound", + domain="participant", + address=policy.address, + message=( + f"Participant activity policy requires admitted stochastic control " + f"{policy.stochastic_control_ref!r}." + ), + ) + ], + ) + + +def _initial_activity_schedule( + policy: ParticipantAutonomousExecutionRuntime, + time_model: CompiledTimeModel, + snapshot: RuntimeSnapshot, + participant_address: str, + segment: int, + activity_control: ParticipantActivityRandomControl | None, +) -> tuple[int | None, int, str]: if activity_control is None: first_tick, _ = _cadence(policy, time_model) - burst_size = 1 - timing_disposition = "cadence" - else: - current_tick = _clock_tick(working, policy.clock_address) - burst_size = draw_activity_integer( + return first_tick, 1, "cadence" + current_tick = _clock_tick(snapshot, policy.clock_address) + burst_size = draw_activity_integer( + ParticipantActivityDrawContext( policy=policy, participant_address=participant_address, time_segment=segment, occurrence_ordinal=0, control=activity_control, - local_coordinate=2, - minimum=1, - maximum=policy.max_burst_size, - ) - timing = next_activity_timing( - policy=policy, - time_model=time_model, - participant_address=participant_address, - time_segment=segment, - occurrence_ordinal=0, - current_tick=current_tick, - control=activity_control, - ) - first_tick = timing.tick - timing_disposition = timing.disposition - expected = ParticipantAutonomousExecutionStateModel( + ), + local_coordinate=2, + minimum=1, + maximum=policy.max_burst_size, + ) + timing = next_activity_timing( + policy=policy, + time_model=time_model, + participant_address=participant_address, + time_segment=segment, + occurrence_ordinal=0, + current_tick=current_tick, + control=activity_control, + ) + return timing.tick, burst_size, timing.disposition + + +def _initial_participant_state( + policy: ParticipantAutonomousExecutionRuntime, + time_model: CompiledTimeModel, + snapshot: RuntimeSnapshot, + participant_address: str, + segment: int, + activity_control: ParticipantActivityRandomControl | None, +) -> ParticipantAutonomousExecutionStateModel: + first_tick, burst_size, timing_disposition = _initial_activity_schedule( + policy, + time_model, + snapshot, + participant_address, + segment, + activity_control, + ) + return ParticipantAutonomousExecutionStateModel( policy_address=policy.address, policy_digest=_policy_digest(policy, time_model), participant_address=participant_address, - episode_id=working.participant_episode_results[participant_address]["episode_id"], + episode_id=snapshot.participant_episode_results[participant_address]["episode_id"], participant_implementation_ref=policy.participant_implementation_ref, clock_address=policy.clock_address, time_segment=segment, lifecycle_state="running" if first_tick is not None else "completed", - next_tick=first_tick if first_tick is not None else _clock_tick(working, policy.clock_address), + next_tick=first_tick if first_tick is not None else _clock_tick(snapshot, policy.clock_address), next_action_index=0, attempted_actions=0, succeeded_actions=0, @@ -164,13 +180,23 @@ def _initialize_participant( burst_size=burst_size, next_timing_disposition=timing_disposition, ) - states = dict(working.participant_autonomous_execution_states) + + +def _persist_initial_participant_state( + policy: ParticipantAutonomousExecutionRuntime, + snapshot: RuntimeSnapshot, + participant_address: str, + expected: ParticipantAutonomousExecutionStateModel, + changed: list[str], +) -> ApplyResult: + key = _state_key(policy.address, participant_address) + states = dict(snapshot.participant_autonomous_execution_states) if key in states and _state_identity(ParticipantAutonomousExecutionStateModel.model_validate(states[key])) != ( _state_identity(expected) ): return ApplyResult( success=False, - snapshot=working, + snapshot=snapshot, diagnostics=[ Diagnostic( code="runtime.participant-autonomous-state-conflict", @@ -180,16 +206,52 @@ def _initialize_participant( ) ], ) + working = snapshot if key not in states: states[key] = expected.model_dump(mode="json") - working = working.with_entries( - dict(working.entries), + working = snapshot.with_entries( + dict(snapshot.entries), participant_autonomous_execution_states=states, ) changed.append(key) return ApplyResult(success=True, snapshot=working, changed_addresses=changed) +def _initialize_participant( + policy: ParticipantAutonomousExecutionRuntime, + time_model: CompiledTimeModel, + participant_runtime: object, + snapshot: RuntimeSnapshot, + participant_address: str, + activity_controls: dict[str, ParticipantActivityRandomControl], +) -> ApplyResult: + episode_result = _ensure_participant_episode(participant_runtime, snapshot, participant_address) + if not episode_result.success: + return episode_result + working = episode_result.snapshot + changed = list(episode_result.changed_addresses) + segment, _ = clock_coordinate(working, policy.clock_address) + activity_control = activity_control_for(policy, activity_controls) + if ( + policy.profile + in { + "participant-autonomous-execution/v2", + _RESOURCE_GOVERNED_PROFILE, + } + and activity_control is None + ): + return _activity_control_unbound_result(policy, working) + expected = _initial_participant_state( + policy, + time_model, + working, + participant_address, + segment, + activity_control, + ) + return _persist_initial_participant_state(policy, working, participant_address, expected, changed) + + def _missing_execution_service_result( policy: ParticipantAutonomousExecutionRuntime, run: SchedulerRunState, @@ -223,14 +285,16 @@ def _run_serial_due( ) -> None: for participant_address in policy.participant_addresses: run_participant_due( - policy, - time_model, - participant_runtime, - participant_address, - current_tick, - cadence_ticks, + participant_due_context( + policy, + time_model, + participant_runtime, + participant_address, + current_tick, + cadence_ticks, + activity_controls, + ), run, - activity_controls, ) if run.failure is not None: break diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py index 85c371626..901cac398 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py @@ -301,7 +301,7 @@ def _finish_due_policy( cadence_ticks: int, run: SchedulerRunState, ) -> None: - from .participant_scheduler_operations import run_participant_due + from .participant_scheduler_operations import participant_due_context, run_participant_due if run.failure is not None: return @@ -309,12 +309,14 @@ def _finish_due_policy( return for participant_address in policy.participant_addresses: run_participant_due( - policy, - time_model, - participant_runtime, - participant_address, - current_tick, - cadence_ticks, + participant_due_context( + policy, + time_model, + participant_runtime, + participant_address, + current_tick, + cadence_ticks, + ), run, ) if run.failure is not None: diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_operations.py b/implementations/python/packages/raes_runtime/participant_scheduler_operations.py index 801fa2f21..f6eaee236 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_operations.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_operations.py @@ -2,7 +2,7 @@ from __future__ import annotations -from dataclasses import replace +from dataclasses import dataclass, replace from typing import cast from raes_contracts.contracts import ( @@ -17,6 +17,7 @@ from .participant_action_validation import autonomous_action_result_violation from .participant_activity import ( + ParticipantActivityDrawContext, ParticipantActivityRandomControl, activity_control_for, draw_activity_integer, @@ -222,42 +223,45 @@ def _run_one_due_action( return next_state -def _next_activity_occurrence_state( +def _activity_attempt_is_retryable( context: _DueActionContext, state: ParticipantAutonomousExecutionStateModel, - request: ParticipantActionAdmissionRequest, *, action_succeeded: bool, failure_class: str | None, protocol_failure: bool, -) -> ParticipantAutonomousExecutionStateModel: - policy = context.policy - control = context.activity_control - if control is None: - raise ValueError("participant activity execution requires a random control") + attempted: int, +) -> bool: index = state.next_action_index - candidate_id = policy.action_candidate_ids[index] - attempted = state.attempted_actions + 1 - failed = state.failed_actions + (0 if action_succeeded else 1) - retryable = ( + return ( not protocol_failure and not action_succeeded - and failure_class in policy.action_candidate_retry_failure_classes[index] - and state.current_retry < policy.action_candidate_max_retries[index] - and attempted < policy.max_action_attempts + and failure_class in context.policy.action_candidate_retry_failure_classes[index] + and state.current_retry < context.policy.action_candidate_max_retries[index] + and attempted < context.policy.max_action_attempts ) - if retryable: - return state.model_copy( - update={ - "next_tick": context.current_tick, - "attempted_actions": attempted, - "failed_actions": failed, - "current_retry": state.current_retry + 1, - "last_candidate_id": candidate_id, - "last_action_instance_id": request.action_instance_id, - } - ) + +@dataclass(frozen=True) +class _ActivityProgress: + candidate_id: str + completed: list[str] + cooldowns: dict[str, int] + occurrence: int + lifecycle: str + + +def _completed_activity_progress( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + *, + action_succeeded: bool, + protocol_failure: bool, + attempted: int, +) -> _ActivityProgress: + policy = context.policy + index = state.next_action_index + candidate_id = policy.action_candidate_ids[index] completed = list(state.completed_candidate_ids) if action_succeeded and candidate_id not in completed: completed.append(candidate_id) @@ -269,57 +273,148 @@ def _next_activity_occurrence_state( lifecycle = "failed" elif occurrence >= policy.max_occurrences or attempted >= policy.max_action_attempts: lifecycle = "completed" + return _ActivityProgress( + candidate_id=candidate_id, + completed=completed, + cooldowns=cooldowns, + occurrence=occurrence, + lifecycle=lifecycle, + ) + + +@dataclass(frozen=True) +class _ActivitySchedule: + lifecycle: str + next_tick: int + burst_position: int + burst_size: int + timing_disposition: str + +def _next_activity_schedule( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + control: ParticipantActivityRandomControl, + progress: _ActivityProgress, +) -> _ActivitySchedule: + lifecycle = progress.lifecycle burst_position = state.burst_position burst_size = state.burst_size next_tick = context.current_tick - if lifecycle == "running": - if burst_position + 1 < burst_size: - burst_position += 1 - else: - burst_position = 0 - burst_size = draw_activity_integer( - policy=policy, - participant_address=context.participant_address, - time_segment=state.time_segment, - occurrence_ordinal=occurrence, - control=control, - local_coordinate=2, - minimum=1, - maximum=policy.max_burst_size, - ) - timing = next_activity_timing( - policy=policy, - time_model=context.time_model, + timing_disposition = state.next_timing_disposition + if lifecycle == "running" and burst_position + 1 < burst_size: + burst_position += 1 + elif lifecycle == "running": + burst_position = 0 + burst_size = draw_activity_integer( + ParticipantActivityDrawContext( + policy=context.policy, participant_address=context.participant_address, time_segment=state.time_segment, - occurrence_ordinal=occurrence, - current_tick=context.current_tick, + occurrence_ordinal=progress.occurrence, control=control, - ) - selected_tick = timing.tick - if selected_tick is None: - lifecycle = "completed" - else: - next_tick = selected_tick + ), + local_coordinate=2, + minimum=1, + maximum=context.policy.max_burst_size, + ) + timing = next_activity_timing( + policy=context.policy, + time_model=context.time_model, + participant_address=context.participant_address, + time_segment=state.time_segment, + occurrence_ordinal=progress.occurrence, + current_tick=context.current_tick, + control=control, + ) + timing_disposition = timing.disposition + if timing.tick is None: + lifecycle = "completed" + else: + next_tick = timing.tick + return _ActivitySchedule( + lifecycle=lifecycle, + next_tick=next_tick, + burst_position=burst_position, + burst_size=burst_size, + timing_disposition=timing_disposition, + ) + + +def _activity_retry_state( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + request: ParticipantActionAdmissionRequest, + *, + attempted: int, + failed: int, +) -> ParticipantAutonomousExecutionStateModel: + candidate_id = context.policy.action_candidate_ids[state.next_action_index] return state.model_copy( update={ - "lifecycle_state": lifecycle, - "next_tick": next_tick, + "next_tick": context.current_tick, + "attempted_actions": attempted, + "failed_actions": failed, + "current_retry": state.current_retry + 1, + "last_candidate_id": candidate_id, + "last_action_instance_id": request.action_instance_id, + } + ) + + +def _next_activity_occurrence_state( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + request: ParticipantActionAdmissionRequest, + *, + action_succeeded: bool, + failure_class: str | None, + protocol_failure: bool, +) -> ParticipantAutonomousExecutionStateModel: + control = context.activity_control + if control is None: + raise ValueError("participant activity execution requires a random control") + attempted = state.attempted_actions + 1 + failed = state.failed_actions + (0 if action_succeeded else 1) + if _activity_attempt_is_retryable( + context, + state, + action_succeeded=action_succeeded, + failure_class=failure_class, + protocol_failure=protocol_failure, + attempted=attempted, + ): + return _activity_retry_state( + context, + state, + request, + attempted=attempted, + failed=failed, + ) + progress = _completed_activity_progress( + context, + state, + action_succeeded=action_succeeded, + protocol_failure=protocol_failure, + attempted=attempted, + ) + schedule = _next_activity_schedule(context, state, control, progress) + return state.model_copy( + update={ + "lifecycle_state": schedule.lifecycle, + "next_tick": schedule.next_tick, "attempted_actions": attempted, "succeeded_actions": state.succeeded_actions + (1 if action_succeeded else 0), "failed_actions": failed, - "occurrence_ordinal": occurrence, + "occurrence_ordinal": progress.occurrence, "current_retry": 0, - "burst_position": burst_position, - "burst_size": burst_size, - "last_candidate_id": candidate_id, - "completed_candidate_ids": completed, - "candidate_cooldown_until": cooldowns, + "burst_position": schedule.burst_position, + "burst_size": schedule.burst_size, + "last_candidate_id": progress.candidate_id, + "completed_candidate_ids": progress.completed, + "candidate_cooldown_until": progress.cooldowns, "last_action_instance_id": request.action_instance_id, - "next_timing_disposition": ( - timing.disposition if lifecycle == "running" and burst_position == 0 else state.next_timing_disposition - ), + "next_timing_disposition": schedule.timing_disposition, } ) @@ -386,82 +481,95 @@ def _run_one_activity_action( return next_state -def _run_participant_activity_due( +def _activity_action_is_due( context: _DueActionContext, state: ParticipantAutonomousExecutionStateModel, run: SchedulerRunState, -) -> None: - while ( - state.lifecycle_state == "running" - and state.next_tick == context.current_tick - and state.attempted_actions < context.policy.max_action_attempts - and run.failure is None - ): - eligible = activity_eligible_indices(context.policy, state, context.current_tick) +) -> bool: + return all( + ( + state.lifecycle_state == "running", + state.next_tick == context.current_tick, + state.attempted_actions < context.policy.max_action_attempts, + run.failure is None, + ) + ) + + +def _selected_activity_index( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, +) -> int | None: + if state.current_retry: + return state.next_action_index + control = context.activity_control + if control is None: + raise ValueError("participant activity execution requires a random control") + return select_activity_candidate( + policy=context.policy, + participant_address=context.participant_address, + time_segment=state.time_segment, + occurrence_ordinal=state.occurrence_ordinal, + control=control, + eligible_indices=activity_eligible_indices(context.policy, state, context.current_tick), + ) + + +def _empty_activity_state( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, +) -> ParticipantAutonomousExecutionStateModel: + lifecycle = "completed" if context.policy.empty_eligible_disposition == "complete" else "running" + selected_tick = None + if lifecycle == "running": control = context.activity_control if control is None: raise ValueError("participant activity execution requires a random control") - selected = ( - state.next_action_index - if state.current_retry - else select_activity_candidate( - policy=context.policy, - participant_address=context.participant_address, - time_segment=state.time_segment, - occurrence_ordinal=state.occurrence_ordinal, - control=control, - eligible_indices=eligible, - ) - ) + selected_tick = next_activity_timing( + policy=context.policy, + time_model=context.time_model, + participant_address=context.participant_address, + time_segment=state.time_segment, + occurrence_ordinal=state.occurrence_ordinal, + current_tick=context.current_tick, + control=control, + ).tick + if selected_tick is None: + lifecycle = "completed" + return state.model_copy( + update={ + "lifecycle_state": lifecycle, + "next_tick": selected_tick if selected_tick is not None else context.current_tick, + } + ) + + +def _run_participant_activity_due( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + run: SchedulerRunState, +) -> None: + while _activity_action_is_due(context, state, run): + selected = _selected_activity_index(context, state) if selected is None: - lifecycle = "completed" if context.policy.empty_eligible_disposition == "complete" else "running" - selected_tick = ( - None - if lifecycle == "completed" - else next_activity_timing( - policy=context.policy, - time_model=context.time_model, - participant_address=context.participant_address, - time_segment=state.time_segment, - occurrence_ordinal=state.occurrence_ordinal, - current_tick=context.current_tick, - control=control, - ).tick - ) - if selected_tick is None: - lifecycle = "completed" - state = state.model_copy( - update={ - "lifecycle_state": lifecycle, - "next_tick": selected_tick if selected_tick is not None else context.current_tick, - } - ) + state = _empty_activity_state(context, state) persist_activity_state(run, context.key, state) return state = state.model_copy(update={"next_action_index": selected}) state = _run_one_activity_action(context, state, run) -def run_participant_due( +def participant_due_context( policy: ParticipantAutonomousExecutionRuntime, time_model: CompiledTimeModel, participant_runtime: object, participant_address: str, current_tick: int, cadence_ticks: int, - run: SchedulerRunState, activity_controls: dict[str, ParticipantActivityRandomControl] | None = None, -) -> None: - """Run one participant at the current governed cadence boundary.""" - +) -> _DueActionContext: key = f"{policy.address}.state.{participant_address}" - state = ParticipantAutonomousExecutionStateModel.model_validate( - run.working.participant_autonomous_execution_states[key] - ) - if state.lifecycle_state == "running" and state.next_tick < current_tick: - run.failure = cadence_missed_result(run.working, key, current_tick, state) - return - action_context = _DueActionContext( + return _DueActionContext( policy=policy, time_model=time_model, participant_runtime=participant_runtime, @@ -471,28 +579,55 @@ def run_participant_due( cadence_ticks=cadence_ticks, activity_control=activity_control_for(policy, activity_controls or {}), ) - if policy.profile in { + + +def _legacy_action_is_due( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + run: SchedulerRunState, +) -> bool: + return all( + ( + state.lifecycle_state == "running", + state.next_tick == context.current_tick, + state.attempted_actions < context.policy.max_action_attempts, + run.failure is None, + ) + ) + + +def _run_legacy_participant_due( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + run: SchedulerRunState, +) -> None: + while _legacy_action_is_due(context, state, run): + state = _run_one_due_action(context, state, run) + + +def run_participant_due( + context: _DueActionContext, + run: SchedulerRunState, +) -> None: + """Run one participant at the current governed cadence boundary.""" + + state = ParticipantAutonomousExecutionStateModel.model_validate( + run.working.participant_autonomous_execution_states[context.key] + ) + if state.lifecycle_state == "running" and state.next_tick < context.current_tick: + run.failure = cadence_missed_result(run.working, context.key, context.current_tick, state) + elif context.policy.profile in { "participant-autonomous-execution/v2", "participant-autonomous-execution/v3", }: - _run_participant_activity_due(action_context, state, run) - return - action_is_due = ( - state.lifecycle_state == "running" - and state.next_tick == current_tick - and state.attempted_actions < policy.max_action_attempts - ) - while action_is_due and run.failure is None: - state = _run_one_due_action(action_context, state, run) - action_is_due = ( - state.lifecycle_state == "running" - and state.next_tick == current_tick - and state.attempted_actions < policy.max_action_attempts - ) + _run_participant_activity_due(context, state, run) + else: + _run_legacy_participant_due(context, state, run) __all__ = [ "SchedulerRunState", + "participant_due_context", "run_participant_due", "run_policy_due_concurrently", ] diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_reset.py b/implementations/python/packages/raes_runtime/participant_scheduler_reset.py index 1ea65ebac..4d98a9dff 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_reset.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_reset.py @@ -9,7 +9,12 @@ from raes_contracts.runtime_state import ApplyResult, RuntimeSnapshot from raes_processor.models import CompiledTimeModel, ParticipantAutonomousExecutionRuntime -from .participant_activity import ParticipantActivityRandomControl, draw_activity_integer, next_activity_timing +from .participant_activity import ( + ParticipantActivityDrawContext, + ParticipantActivityRandomControl, + draw_activity_integer, + next_activity_timing, +) from .participant_scheduler_time import cadence @@ -82,11 +87,13 @@ def reset_scheduler_participant( burst_size = 1 if context.activity_control is not None: burst_size = draw_activity_integer( - policy=context.policy, - participant_address=participant_address, - time_segment=context.segment, - occurrence_ordinal=0, - control=context.activity_control, + ParticipantActivityDrawContext( + policy=context.policy, + participant_address=participant_address, + time_segment=context.segment, + occurrence_ordinal=0, + control=context.activity_control, + ), local_coordinate=2, minimum=1, maximum=context.policy.max_burst_size, diff --git a/implementations/python/tests/test_dsl_437_benign_participant_execution.py b/implementations/python/tests/test_dsl_437_benign_participant_execution.py index 627056bb6..0f5d69547 100644 --- a/implementations/python/tests/test_dsl_437_benign_participant_execution.py +++ b/implementations/python/tests/test_dsl_437_benign_participant_execution.py @@ -755,25 +755,28 @@ def test_activity_policy_v2_rejects_non_window_availability_constraint() -> None payload["behavior_specifications"]["participant-behavior"]["autonomous_execution"]["work_window_refs"] = [ "green-cadence" ] + rendered = yaml.safe_dump(payload, sort_keys=False) with pytest.raises(SDLValidationError, match="work and pause refs must resolve to window constraints"): - parse_sdl(yaml.safe_dump(payload, sort_keys=False)) + parse_sdl(rendered) def test_activity_policy_v2_rejects_timing_bounds_unreachable_by_stepped_progression() -> None: payload = yaml.safe_load(_activity_policy_yaml()) payload["behavior_specifications"]["participant-behavior"]["autonomous_execution"]["timing"]["minimum_ticks"] = 15 + rendered = yaml.safe_dump(payload, sort_keys=False) with pytest.raises(SDLValidationError, match="activity timing bounds are unreachable by stepped progression"): - parse_sdl(yaml.safe_dump(payload, sort_keys=False)) + parse_sdl(rendered) def test_activity_policy_v2_rejects_window_for_unrelated_subject() -> None: payload = yaml.safe_load(_activity_policy_yaml()) payload["temporal_constraints"]["work-window"]["subject_refs"] = ["nodes.customer-portal"] + rendered = yaml.safe_dump(payload, sort_keys=False) with pytest.raises(SDLValidationError, match="must name the behavior specification or every governed participant"): - parse_sdl(yaml.safe_dump(payload, sort_keys=False)) + parse_sdl(rendered) def test_non_evaluated_autonomous_participant_must_be_green() -> None: @@ -1488,8 +1491,9 @@ def test_runtime_manager_fails_closed_for_unresolved_governed_activity_entropy() } ) + target = create_stub_target() with pytest.raises(ValueError, match="governed entropy without a resolver"): - RuntimeManager(create_stub_target(), stochastic_controls=[governed]) + RuntimeManager(target, stochastic_controls=[governed]) def test_runtime_manager_rolls_back_clock_when_participant_reset_fails() -> None: diff --git a/implementations/python/tests/test_random_stream_profile.py b/implementations/python/tests/test_random_stream_profile.py index e3788f0be..9baf6ed9e 100644 --- a/implementations/python/tests/test_random_stream_profile.py +++ b/implementations/python/tests/test_random_stream_profile.py @@ -97,10 +97,11 @@ def test_rejects_unsupported_but_syntactically_valid_id(self) -> None: random_stream_profile_path("nonexistent-profile-v1") def test_supported_profile_ids_contains_blake3_xof(self) -> None: - assert { + expected_ids = { "blake3-xof-participant-v1", "blake3-xof-v1", - } == SUPPORTED_RANDOM_STREAM_PROFILE_IDS + } + assert not SUPPORTED_RANDOM_STREAM_PROFILE_IDS.symmetric_difference(expected_ids) def test_load_unsupported_profile_fails_closed_without_file_probe(self) -> None: with pytest.raises(ValueError, match="unsupported"): From 18c9ab54abab8c66f33d7582b6209c02496296fc Mon Sep 17 00:00:00 2001 From: Brad Edwards Date: Mon, 27 Jul 2026 08:10:57 +0200 Subject: [PATCH 2/5] fix: split scheduler helper modules --- .../raes_runtime/participant_scheduler.py | 228 +---------------- .../participant_scheduler_activity_state.py | 213 ++++++++++++++++ .../participant_scheduler_initialization.py | 229 ++++++++++++++++++ .../participant_scheduler_operations.py | 203 +--------------- 4 files changed, 450 insertions(+), 423 deletions(-) create mode 100644 implementations/python/packages/raes_runtime/participant_scheduler_activity_state.py create mode 100644 implementations/python/packages/raes_runtime/participant_scheduler_initialization.py diff --git a/implementations/python/packages/raes_runtime/participant_scheduler.py b/implementations/python/packages/raes_runtime/participant_scheduler.py index eab1b58d2..0f6a68333 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler.py @@ -4,23 +4,13 @@ from collections.abc import Iterable -from raes_contracts.contracts import ( - ParticipantAutonomousExecutionStateModel, -) from raes_contracts.contracts.participant_execution import ParticipantExecutionServiceStateModel from raes_contracts.diagnostics import Diagnostic -from raes_contracts.participant_episode import ( - ParticipantEpisodeInitializeRequest, -) from raes_contracts.runtime_state import ApplyResult, RuntimeSnapshot from raes_processor.models import CompiledTimeModel, ParticipantAutonomousExecutionRuntime from .participant_activity import ( - ParticipantActivityDrawContext, ParticipantActivityRandomControl, - activity_control_for, - draw_activity_integer, - next_activity_timing, ) from .participant_execution_scheduler_state import ( execution_service_state, @@ -28,6 +18,10 @@ ) from .participant_resource_budgets import initialize_participant_resource_budgets from .participant_scheduler_lifecycle import reset_policy_at_clock +from .participant_scheduler_initialization import ( + clock_tick as _clock_tick, + initialize_participant as _initialize_participant, +) from .participant_scheduler_operations import ( SchedulerRunState, participant_due_context, @@ -38,220 +32,6 @@ from .participant_scheduler_time import cadence as _cadence from .participant_scheduler_time import clock_coordinate -_RESOURCE_GOVERNED_PROFILE = "participant-autonomous-execution/v3" - - -def _state_key(policy_address: str, participant_address: str) -> str: - return f"{policy_address}.state.{participant_address}" - - -def _clock_tick(snapshot: RuntimeSnapshot, clock_address: str) -> int: - if snapshot.time_model_state is None: - raise ValueError("autonomous participant execution requires typed shared-time state") - clock = snapshot.time_model_state.clocks.get(clock_address) - if clock is None: - raise ValueError(f"autonomous participant clock {clock_address!r} has no runtime state") - return clock.coordinate.tick - - -def _state_identity(state: ParticipantAutonomousExecutionStateModel) -> tuple[object, ...]: - return ( - state.policy_address, - state.policy_digest, - state.participant_address, - state.episode_id, - state.participant_implementation_ref, - state.clock_address, - state.time_segment, - state.profile, - state.random_control_id, - state.random_profile_id, - state.random_namespace, - ) - - -def _ensure_participant_episode( - participant_runtime: object, - snapshot: RuntimeSnapshot, - participant_address: str, -) -> ApplyResult: - if participant_address in snapshot.participant_episode_results: - return ApplyResult(success=True, snapshot=snapshot) - return participant_runtime.initialize( - ParticipantEpisodeInitializeRequest( - participant_address=participant_address, - episode_id=f"{participant_address}-autonomous-0", - ), - snapshot, - ) - - -def _activity_control_unbound_result( - policy: ParticipantAutonomousExecutionRuntime, - snapshot: RuntimeSnapshot, -) -> ApplyResult: - return ApplyResult( - success=False, - snapshot=snapshot, - diagnostics=[ - Diagnostic( - code="runtime.participant-activity-control-unbound", - domain="participant", - address=policy.address, - message=( - f"Participant activity policy requires admitted stochastic control " - f"{policy.stochastic_control_ref!r}." - ), - ) - ], - ) - - -def _initial_activity_schedule( - policy: ParticipantAutonomousExecutionRuntime, - time_model: CompiledTimeModel, - snapshot: RuntimeSnapshot, - participant_address: str, - segment: int, - activity_control: ParticipantActivityRandomControl | None, -) -> tuple[int | None, int, str]: - if activity_control is None: - first_tick, _ = _cadence(policy, time_model) - return first_tick, 1, "cadence" - current_tick = _clock_tick(snapshot, policy.clock_address) - burst_size = draw_activity_integer( - ParticipantActivityDrawContext( - policy=policy, - participant_address=participant_address, - time_segment=segment, - occurrence_ordinal=0, - control=activity_control, - ), - local_coordinate=2, - minimum=1, - maximum=policy.max_burst_size, - ) - timing = next_activity_timing( - policy=policy, - time_model=time_model, - participant_address=participant_address, - time_segment=segment, - occurrence_ordinal=0, - current_tick=current_tick, - control=activity_control, - ) - return timing.tick, burst_size, timing.disposition - - -def _initial_participant_state( - policy: ParticipantAutonomousExecutionRuntime, - time_model: CompiledTimeModel, - snapshot: RuntimeSnapshot, - participant_address: str, - segment: int, - activity_control: ParticipantActivityRandomControl | None, -) -> ParticipantAutonomousExecutionStateModel: - first_tick, burst_size, timing_disposition = _initial_activity_schedule( - policy, - time_model, - snapshot, - participant_address, - segment, - activity_control, - ) - return ParticipantAutonomousExecutionStateModel( - policy_address=policy.address, - policy_digest=_policy_digest(policy, time_model), - participant_address=participant_address, - episode_id=snapshot.participant_episode_results[participant_address]["episode_id"], - participant_implementation_ref=policy.participant_implementation_ref, - clock_address=policy.clock_address, - time_segment=segment, - lifecycle_state="running" if first_tick is not None else "completed", - next_tick=first_tick if first_tick is not None else _clock_tick(snapshot, policy.clock_address), - next_action_index=0, - attempted_actions=0, - succeeded_actions=0, - failed_actions=0, - profile=policy.profile, - random_control_id=activity_control.control_id if activity_control is not None else None, - random_profile_id=activity_control.profile_id if activity_control is not None else None, - random_namespace=activity_control.namespace if activity_control is not None else None, - burst_size=burst_size, - next_timing_disposition=timing_disposition, - ) - - -def _persist_initial_participant_state( - policy: ParticipantAutonomousExecutionRuntime, - snapshot: RuntimeSnapshot, - participant_address: str, - expected: ParticipantAutonomousExecutionStateModel, - changed: list[str], -) -> ApplyResult: - key = _state_key(policy.address, participant_address) - states = dict(snapshot.participant_autonomous_execution_states) - if key in states and _state_identity(ParticipantAutonomousExecutionStateModel.model_validate(states[key])) != ( - _state_identity(expected) - ): - return ApplyResult( - success=False, - snapshot=snapshot, - diagnostics=[ - Diagnostic( - code="runtime.participant-autonomous-state-conflict", - domain="participant", - address=policy.address, - message="Existing autonomous participant state does not match the compiled policy.", - ) - ], - ) - working = snapshot - if key not in states: - states[key] = expected.model_dump(mode="json") - working = snapshot.with_entries( - dict(snapshot.entries), - participant_autonomous_execution_states=states, - ) - changed.append(key) - return ApplyResult(success=True, snapshot=working, changed_addresses=changed) - - -def _initialize_participant( - policy: ParticipantAutonomousExecutionRuntime, - time_model: CompiledTimeModel, - participant_runtime: object, - snapshot: RuntimeSnapshot, - participant_address: str, - activity_controls: dict[str, ParticipantActivityRandomControl], -) -> ApplyResult: - episode_result = _ensure_participant_episode(participant_runtime, snapshot, participant_address) - if not episode_result.success: - return episode_result - working = episode_result.snapshot - changed = list(episode_result.changed_addresses) - segment, _ = clock_coordinate(working, policy.clock_address) - activity_control = activity_control_for(policy, activity_controls) - if ( - policy.profile - in { - "participant-autonomous-execution/v2", - _RESOURCE_GOVERNED_PROFILE, - } - and activity_control is None - ): - return _activity_control_unbound_result(policy, working) - expected = _initial_participant_state( - policy, - time_model, - working, - participant_address, - segment, - activity_control, - ) - return _persist_initial_participant_state(policy, working, participant_address, expected, changed) - - def _missing_execution_service_result( policy: ParticipantAutonomousExecutionRuntime, run: SchedulerRunState, diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_activity_state.py b/implementations/python/packages/raes_runtime/participant_scheduler_activity_state.py new file mode 100644 index 000000000..c5002afd9 --- /dev/null +++ b/implementations/python/packages/raes_runtime/participant_scheduler_activity_state.py @@ -0,0 +1,213 @@ +"""State transitions for participant activity occurrences.""" + +from dataclasses import dataclass + +from raes_contracts.contracts import ParticipantAutonomousExecutionStateModel +from raes_contracts.participant_binding import ParticipantActionAdmissionRequest + +from .participant_activity import ( + ParticipantActivityDrawContext, + ParticipantActivityRandomControl, + draw_activity_integer, + next_activity_timing, +) +from .participant_scheduler_types import _DueActionContext + + +def _activity_attempt_is_retryable( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + *, + action_succeeded: bool, + failure_class: str | None, + protocol_failure: bool, + attempted: int, +) -> bool: + index = state.next_action_index + return ( + not protocol_failure + and not action_succeeded + and failure_class in context.policy.action_candidate_retry_failure_classes[index] + and state.current_retry < context.policy.action_candidate_max_retries[index] + and attempted < context.policy.max_action_attempts + ) + + +@dataclass(frozen=True) +class _ActivityProgress: + candidate_id: str + completed: list[str] + cooldowns: dict[str, int] + occurrence: int + lifecycle: str + + +def _completed_activity_progress( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + *, + action_succeeded: bool, + protocol_failure: bool, + attempted: int, +) -> _ActivityProgress: + policy = context.policy + index = state.next_action_index + candidate_id = policy.action_candidate_ids[index] + completed = list(state.completed_candidate_ids) + if action_succeeded and candidate_id not in completed: + completed.append(candidate_id) + cooldowns = dict(state.candidate_cooldown_until) + cooldowns[candidate_id] = context.current_tick + policy.action_candidate_cooldown_ticks[index] + occurrence = state.occurrence_ordinal + 1 + lifecycle = state.lifecycle_state + if protocol_failure or (not action_succeeded and policy.failure_policy == "stop"): + lifecycle = "failed" + elif occurrence >= policy.max_occurrences or attempted >= policy.max_action_attempts: + lifecycle = "completed" + return _ActivityProgress( + candidate_id=candidate_id, + completed=completed, + cooldowns=cooldowns, + occurrence=occurrence, + lifecycle=lifecycle, + ) + + +@dataclass(frozen=True) +class _ActivitySchedule: + lifecycle: str + next_tick: int + burst_position: int + burst_size: int + timing_disposition: str + + +def _next_activity_schedule( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + control: ParticipantActivityRandomControl, + progress: _ActivityProgress, +) -> _ActivitySchedule: + lifecycle = progress.lifecycle + burst_position = state.burst_position + burst_size = state.burst_size + next_tick = context.current_tick + timing_disposition = state.next_timing_disposition + if lifecycle == "running" and burst_position + 1 < burst_size: + burst_position += 1 + elif lifecycle == "running": + burst_position = 0 + burst_size = draw_activity_integer( + ParticipantActivityDrawContext( + policy=context.policy, + participant_address=context.participant_address, + time_segment=state.time_segment, + occurrence_ordinal=progress.occurrence, + control=control, + ), + local_coordinate=2, + minimum=1, + maximum=context.policy.max_burst_size, + ) + timing = next_activity_timing( + policy=context.policy, + time_model=context.time_model, + participant_address=context.participant_address, + time_segment=state.time_segment, + occurrence_ordinal=progress.occurrence, + current_tick=context.current_tick, + control=control, + ) + timing_disposition = timing.disposition + if timing.tick is None: + lifecycle = "completed" + else: + next_tick = timing.tick + return _ActivitySchedule( + lifecycle=lifecycle, + next_tick=next_tick, + burst_position=burst_position, + burst_size=burst_size, + timing_disposition=timing_disposition, + ) + + +def _activity_retry_state( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + request: ParticipantActionAdmissionRequest, + *, + attempted: int, + failed: int, +) -> ParticipantAutonomousExecutionStateModel: + candidate_id = context.policy.action_candidate_ids[state.next_action_index] + return state.model_copy( + update={ + "next_tick": context.current_tick, + "attempted_actions": attempted, + "failed_actions": failed, + "current_retry": state.current_retry + 1, + "last_candidate_id": candidate_id, + "last_action_instance_id": request.action_instance_id, + } + ) + + +def next_activity_occurrence_state( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + request: ParticipantActionAdmissionRequest, + *, + action_succeeded: bool, + failure_class: str | None, + protocol_failure: bool, +) -> ParticipantAutonomousExecutionStateModel: + control = context.activity_control + if control is None: + raise ValueError("participant activity execution requires a random control") + attempted = state.attempted_actions + 1 + failed = state.failed_actions + (0 if action_succeeded else 1) + if _activity_attempt_is_retryable( + context, + state, + action_succeeded=action_succeeded, + failure_class=failure_class, + protocol_failure=protocol_failure, + attempted=attempted, + ): + return _activity_retry_state( + context, + state, + request, + attempted=attempted, + failed=failed, + ) + progress = _completed_activity_progress( + context, + state, + action_succeeded=action_succeeded, + protocol_failure=protocol_failure, + attempted=attempted, + ) + schedule = _next_activity_schedule(context, state, control, progress) + return state.model_copy( + update={ + "lifecycle_state": schedule.lifecycle, + "next_tick": schedule.next_tick, + "attempted_actions": attempted, + "succeeded_actions": state.succeeded_actions + (1 if action_succeeded else 0), + "failed_actions": failed, + "occurrence_ordinal": progress.occurrence, + "current_retry": 0, + "burst_position": schedule.burst_position, + "burst_size": schedule.burst_size, + "last_candidate_id": progress.candidate_id, + "completed_candidate_ids": progress.completed, + "candidate_cooldown_until": progress.cooldowns, + "last_action_instance_id": request.action_instance_id, + "next_timing_disposition": schedule.timing_disposition, + } + ) + + +__all__ = ["next_activity_occurrence_state"] diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_initialization.py b/implementations/python/packages/raes_runtime/participant_scheduler_initialization.py new file mode 100644 index 000000000..233620aa2 --- /dev/null +++ b/implementations/python/packages/raes_runtime/participant_scheduler_initialization.py @@ -0,0 +1,229 @@ +"""Participant state initialization for autonomous scheduling.""" + +from raes_contracts.contracts import ParticipantAutonomousExecutionStateModel +from raes_contracts.diagnostics import Diagnostic +from raes_contracts.participant_episode import ParticipantEpisodeInitializeRequest +from raes_contracts.runtime_state import ApplyResult, RuntimeSnapshot +from raes_processor.models import CompiledTimeModel, ParticipantAutonomousExecutionRuntime + +from .participant_activity import ( + ParticipantActivityDrawContext, + ParticipantActivityRandomControl, + activity_control_for, + draw_activity_integer, + next_activity_timing, +) +from .participant_scheduler_policy import _policy_digest +from .participant_scheduler_time import cadence, clock_coordinate + +_RESOURCE_GOVERNED_PROFILE = "participant-autonomous-execution/v3" + + +def clock_tick(snapshot: RuntimeSnapshot, clock_address: str) -> int: + if snapshot.time_model_state is None: + raise ValueError("autonomous participant execution requires typed shared-time state") + clock = snapshot.time_model_state.clocks.get(clock_address) + if clock is None: + raise ValueError(f"autonomous participant clock {clock_address!r} has no runtime state") + return clock.coordinate.tick + + +def _state_identity(state: ParticipantAutonomousExecutionStateModel) -> tuple[object, ...]: + return ( + state.policy_address, + state.policy_digest, + state.participant_address, + state.episode_id, + state.participant_implementation_ref, + state.clock_address, + state.time_segment, + state.profile, + state.random_control_id, + state.random_profile_id, + state.random_namespace, + ) + + +def _ensure_participant_episode( + participant_runtime: object, + snapshot: RuntimeSnapshot, + participant_address: str, +) -> ApplyResult: + if participant_address in snapshot.participant_episode_results: + return ApplyResult(success=True, snapshot=snapshot) + return participant_runtime.initialize( + ParticipantEpisodeInitializeRequest( + participant_address=participant_address, + episode_id=f"{participant_address}-autonomous-0", + ), + snapshot, + ) + + +def _activity_control_unbound_result( + policy: ParticipantAutonomousExecutionRuntime, + snapshot: RuntimeSnapshot, +) -> ApplyResult: + return ApplyResult( + success=False, + snapshot=snapshot, + diagnostics=[ + Diagnostic( + code="runtime.participant-activity-control-unbound", + domain="participant", + address=policy.address, + message=( + f"Participant activity policy requires admitted stochastic control " + f"{policy.stochastic_control_ref!r}." + ), + ) + ], + ) + + +def _initial_activity_schedule( + policy: ParticipantAutonomousExecutionRuntime, + time_model: CompiledTimeModel, + snapshot: RuntimeSnapshot, + participant_address: str, + segment: int, + activity_control: ParticipantActivityRandomControl | None, +) -> tuple[int | None, int, str]: + if activity_control is None: + first_tick, _ = cadence(policy, time_model) + return first_tick, 1, "cadence" + current_tick = clock_tick(snapshot, policy.clock_address) + burst_size = draw_activity_integer( + ParticipantActivityDrawContext( + policy=policy, + participant_address=participant_address, + time_segment=segment, + occurrence_ordinal=0, + control=activity_control, + ), + local_coordinate=2, + minimum=1, + maximum=policy.max_burst_size, + ) + timing = next_activity_timing( + policy=policy, + time_model=time_model, + participant_address=participant_address, + time_segment=segment, + occurrence_ordinal=0, + current_tick=current_tick, + control=activity_control, + ) + return timing.tick, burst_size, timing.disposition + + +def _initial_participant_state( + policy: ParticipantAutonomousExecutionRuntime, + time_model: CompiledTimeModel, + snapshot: RuntimeSnapshot, + participant_address: str, + segment: int, + activity_control: ParticipantActivityRandomControl | None, +) -> ParticipantAutonomousExecutionStateModel: + first_tick, burst_size, timing_disposition = _initial_activity_schedule( + policy, + time_model, + snapshot, + participant_address, + segment, + activity_control, + ) + return ParticipantAutonomousExecutionStateModel( + policy_address=policy.address, + policy_digest=_policy_digest(policy, time_model), + participant_address=participant_address, + episode_id=snapshot.participant_episode_results[participant_address]["episode_id"], + participant_implementation_ref=policy.participant_implementation_ref, + clock_address=policy.clock_address, + time_segment=segment, + lifecycle_state="running" if first_tick is not None else "completed", + next_tick=first_tick if first_tick is not None else clock_tick(snapshot, policy.clock_address), + next_action_index=0, + attempted_actions=0, + succeeded_actions=0, + failed_actions=0, + profile=policy.profile, + random_control_id=activity_control.control_id if activity_control is not None else None, + random_profile_id=activity_control.profile_id if activity_control is not None else None, + random_namespace=activity_control.namespace if activity_control is not None else None, + burst_size=burst_size, + next_timing_disposition=timing_disposition, + ) + + +def _persist_initial_participant_state( + policy: ParticipantAutonomousExecutionRuntime, + snapshot: RuntimeSnapshot, + participant_address: str, + expected: ParticipantAutonomousExecutionStateModel, + changed: list[str], +) -> ApplyResult: + key = f"{policy.address}.state.{participant_address}" + states = dict(snapshot.participant_autonomous_execution_states) + if key in states and _state_identity(ParticipantAutonomousExecutionStateModel.model_validate(states[key])) != ( + _state_identity(expected) + ): + return ApplyResult( + success=False, + snapshot=snapshot, + diagnostics=[ + Diagnostic( + code="runtime.participant-autonomous-state-conflict", + domain="participant", + address=policy.address, + message="Existing autonomous participant state does not match the compiled policy.", + ) + ], + ) + working = snapshot + if key not in states: + states[key] = expected.model_dump(mode="json") + working = snapshot.with_entries( + dict(snapshot.entries), + participant_autonomous_execution_states=states, + ) + changed.append(key) + return ApplyResult(success=True, snapshot=working, changed_addresses=changed) + + +def initialize_participant( + policy: ParticipantAutonomousExecutionRuntime, + time_model: CompiledTimeModel, + participant_runtime: object, + snapshot: RuntimeSnapshot, + participant_address: str, + activity_controls: dict[str, ParticipantActivityRandomControl], +) -> ApplyResult: + episode_result = _ensure_participant_episode(participant_runtime, snapshot, participant_address) + if not episode_result.success: + return episode_result + working = episode_result.snapshot + changed = list(episode_result.changed_addresses) + segment, _ = clock_coordinate(working, policy.clock_address) + activity_control = activity_control_for(policy, activity_controls) + if ( + policy.profile + in { + "participant-autonomous-execution/v2", + _RESOURCE_GOVERNED_PROFILE, + } + and activity_control is None + ): + return _activity_control_unbound_result(policy, working) + expected = _initial_participant_state( + policy, + time_model, + working, + participant_address, + segment, + activity_control, + ) + return _persist_initial_participant_state(policy, working, participant_address, expected, changed) + + +__all__ = ["clock_tick", "initialize_participant"] diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_operations.py b/implementations/python/packages/raes_runtime/participant_scheduler_operations.py index f6eaee236..aade1d2de 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_operations.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_operations.py @@ -2,7 +2,7 @@ from __future__ import annotations -from dataclasses import dataclass, replace +from dataclasses import replace from typing import cast from raes_contracts.contracts import ( @@ -17,10 +17,8 @@ from .participant_action_validation import autonomous_action_result_violation from .participant_activity import ( - ParticipantActivityDrawContext, ParticipantActivityRandomControl, activity_control_for, - draw_activity_integer, next_activity_timing, select_activity_candidate, ) @@ -31,6 +29,9 @@ persist_activity_state, ) from .participant_scheduler_concurrency import participant_generation_commit_diagnostic, run_policy_due_concurrently +from .participant_scheduler_activity_state import ( + next_activity_occurrence_state as _next_activity_occurrence_state, +) from .participant_scheduler_resources import ( commit_activity_resources, measurement_requirements, @@ -223,202 +224,6 @@ def _run_one_due_action( return next_state -def _activity_attempt_is_retryable( - context: _DueActionContext, - state: ParticipantAutonomousExecutionStateModel, - *, - action_succeeded: bool, - failure_class: str | None, - protocol_failure: bool, - attempted: int, -) -> bool: - index = state.next_action_index - return ( - not protocol_failure - and not action_succeeded - and failure_class in context.policy.action_candidate_retry_failure_classes[index] - and state.current_retry < context.policy.action_candidate_max_retries[index] - and attempted < context.policy.max_action_attempts - ) - - -@dataclass(frozen=True) -class _ActivityProgress: - candidate_id: str - completed: list[str] - cooldowns: dict[str, int] - occurrence: int - lifecycle: str - - -def _completed_activity_progress( - context: _DueActionContext, - state: ParticipantAutonomousExecutionStateModel, - *, - action_succeeded: bool, - protocol_failure: bool, - attempted: int, -) -> _ActivityProgress: - policy = context.policy - index = state.next_action_index - candidate_id = policy.action_candidate_ids[index] - completed = list(state.completed_candidate_ids) - if action_succeeded and candidate_id not in completed: - completed.append(candidate_id) - cooldowns = dict(state.candidate_cooldown_until) - cooldowns[candidate_id] = context.current_tick + policy.action_candidate_cooldown_ticks[index] - occurrence = state.occurrence_ordinal + 1 - lifecycle = state.lifecycle_state - if protocol_failure or (not action_succeeded and policy.failure_policy == "stop"): - lifecycle = "failed" - elif occurrence >= policy.max_occurrences or attempted >= policy.max_action_attempts: - lifecycle = "completed" - return _ActivityProgress( - candidate_id=candidate_id, - completed=completed, - cooldowns=cooldowns, - occurrence=occurrence, - lifecycle=lifecycle, - ) - - -@dataclass(frozen=True) -class _ActivitySchedule: - lifecycle: str - next_tick: int - burst_position: int - burst_size: int - timing_disposition: str - - -def _next_activity_schedule( - context: _DueActionContext, - state: ParticipantAutonomousExecutionStateModel, - control: ParticipantActivityRandomControl, - progress: _ActivityProgress, -) -> _ActivitySchedule: - lifecycle = progress.lifecycle - burst_position = state.burst_position - burst_size = state.burst_size - next_tick = context.current_tick - timing_disposition = state.next_timing_disposition - if lifecycle == "running" and burst_position + 1 < burst_size: - burst_position += 1 - elif lifecycle == "running": - burst_position = 0 - burst_size = draw_activity_integer( - ParticipantActivityDrawContext( - policy=context.policy, - participant_address=context.participant_address, - time_segment=state.time_segment, - occurrence_ordinal=progress.occurrence, - control=control, - ), - local_coordinate=2, - minimum=1, - maximum=context.policy.max_burst_size, - ) - timing = next_activity_timing( - policy=context.policy, - time_model=context.time_model, - participant_address=context.participant_address, - time_segment=state.time_segment, - occurrence_ordinal=progress.occurrence, - current_tick=context.current_tick, - control=control, - ) - timing_disposition = timing.disposition - if timing.tick is None: - lifecycle = "completed" - else: - next_tick = timing.tick - return _ActivitySchedule( - lifecycle=lifecycle, - next_tick=next_tick, - burst_position=burst_position, - burst_size=burst_size, - timing_disposition=timing_disposition, - ) - - -def _activity_retry_state( - context: _DueActionContext, - state: ParticipantAutonomousExecutionStateModel, - request: ParticipantActionAdmissionRequest, - *, - attempted: int, - failed: int, -) -> ParticipantAutonomousExecutionStateModel: - candidate_id = context.policy.action_candidate_ids[state.next_action_index] - return state.model_copy( - update={ - "next_tick": context.current_tick, - "attempted_actions": attempted, - "failed_actions": failed, - "current_retry": state.current_retry + 1, - "last_candidate_id": candidate_id, - "last_action_instance_id": request.action_instance_id, - } - ) - - -def _next_activity_occurrence_state( - context: _DueActionContext, - state: ParticipantAutonomousExecutionStateModel, - request: ParticipantActionAdmissionRequest, - *, - action_succeeded: bool, - failure_class: str | None, - protocol_failure: bool, -) -> ParticipantAutonomousExecutionStateModel: - control = context.activity_control - if control is None: - raise ValueError("participant activity execution requires a random control") - attempted = state.attempted_actions + 1 - failed = state.failed_actions + (0 if action_succeeded else 1) - if _activity_attempt_is_retryable( - context, - state, - action_succeeded=action_succeeded, - failure_class=failure_class, - protocol_failure=protocol_failure, - attempted=attempted, - ): - return _activity_retry_state( - context, - state, - request, - attempted=attempted, - failed=failed, - ) - progress = _completed_activity_progress( - context, - state, - action_succeeded=action_succeeded, - protocol_failure=protocol_failure, - attempted=attempted, - ) - schedule = _next_activity_schedule(context, state, control, progress) - return state.model_copy( - update={ - "lifecycle_state": schedule.lifecycle, - "next_tick": schedule.next_tick, - "attempted_actions": attempted, - "succeeded_actions": state.succeeded_actions + (1 if action_succeeded else 0), - "failed_actions": failed, - "occurrence_ordinal": progress.occurrence, - "current_retry": 0, - "burst_position": schedule.burst_position, - "burst_size": schedule.burst_size, - "last_candidate_id": progress.candidate_id, - "completed_candidate_ids": progress.completed, - "candidate_cooldown_until": progress.cooldowns, - "last_action_instance_id": request.action_instance_id, - "next_timing_disposition": schedule.timing_disposition, - } - ) - - def _run_one_activity_action( context: _DueActionContext, state: ParticipantAutonomousExecutionStateModel, From e912c77681fca8b4c30881f7b9b9ef8eb8f6f896 Mon Sep 17 00:00:00 2001 From: Brad Edwards Date: Mon, 27 Jul 2026 08:12:57 +0200 Subject: [PATCH 3/5] style: format scheduler module --- .../python/packages/raes_runtime/participant_scheduler.py | 1 + 1 file changed, 1 insertion(+) diff --git a/implementations/python/packages/raes_runtime/participant_scheduler.py b/implementations/python/packages/raes_runtime/participant_scheduler.py index 0f6a68333..9c6dc3168 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler.py @@ -32,6 +32,7 @@ from .participant_scheduler_time import cadence as _cadence from .participant_scheduler_time import clock_coordinate + def _missing_execution_service_result( policy: ParticipantAutonomousExecutionRuntime, run: SchedulerRunState, From 848de13dc3385239299e8200306989657c0a55ca Mon Sep 17 00:00:00 2001 From: Brad Edwards Date: Mon, 27 Jul 2026 08:15:17 +0200 Subject: [PATCH 4/5] fix: correct scheduler lint errors --- .../python/packages/raes_runtime/participant_scheduler.py | 6 +++++- .../raes_runtime/participant_scheduler_operations.py | 2 +- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/implementations/python/packages/raes_runtime/participant_scheduler.py b/implementations/python/packages/raes_runtime/participant_scheduler.py index 9c6dc3168..2e4cff1b4 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler.py @@ -17,11 +17,13 @@ set_execution_clock_lifecycle, ) from .participant_resource_budgets import initialize_participant_resource_budgets -from .participant_scheduler_lifecycle import reset_policy_at_clock from .participant_scheduler_initialization import ( clock_tick as _clock_tick, +) +from .participant_scheduler_initialization import ( initialize_participant as _initialize_participant, ) +from .participant_scheduler_lifecycle import reset_policy_at_clock from .participant_scheduler_operations import ( SchedulerRunState, participant_due_context, @@ -32,6 +34,8 @@ from .participant_scheduler_time import cadence as _cadence from .participant_scheduler_time import clock_coordinate +_RESOURCE_GOVERNED_PROFILE = "participant-autonomous-execution/v3" + def _missing_execution_service_result( policy: ParticipantAutonomousExecutionRuntime, diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_operations.py b/implementations/python/packages/raes_runtime/participant_scheduler_operations.py index aade1d2de..67ab20a6d 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_operations.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_operations.py @@ -28,10 +28,10 @@ annotate_activity_history, persist_activity_state, ) -from .participant_scheduler_concurrency import participant_generation_commit_diagnostic, run_policy_due_concurrently from .participant_scheduler_activity_state import ( next_activity_occurrence_state as _next_activity_occurrence_state, ) +from .participant_scheduler_concurrency import participant_generation_commit_diagnostic, run_policy_due_concurrently from .participant_scheduler_resources import ( commit_activity_resources, measurement_requirements, From 2f7151702e0ab2446cb4ffdbfc477b9602fc040c Mon Sep 17 00:00:00 2001 From: Brad Edwards Date: Mon, 27 Jul 2026 08:22:51 +0200 Subject: [PATCH 5/5] fix: preserve activity draw call semantics --- .../python/packages/raes_runtime/participant_activity.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/implementations/python/packages/raes_runtime/participant_activity.py b/implementations/python/packages/raes_runtime/participant_activity.py index aeef0ab49..e97d9a4dd 100644 --- a/implementations/python/packages/raes_runtime/participant_activity.py +++ b/implementations/python/packages/raes_runtime/participant_activity.py @@ -197,7 +197,7 @@ def next_activity_timing( minimum = _timing_units(policy.timing_minimum_ticks, step_ticks) maximum = _timing_units(policy.timing_maximum_ticks, step_ticks) interval_units = draw_activity_integer( - ParticipantActivityDrawContext( + context=ParticipantActivityDrawContext( policy=policy, participant_address=participant_address, time_segment=time_segment, @@ -295,7 +295,7 @@ def select_activity_candidate( return None total = sum(policy.action_candidate_weights[index] for index in eligible_indices) selected = draw_activity_integer( - ParticipantActivityDrawContext( + context=ParticipantActivityDrawContext( policy=policy, participant_address=participant_address, time_segment=time_segment,