diff --git a/docs/decisions/issue-1101-concurrent-participant-rollback-preflight.md b/docs/decisions/issue-1101-concurrent-participant-rollback-preflight.md new file mode 100644 index 00000000..eee7a20d --- /dev/null +++ b/docs/decisions/issue-1101-concurrent-participant-rollback-preflight.md @@ -0,0 +1,79 @@ +# Issue 1101 Concurrent Participant Rollback Preflight + +Date: 2026-08-11 + +Issue: #1101. Requirement: RUN-308. + +## Decision + +A concurrent dispatch and its serialized commit have two distinct failure +boundaries. Before native dispatch, the scheduler retains a deeply isolated +pre-batch `RuntimeSnapshot`; binding, reservation, or dispatch-copy failure can +restore that snapshot because no native action was submitted. Once the backend +batch method is entered, an exception, cancellation, or result collection that +cannot be paired with the submitted requests is indeterminate native work. The +scheduler settles every submitted action id as a failed, non-retryable attempt +and releases only that batch's accounting delta. It never restores those action +ids to a due state that could replay an unknown side effect. + +Once results can be paired, each result envelope is deeply detached from the +backend and every dispatched peer is settled. Each worker receives an isolated +predecessor. Valid peer snapshots are revision-checked and committed in +deterministic request order. The merge classifies every `RuntimeSnapshot` field: +all backend-owned mapping and value fields use three-way merge, while autonomous +scheduler state and execution-service state are protected and must equal the +reserved predecessor. This exhaustive ownership guard makes a new snapshot +field fail at import/test time until it receives an explicit owner. Committed +nested values are detached so a retained backend result cannot mutate +authoritative state after return. + +An individually invalid result never contributes its backend snapshot; its own +scheduler occurrence transitions to protocol failure while valid peers remain +committed. A normal failed outcome under `failure_policy: stop` is recorded, but +stop takes effect only after all peers in the already-dispatched chunk are +settled. Later chunks are not dispatched. A merge conflict rejects the +conflicting peer with a stable diagnostic rather than leaking a backend key or +discarding peers committed earlier in the serialized order. + +Service accounting is delta based. Admission adds only the chunk's in-flight +count, completion subtracts the same count, and pre-existing reserved or +in-flight work remains authoritative. If normal final settlement raises, the +portable boundary records a stable failure and restores the isolated pre-batch +service counters rather than leaking the exception or leaving the batch live. +Already in-flight participants are not selected again. Due work is scanned once +and processed by an iterative, capacity-bounded chunk loop; participant count +therefore cannot consume Python call-stack depth. + +The backend call remains a trust boundary. Portable diagnostics contain stable +codes and fixed messages, never native exception text or type, traceback, host +path, mapping key, credential, or participant data. This change does not claim +that native side effects can be rolled back. It records an indeterminate +dispatch as non-retryable precisely because rollback cannot be proved. + +## Rejected Alternatives + +- Treat the whole paired batch as all-or-nothing: native peers have already run, + so discarding valid results makes portable state diverge from observed work. +- Restore scheduler attempts after entering backend dispatch: transport failure + cannot prove absence of a native side effect, and restoring the stable action + id permits blind replay. +- Stop committing at the first failed peer: later peers in the same chunk have + also run and must be settled before stop affects undispatched work. +- Clear all service counters: this releases reservations owned by other work. +- Retain a shallow rollback alias: nested backend mutation can corrupt the + supposed pre-batch snapshot even when dispatch raises. +- Recompute all due work recursively after each chunk: this is quadratic and + fails at ordinary Python recursion limits. + +## Verification + +Tests cover pre-dispatch rollback; raised, cancelled, and miscounted dispatched +backends; non-retryable action identity; per-worker and post-result mutation +isolation; exhaustive snapshot-field ownership; serial/concurrent projection +equivalence; protocol-invalid typed and untyped results; public failure +snapshots; delta-preserved service accounting; revision-checked metadata; mixed +success/failure peers under stop; rejected changed-address isolation; normalized +service-settlement exceptions; stable diagnostics; explicit capacity backpressure; +and an 800-participant iterative run through real reservation and settlement with +one due scan. The focused participant scheduler suite, runtime suite, lint, +policy, and canonical verification remain required. diff --git a/docs/requirements/RUN-308/requirement.md b/docs/requirements/RUN-308/requirement.md index b380e61e..7e397a88 100644 --- a/docs/requirements/RUN-308/requirement.md +++ b/docs/requirements/RUN-308/requirement.md @@ -29,3 +29,9 @@ Requirement inventory expansion. Multi-participant experiments require runtime s - IMPLEMENTS → SPEC `contracts/schemas/participant-runtime/participant-time-management-context-v1.json` (Participant time-management context schema v1) - TESTS → TEST `implementations/python/tests/test_run_308_concurrent_participant_execution.py` (RUN-308 concurrent participant execution contract tests) - TESTS → TEST `implementations/python/tests/test_participant_backend_contracts.py` (Participant backend runtime contract regression tests) +- DOCUMENTS → GITHUB_ISSUE `1101` (Transactional rollback for concurrent participant reservations) +- DOCUMENTS → DOCUMENTATION `docs/decisions/issue-1101-concurrent-participant-rollback-preflight.md` (Concurrent batch rollback architecture and nonclaims) +- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py` (Batch-scoped reservation and snapshot rollback) +- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/participant_scheduler_concurrent_state.py` (Revision-safe snapshot merge and delta service accounting) +- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/participant_scheduler_concurrent_settlement.py` (Pre-dispatch rollback and post-dispatch indeterminate settlement) +- TESTS → TEST `implementations/python/tests/test_participant_concurrent_batch_reservations.py` (Concurrent reservation failure and rollback regressions) diff --git a/implementations/python/packages/raes_backend_protocols/participant_execution_runtime.py b/implementations/python/packages/raes_backend_protocols/participant_execution_runtime.py index d72b88eb..4b51e026 100644 --- a/implementations/python/packages/raes_backend_protocols/participant_execution_runtime.py +++ b/implementations/python/packages/raes_backend_protocols/participant_execution_runtime.py @@ -3,6 +3,7 @@ from __future__ import annotations from concurrent.futures import ThreadPoolExecutor +from copy import deepcopy from raes_contracts.contracts.participant_execution import ( ParticipantExecutionServiceStateModel, @@ -108,9 +109,17 @@ def admit_actions_concurrently( raise ValueError("concurrent participant execution requires at least two workers") if len(requests) > max_workers: raise ValueError("participant action batch exceeds its worker bound") + # RuntimeSnapshot is a mutable carrier even though workers are required + # to treat their predecessor as immutable. Give each native call an + # independently owned copy so an in-place adapter bug cannot race with + # or contaminate a peer's input. + predecessors = tuple(deepcopy(snapshot) for _request in requests) with ThreadPoolExecutor( max_workers=max_workers, thread_name_prefix="raes-participant", ) as executor: - futures = [executor.submit(self.admit_action, request, snapshot) for request in requests] + futures = [ + executor.submit(self.admit_action, request, predecessor) + for request, predecessor in zip(requests, predecessors, strict=True) + ] return tuple(future.result() for future in futures) diff --git a/implementations/python/packages/raes_runtime/participant_action_validation.py b/implementations/python/packages/raes_runtime/participant_action_validation.py index 26df050d..c231f08d 100644 --- a/implementations/python/packages/raes_runtime/participant_action_validation.py +++ b/implementations/python/packages/raes_runtime/participant_action_validation.py @@ -13,6 +13,10 @@ from raes_contracts.runtime_state import RuntimeSnapshot _TERMINAL_ACTION_STATUSES = frozenset({"succeeded", "failed", "partial_success", "rejected", "withheld"}) +_PROTECTED_SCHEDULER_SNAPSHOT_FIELDS = ( + "participant_autonomous_execution_states", + "participant_execution_services", +) def autonomous_action_result_violation( @@ -25,6 +29,9 @@ def autonomous_action_result_violation( """Return why a native result cannot be committed, or ``None``.""" violation = _native_result_violation(result, episode_id) + if violation is None: + assert isinstance(result, ParticipantActionApplyResult) + violation = _protected_scheduler_state_violation(result, predecessor) if violation is None: assert isinstance(result, ParticipantActionApplyResult) action_result = result.action_result @@ -40,6 +47,18 @@ def autonomous_action_result_violation( return violation +def _protected_scheduler_state_violation( + result: ParticipantActionApplyResult, + predecessor: RuntimeSnapshot, +) -> str | None: + """Keep scheduler reservations and counters under serialized ownership.""" + + for field_name in _PROTECTED_SCHEDULER_SNAPSHOT_FIELDS: + if getattr(result.snapshot, field_name) != getattr(predecessor, field_name): + return f"participant runtime changed scheduler-owned snapshot field {field_name}" + return None + + def _bound_request_result( request: ParticipantActionAdmissionRequest, action_result: object, diff --git a/implementations/python/packages/raes_runtime/participant_clock_driver.py b/implementations/python/packages/raes_runtime/participant_clock_driver.py index 85407be2..66a81bf2 100644 --- a/implementations/python/packages/raes_runtime/participant_clock_driver.py +++ b/implementations/python/packages/raes_runtime/participant_clock_driver.py @@ -238,7 +238,11 @@ def _next_participant_tick(snapshot: RuntimeSnapshot, clock_address: str) -> int next_ticks = [ int(payload["next_tick"]) for payload in snapshot.participant_autonomous_execution_states.values() - if payload.get("clock_address") == clock_address and payload.get("lifecycle_state") == "running" + if ( + payload.get("clock_address") == clock_address + and payload.get("lifecycle_state") == "running" + and int(payload.get("in_flight", 0)) == 0 + ) ] return min(next_ticks) if next_ticks else None diff --git a/implementations/python/packages/raes_runtime/participant_scheduler.py b/implementations/python/packages/raes_runtime/participant_scheduler.py index 2e4cff1b..d0561c2d 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler.py @@ -4,6 +4,7 @@ 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.runtime_state import ApplyResult, RuntimeSnapshot @@ -32,7 +33,7 @@ ) from .participant_scheduler_policy import _policy_digest from .participant_scheduler_time import cadence as _cadence -from .participant_scheduler_time import clock_coordinate +from .participant_scheduler_time import cadence_missed_result, clock_coordinate _RESOURCE_GOVERNED_PROFILE = "participant-autonomous-execution/v3" @@ -59,6 +60,46 @@ def _execution_service_accepts_work(service: ParticipantExecutionServiceStateMod return service.observed_lifecycle == "running" and service.accepting_new_work and service.readiness == "ready" +def _execution_service_has_capacity(service: ParticipantExecutionServiceStateModel) -> bool: + return service.reserved + service.in_flight < service.capacity + + +def _capacity_blocked_due_result( + policy: ParticipantAutonomousExecutionRuntime, + current_tick: int, + run: SchedulerRunState, +) -> ApplyResult | None: + for participant_address in policy.participant_addresses: + 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: + return cadence_missed_result(run.working, key, current_tick, state) + due = ( + state.lifecycle_state == "running" + and state.next_tick == current_tick + and state.attempted_actions < policy.max_action_attempts + and state.in_flight == 0 + ) + if due: + return ApplyResult( + success=False, + snapshot=run.working, + diagnostics=[ + Diagnostic( + code="runtime.participant-execution-capacity-blocked", + domain="participant", + address=policy.address, + message=( + "Due participant work could not progress because execution-service capacity is exhausted." + ), + ) + ], + ) + return None + + def _run_serial_due( policy: ParticipantAutonomousExecutionRuntime, time_model: CompiledTimeModel, @@ -101,6 +142,9 @@ def _run_due_policy( return cadence_ticks = _cadence(policy, time_model)[1] if policy.profile == "participant-autonomous-execution/v1" else 0 current_tick = _clock_tick(run.working, policy.clock_address) + if not _execution_service_has_capacity(service): + run.failure = _capacity_blocked_due_result(policy, current_tick, run) + return if not run_policy_due_concurrently(policy, time_model, participant_runtime, current_tick, cadence_ticks, run): _run_serial_due(policy, time_model, participant_runtime, current_tick, cadence_ticks, activity_controls, run) diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py index 901cac39..ec3b901c 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py @@ -2,160 +2,29 @@ from __future__ import annotations -from dataclasses import dataclass +from asyncio import CancelledError +from copy import deepcopy from typing import TYPE_CHECKING from raes_contracts.contracts import ParticipantAutonomousExecutionStateModel -from raes_contracts.contracts.participant_execution import ParticipantExecutionServiceStateModel from raes_contracts.diagnostics import Diagnostic -from raes_contracts.participant_binding import ParticipantActionAdmissionRequest, ParticipantActionApplyResult -from raes_contracts.runtime_state import ApplyResult, RuntimeSnapshot from raes_processor.models import CompiledTimeModel, ParticipantAutonomousExecutionRuntime -from .participant_action_validation import autonomous_action_result_violation +from .participant_scheduler_concurrent_dispatch import ( + _CONCURRENT_SNAPSHOT_ISOLATION_FAILED, + _ConcurrentBatch, + _execute_concurrent_batch, +) +from .participant_scheduler_concurrent_settlement import _set_concurrent_failure +from .participant_scheduler_concurrent_state import ( + _available_concurrent_capacity, + _materialize_concurrent_snapshot, +) if TYPE_CHECKING: from .participant_scheduler_types import SchedulerRunState, _DueActionContext -def _changed_mapping( - base: dict[str, object], - incoming: dict[str, object], -) -> dict[str, object]: - return {key: value for key, value in incoming.items() if base.get(key) != value} - - -def _merge_mapping_revision_checked( - *, - base: dict[str, object], - current: dict[str, object], - incoming: dict[str, object], - field_name: str, -) -> dict[str, object]: - merged = dict(current) - for key, value in _changed_mapping(base, incoming).items(): - current_value = current.get(key) - base_value = base.get(key) - if current_value != base_value and current_value != value: - raise ValueError(f"concurrent participant commit conflict in {field_name}[{key!r}]") - merged[key] = value - return merged - - -def _merge_concurrent_action_snapshot( - base: RuntimeSnapshot, - current: RuntimeSnapshot, - incoming: RuntimeSnapshot, -) -> RuntimeSnapshot: - """Merge one native result without replacing a newer whole snapshot.""" - - entries = _merge_mapping_revision_checked( - base=dict(base.entries), - current=dict(current.entries), - incoming=dict(incoming.entries), - field_name="entries", - ) - mapping_fields = ( - "participant_episode_results", - "participant_episode_history", - "participant_behavior_history", - "participant_control_history", - "shared_state_records", - "shared_state_history", - "joint_action_records", - "time_management_contexts", - ) - updates: dict[str, object] = {} - for field_name in mapping_fields: - updates[field_name] = _merge_mapping_revision_checked( - base=dict(getattr(base, field_name)), - current=dict(getattr(current, field_name)), - incoming=dict(getattr(incoming, field_name)), - field_name=field_name, - ) - metadata = dict(current.metadata) - metadata.update(incoming.metadata) - updates["metadata"] = metadata - return current.with_entries(entries, **updates) - - -def participant_generation_commit_diagnostic( - request: ParticipantActionAdmissionRequest, - authoritative: RuntimeSnapshot, -) -> Diagnostic | None: - """Fence native completion against the serialized commit owner's state.""" - - scope = request.execution_scope_ref - if scope is None: - return None - payload = authoritative.participant_execution_services.get(scope) - expected = request.execution_generation - if payload is not None: - service = ParticipantExecutionServiceStateModel.model_validate(payload) - if service.generation == expected and service.observed_generation == expected: - return None - return Diagnostic( - code="runtime.participant-execution-stale-completion", - domain="participant", - address=request.participant_address, - message=( - "Participant action completion was rejected because the authoritative serialized commit generation changed." - ), - ) - - -def _reserve_concurrent_actions( - run: SchedulerRunState, - contexts: tuple[_DueActionContext, ...], -) -> None: - states = dict(run.working.participant_autonomous_execution_states) - for context in contexts: - state = ParticipantAutonomousExecutionStateModel.model_validate(states[context.key]) - states[context.key] = state.model_copy( - update={ - "attempted_actions": state.attempted_actions + 1, - "in_flight": state.in_flight + 1, - } - ).model_dump(mode="json") - services = dict(run.working.participant_execution_services) - for policy_address in {context.policy.address for context in contexts}: - payload = services.get(policy_address) - if payload is None: - continue - service = ParticipantExecutionServiceStateModel.model_validate(payload) - count = sum(1 for context in contexts if context.policy.address == policy_address) - services[policy_address] = service.model_copy( - update={ - "reserved": 0, - "in_flight": count, - "quiescent": False, - } - ).model_dump(mode="json") - run.working = run.working.with_entries( - dict(run.working.entries), - participant_autonomous_execution_states=states, - participant_execution_services=services, - ) - - -def _finish_concurrent_service_state( - run: SchedulerRunState, - policy_address: str, -) -> None: - services = dict(run.working.participant_execution_services) - payload = services.get(policy_address) - if payload is None: - return - service = ParticipantExecutionServiceStateModel.model_validate(payload) - services[policy_address] = service.model_copy(update={"reserved": 0, "in_flight": 0, "quiescent": True}).model_dump( - mode="json" - ) - run.working = run.working.with_entries( - dict(run.working.entries), - participant_execution_services=services, - ) - - def _due_contexts( policy: ParticipantAutonomousExecutionRuntime, time_model: CompiledTimeModel, @@ -181,6 +50,7 @@ def _due_contexts( state.lifecycle_state == "running" and state.next_tick == current_tick and state.attempted_actions < policy.max_action_attempts + and state.in_flight == 0 ) if due: contexts.append( @@ -198,174 +68,107 @@ def _due_contexts( return contexts, states -def _set_concurrent_failure(run: SchedulerRunState, diagnostic: Diagnostic) -> None: - run.diagnostics.append(diagnostic) - run.failure = ApplyResult( - success=False, - snapshot=run.working, - diagnostics=run.diagnostics, - changed_addresses=list(dict.fromkeys(run.changed)), - ) - - -def _unsupported_concurrency_failure(policy: ParticipantAutonomousExecutionRuntime, run: SchedulerRunState) -> None: - run.failure = ApplyResult( - success=False, - snapshot=run.working, - diagnostics=[ - Diagnostic( - code="runtime.participant-concurrency-unsupported", - domain="participant", - address=policy.address, - message="Backend declared bounded participant concurrency without an executable batch method.", - ) - ], - ) - - -def _commit_concurrent_result( - context: _DueActionContext, - state: ParticipantAutonomousExecutionStateModel, - request: ParticipantActionAdmissionRequest, - result: ParticipantActionApplyResult, - base: RuntimeSnapshot, +def _isolate_concurrent_policy_snapshot( + policy: ParticipantAutonomousExecutionRuntime, run: SchedulerRunState, -) -> None: - from .participant_scheduler_operations import _next_action_state - - stale_completion = participant_generation_commit_diagnostic(request, run.working) - if stale_completion is not None: - _set_concurrent_failure(run, stale_completion) - return +) -> bool: + isolated = True try: - run.working = _merge_concurrent_action_snapshot(base, run.working, result.snapshot) - except ValueError as exc: + run.working = deepcopy(run.working) + except (Exception, CancelledError): # NOSONAR - no native action has been submitted _set_concurrent_failure( run, Diagnostic( - code="runtime.participant-concurrent-commit-conflict", + code=_CONCURRENT_SNAPSHOT_ISOLATION_FAILED, domain="participant", - address=context.key, - message=str(exc), + address=policy.address, + message="Concurrent participant snapshot isolation did not complete before dispatch.", ), ) - return - protocol_violation = autonomous_action_result_violation( - request, - result, - episode_id=state.episode_id, - predecessor=base, - ) - if protocol_violation is not None: - _set_concurrent_failure( - run, - Diagnostic( - code="runtime.participant-autonomous-action-protocol-invalid", - domain="participant", - address=context.participant_address, - message=protocol_violation, - ), + isolated = False + return isolated + + +def _execute_capacity_bounded_batches( + policy: ParticipantAutonomousExecutionRuntime, + time_model: CompiledTimeModel, + participant_runtime: object, + current_tick: int, + cadence_ticks: int, + run: SchedulerRunState, + contexts: list[_DueActionContext], +) -> int: + offset = 0 + while len(contexts) - offset >= 2 and run.failure is None: + available = _available_concurrent_capacity(policy, run) + if available == 0: + _set_concurrent_failure( + run, + Diagnostic( + code="runtime.participant-execution-capacity-blocked", + domain="participant", + address=policy.address, + message="Due participant work could not progress because execution-service capacity is exhausted.", + ), + ) + break + if available < 2: + break + batch_size = min(available, len(contexts) - offset) + selected_contexts = tuple(contexts[offset : offset + batch_size]) + selected_states = tuple( + ParticipantAutonomousExecutionStateModel.model_validate( + run.working.participant_autonomous_execution_states[context.key] + ) + for context in selected_contexts ) - return - action_result = result.action_result - action_succeeded = bool(result.success and action_result is not None and action_result.status == "succeeded") - next_state = _next_action_state( - context, - state, - request, - action_succeeded=action_succeeded, - protocol_failure=False, - ).model_copy(update={"in_flight": 0}) - scheduler_states = dict(run.working.participant_autonomous_execution_states) - scheduler_states[context.key] = next_state.model_dump(mode="json") - run.working = run.working.with_entries( - dict(run.working.entries), - participant_autonomous_execution_states=scheduler_states, - ) - run.diagnostics.extend(result.diagnostics) - run.changed.extend([*result.changed_addresses, context.key]) - if not action_succeeded and context.policy.failure_policy == "stop": - run.failure = ApplyResult( - success=False, - snapshot=run.working, - diagnostics=run.diagnostics, - changed_addresses=list(dict.fromkeys(run.changed)), + _execute_concurrent_batch( + _ConcurrentBatch( + policy=policy, + time_model=time_model, + participant_runtime=participant_runtime, + current_tick=current_tick, + cadence_ticks=cadence_ticks, + run=run, + contexts=selected_contexts, + states=selected_states, + pre_batch=run.working, + materialize=False, + ) ) + offset += batch_size + return offset -def _finish_due_policy( +def _execute_concurrent_due_contexts( policy: ParticipantAutonomousExecutionRuntime, time_model: CompiledTimeModel, participant_runtime: object, current_tick: int, cadence_ticks: int, run: SchedulerRunState, + contexts: list[_DueActionContext], ) -> None: - from .participant_scheduler_operations import participant_due_context, run_participant_due - - if run.failure is not None: - return - if run_policy_due_concurrently(policy, time_model, participant_runtime, current_tick, cadence_ticks, run): + if not _isolate_concurrent_policy_snapshot(policy, run): return - for participant_address in policy.participant_addresses: - run_participant_due( - participant_due_context( - policy, - time_model, - participant_runtime, - participant_address, - current_tick, - cadence_ticks, - ), - run, - ) - if run.failure is not None: - break - - -@dataclass(frozen=True) -class _ConcurrentBatch: - policy: ParticipantAutonomousExecutionRuntime - time_model: CompiledTimeModel - participant_runtime: object - current_tick: int - cadence_ticks: int - run: SchedulerRunState - contexts: list[_DueActionContext] - states: list[ParticipantAutonomousExecutionStateModel] - + offset = _execute_capacity_bounded_batches( + policy, + time_model, + participant_runtime, + current_tick, + cadence_ticks, + run, + contexts, + ) -def _execute_concurrent_batch(batch: _ConcurrentBatch) -> None: - batch_method = getattr(batch.participant_runtime, "admit_actions_concurrently", None) - if not callable(batch_method): - _unsupported_concurrency_failure(batch.policy, batch.run) - return - selected_contexts = tuple(batch.contexts[: batch.policy.max_in_flight]) - selected_states = batch.states[: batch.policy.max_in_flight] - from .participant_scheduler_operations import _bound_action_request + if run.failure is None: + run.working = _materialize_concurrent_snapshot(run.working) + from .participant_scheduler_operations import run_participant_due - requests = tuple( - _bound_action_request(context, batch.run.working, state) - for context, state in zip(selected_contexts, selected_states, strict=True) - ) - _reserve_concurrent_actions(batch.run, selected_contexts) - base = batch.run.working - results = batch_method(requests, base, len(requests)) - if len(results) != len(requests): - raise ValueError("concurrent participant result count must match requests") - for context, state, request, result in zip(selected_contexts, selected_states, requests, results, strict=True): - _commit_concurrent_result(context, state, request, result, base, batch.run) - if batch.run.failure is not None: - break - _finish_concurrent_service_state(batch.run, batch.policy.address) - _finish_due_policy( - batch.policy, - batch.time_model, - batch.participant_runtime, - batch.current_tick, - batch.cadence_ticks, - batch.run, - ) + for context in contexts[offset:]: + run_participant_due(context, run) + if run.failure is not None: + break def run_policy_due_concurrently( @@ -380,21 +183,16 @@ def run_policy_due_concurrently( if policy.profile != "participant-autonomous-execution/v1": return False - contexts, states = _due_contexts(policy, time_model, participant_runtime, current_tick, cadence_ticks, run) - enough_due_work = len(contexts) >= 2 and policy.max_in_flight >= 2 - if run.failure is None and not enough_due_work: - return False - if run.failure is None: - _execute_concurrent_batch( - _ConcurrentBatch( - policy=policy, - time_model=time_model, - participant_runtime=participant_runtime, - current_tick=current_tick, - cadence_ticks=cadence_ticks, - run=run, - contexts=contexts, - states=states, - ) + contexts, _states = _due_contexts(policy, time_model, participant_runtime, current_tick, cadence_ticks, run) + handled = run.failure is not None or (len(contexts) >= 2 and policy.max_in_flight >= 2) + if handled and run.failure is None: + _execute_concurrent_due_contexts( + policy, + time_model, + participant_runtime, + current_tick, + cadence_ticks, + run, + contexts, ) - return run.failure is not None or enough_due_work + return handled diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_commit.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_commit.py new file mode 100644 index 00000000..75f5da4e --- /dev/null +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_commit.py @@ -0,0 +1,201 @@ +"""Validation and serialized commit for concurrent participant results.""" + +from __future__ import annotations + +from asyncio import CancelledError +from typing import TYPE_CHECKING, cast + +from raes_contracts.addressing import require_compiled_address +from raes_contracts.contracts import ParticipantAutonomousExecutionStateModel +from raes_contracts.contracts.participant_execution import ParticipantExecutionServiceStateModel +from raes_contracts.diagnostics import Diagnostic +from raes_contracts.participant_binding import ParticipantActionAdmissionRequest, ParticipantActionApplyResult +from raes_contracts.runtime_state import RuntimeSnapshot + +from .participant_action_validation import autonomous_action_result_violation +from .participant_scheduler_concurrent_settlement import _settle_failed_concurrent_occurrence +from .participant_scheduler_concurrent_state import ( + _stage_concurrent_action_snapshot, + _with_concurrent_scheduler_updates, +) + +if TYPE_CHECKING: + from .participant_scheduler_types import SchedulerRunState, _DueActionContext + + +def participant_generation_commit_diagnostic( + request: ParticipantActionAdmissionRequest, + authoritative: RuntimeSnapshot, +) -> Diagnostic | None: + """Fence native completion against the serialized commit owner's state.""" + + scope = request.execution_scope_ref + if scope is None: + return None + payload = authoritative.participant_execution_services.get(scope) + expected = request.execution_generation + if payload is not None: + service = ParticipantExecutionServiceStateModel.model_validate(payload) + if service.generation == expected and service.observed_generation == expected: + return None + return Diagnostic( + code="runtime.participant-execution-stale-completion", + domain="participant", + address=request.participant_address, + message=( + "Participant action completion was rejected because the authoritative serialized commit generation changed." + ), + ) + + +def _concurrent_result_protocol_invalid( + request: ParticipantActionAdmissionRequest, + result: object, + *, + episode_id: str, + predecessor: RuntimeSnapshot, +) -> bool: + """Validate one backend result without copying backend-controlled detail.""" + + try: + invalid = _concurrent_result_envelope_invalid(result) + if not invalid: + typed_result = cast(ParticipantActionApplyResult, result) + invalid = ( + autonomous_action_result_violation( + request, + typed_result, + episode_id=episode_id, + predecessor=predecessor, + ) + is not None + ) + except (Exception, CancelledError): # NOSONAR - a malformed backend envelope must fail closed + invalid = True + return invalid + + +def _concurrent_result_envelope_invalid(result: object) -> bool: + """Check the structural result envelope before semantic validation.""" + + if isinstance(result, ParticipantActionApplyResult): + invalid = not all( + ( + isinstance(result.snapshot, RuntimeSnapshot), + type(result.success) is bool, + type(result.diagnostics) is list, + type(result.changed_addresses) is list, + ) + ) + else: + invalid = True + if not invalid: + invalid = any(not isinstance(diagnostic, Diagnostic) for diagnostic in result.diagnostics) + if not invalid: + for address in result.changed_addresses: + require_compiled_address(address, field_name="changed address") + invalid = len(result.changed_addresses) != len(set(result.changed_addresses)) + return invalid + + +def _protocol_invalid_diagnostic(context: _DueActionContext) -> Diagnostic: + return Diagnostic( + code="runtime.participant-autonomous-action-protocol-invalid", + domain="participant", + address=context.participant_address, + message=( + "Backend returned a concurrent participant result that did not satisfy the bound terminal-result protocol." + ), + ) + + +def _commit_valid_concurrent_result( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + request: ParticipantActionAdmissionRequest, + result: ParticipantActionApplyResult, + base: RuntimeSnapshot, + run: SchedulerRunState, +) -> bool: + from .participant_scheduler_operations import _next_action_state + + try: + run.working = _stage_concurrent_action_snapshot(base, run.working, result.snapshot) + except (Exception, CancelledError): # NOSONAR - backend snapshot merge is a trust boundary + should_stop = _settle_failed_concurrent_occurrence( + context, + state, + request, + run, + diagnostic=Diagnostic( + code="runtime.participant-concurrent-commit-conflict", + domain="participant", + address=context.key, + message="Backend concurrent participant state could not be merged at the serialized commit boundary.", + ), + ) + else: + action_result = result.action_result + action_succeeded = bool(result.success and action_result is not None and action_result.status == "succeeded") + next_state = _next_action_state( + context, + state, + request, + action_succeeded=action_succeeded, + protocol_failure=False, + ).model_copy(update={"in_flight": state.in_flight}) + scheduler_states = dict(run.working.participant_autonomous_execution_states) + scheduler_states[context.key] = next_state.model_dump(mode="json") + run.working = _with_concurrent_scheduler_updates( + run.working, + states=scheduler_states, + ) + run.diagnostics.extend(result.diagnostics) + run.changed.extend([*result.changed_addresses, context.key]) + should_stop = not action_succeeded and context.policy.failure_policy == "stop" + return should_stop + + +def _commit_concurrent_result( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + request: ParticipantActionAdmissionRequest, + result: object, + protocol_invalid: bool, + base: RuntimeSnapshot, + run: SchedulerRunState, +) -> bool: + stale_completion = participant_generation_commit_diagnostic(request, run.working) + if stale_completion is not None: + should_stop = _settle_failed_concurrent_occurrence( + context, + state, + request, + run, + diagnostic=stale_completion, + ) + elif protocol_invalid: + should_stop = _settle_failed_concurrent_occurrence( + context, + state, + request, + run, + diagnostic=_protocol_invalid_diagnostic(context), + ) + else: + should_stop = _commit_valid_concurrent_result( + context, + state, + request, + cast(ParticipantActionApplyResult, result), + base, + run, + ) + return should_stop + + +__all__ = ( + "_commit_concurrent_result", + "_concurrent_result_protocol_invalid", + "participant_generation_commit_diagnostic", +) diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_dispatch.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_dispatch.py new file mode 100644 index 00000000..367f3a7f --- /dev/null +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_dispatch.py @@ -0,0 +1,354 @@ +"""Preparation, dispatch, and settlement orchestration for concurrent batches.""" + +from __future__ import annotations + +from asyncio import CancelledError +from collections.abc import Callable +from copy import deepcopy +from dataclasses import dataclass +from itertools import islice +from typing import TYPE_CHECKING, cast + +from raes_contracts.contracts import ParticipantAutonomousExecutionStateModel +from raes_contracts.diagnostics import Diagnostic +from raes_contracts.participant_binding import ParticipantActionAdmissionRequest +from raes_contracts.runtime_state import ApplyResult, RuntimeSnapshot +from raes_processor.models import CompiledTimeModel, ParticipantAutonomousExecutionRuntime + +from .participant_scheduler_concurrent_commit import ( + _commit_concurrent_result, + _concurrent_result_protocol_invalid, +) +from .participant_scheduler_concurrent_settlement import ( + _ConcurrentBatchSettlement, + _fail_concurrent_batch_before_dispatch, + _set_concurrent_failure, + _settle_concurrent_service_state, + _settle_indeterminate_concurrent_batch, +) +from .participant_scheduler_concurrent_state import ( + _freeze_concurrent_results, + _materialize_concurrent_snapshot, + _reserve_concurrent_actions, +) + +if TYPE_CHECKING: + from .participant_scheduler_types import SchedulerRunState, _DueActionContext + + +_CONCURRENT_SNAPSHOT_ISOLATION_FAILED = "runtime.participant-concurrent-snapshot-isolation-failed" + + +def _unsupported_concurrency_failure(policy: ParticipantAutonomousExecutionRuntime, run: SchedulerRunState) -> None: + run.failure = ApplyResult( + success=False, + snapshot=run.working, + diagnostics=[ + Diagnostic( + code="runtime.participant-concurrency-unsupported", + domain="participant", + address=policy.address, + message="Backend declared bounded participant concurrency without an executable batch method.", + ) + ], + ) + + +@dataclass(frozen=True) +class _ConcurrentBatch: + policy: ParticipantAutonomousExecutionRuntime + time_model: CompiledTimeModel + participant_runtime: object + current_tick: int + cadence_ticks: int + run: SchedulerRunState + contexts: tuple[_DueActionContext, ...] + states: tuple[ParticipantAutonomousExecutionStateModel, ...] + pre_batch: RuntimeSnapshot | None = None + materialize: bool = True + + +_ConcurrentBatchMethod = Callable[ + [tuple[ParticipantActionAdmissionRequest, ...], RuntimeSnapshot, int], + object, +] + + +@dataclass(frozen=True) +class _PreparedConcurrentBatch: + batch_method: _ConcurrentBatchMethod + settlement: _ConcurrentBatchSettlement + base: RuntimeSnapshot + dispatch_snapshot: RuntimeSnapshot + materialize: bool + + +def _isolate_concurrent_batch_predecessors( + batch: _ConcurrentBatch, +) -> tuple[RuntimeSnapshot, RuntimeSnapshot] | None: + # RuntimeSnapshot owns mutable nested mappings. Keep both the scheduler's + # pre-dispatch rollback point and the binding predecessor isolated. Once the + # backend dispatch boundary is entered, rollback is no longer truthful: + # missing results are indeterminate native work and must become non-retryable. + isolated = None + try: + provided_pre_batch = getattr(batch, "pre_batch", None) + pre_batch = provided_pre_batch if provided_pre_batch is not None else deepcopy(batch.run.working) + binding_snapshot = deepcopy(pre_batch) + except (Exception, CancelledError): # NOSONAR - local snapshot isolation must fail closed + _fail_concurrent_batch_before_dispatch( + batch.run, + batch.run.working, + policy_address=batch.policy.address, + code=_CONCURRENT_SNAPSHOT_ISOLATION_FAILED, + message="Concurrent participant snapshot isolation did not complete before dispatch.", + ) + else: + isolated = (pre_batch, binding_snapshot) + return isolated + + +def _bind_concurrent_batch_requests( + batch: _ConcurrentBatch, + binding_snapshot: RuntimeSnapshot, + pre_batch: RuntimeSnapshot, +) -> tuple[ParticipantActionAdmissionRequest, ...] | None: + from .participant_scheduler_operations import _bound_action_request + + requests = None + try: + requests = tuple( + _bound_action_request(context, binding_snapshot, state) + for context, state in zip(batch.contexts, batch.states, strict=True) + ) + except (Exception, CancelledError): # NOSONAR - backend binding is a trust boundary + _fail_concurrent_batch_before_dispatch( + batch.run, + pre_batch, + policy_address=batch.policy.address, + code="runtime.participant-concurrent-binding-failed", + message="Backend concurrent participant request binding did not complete.", + ) + return requests + + +def _reserve_concurrent_batch_for_dispatch( + batch: _ConcurrentBatch, + pre_batch: RuntimeSnapshot, +) -> tuple[RuntimeSnapshot, RuntimeSnapshot] | None: + reserved = None + batch.run.working = pre_batch + try: + _reserve_concurrent_actions(batch.run, batch.contexts) + except (Exception, CancelledError): # NOSONAR - capacity/readback drift fails before dispatch + _fail_concurrent_batch_before_dispatch( + batch.run, + pre_batch, + policy_address=batch.policy.address, + code="runtime.participant-concurrent-reservation-failed", + message="Concurrent participant reservations could not be admitted within service capacity.", + ) + else: + base = batch.run.working + try: + dispatch_snapshot = deepcopy(base) + except (Exception, CancelledError): # NOSONAR - no backend action has been submitted yet + _fail_concurrent_batch_before_dispatch( + batch.run, + pre_batch, + policy_address=batch.policy.address, + code=_CONCURRENT_SNAPSHOT_ISOLATION_FAILED, + message="Concurrent participant dispatch isolation did not complete before dispatch.", + ) + else: + reserved = (base, dispatch_snapshot) + return reserved + + +def _prepare_concurrent_batch( + batch: _ConcurrentBatch, + batch_method: _ConcurrentBatchMethod, +) -> _PreparedConcurrentBatch | None: + isolated = _isolate_concurrent_batch_predecessors(batch) + if isolated is None: + return None + pre_batch, binding_snapshot = isolated + requests = _bind_concurrent_batch_requests(batch, binding_snapshot, pre_batch) + if requests is None: + return None + reserved = _reserve_concurrent_batch_for_dispatch(batch, pre_batch) + prepared = None + if reserved is not None: + base, dispatch_snapshot = reserved + prepared = _PreparedConcurrentBatch( + batch_method=batch_method, + settlement=_ConcurrentBatchSettlement( + run=batch.run, + policy_address=batch.policy.address, + contexts=tuple(batch.contexts), + states=tuple(batch.states), + requests=requests, + pre_batch=pre_batch, + ), + base=base, + dispatch_snapshot=dispatch_snapshot, + materialize=cast(bool, getattr(batch, "materialize", True)), + ) + return prepared + + +def _dispatch_concurrent_results( + prepared: _PreparedConcurrentBatch, +) -> tuple[object, ...] | None: + requests = prepared.settlement.requests + results = None + # The batch method is backend-supplied. Once called, every submitted request + # is potentially side-effecting. A raised/cancelled call or an unpairable + # result collection must settle those action ids instead of restoring them. + try: + raw_results = prepared.batch_method(requests, prepared.dispatch_snapshot, len(requests)) + # Consume at most one result beyond the declared count. This freezes a + # mutable/generator response and cannot hang on an unbounded iterable. + results = tuple(islice(iter(raw_results), len(requests) + 1)) + except (Exception, CancelledError): # NOSONAR - dispatched work is now indeterminate + _settle_indeterminate_concurrent_batch( + prepared.settlement, + code="runtime.participant-concurrent-batch-failed", + message="Backend concurrent participant batch became indeterminate after dispatch.", + ) + if results is not None and len(results) != len(requests): + _settle_indeterminate_concurrent_batch( + prepared.settlement, + code="runtime.participant-concurrent-result-count-invalid", + message=( + "Backend concurrent participant result count did not match submitted work; " + "the dispatched actions are indeterminate." + ), + ) + results = None + return results + + +def _freeze_dispatched_concurrent_results( + prepared: _PreparedConcurrentBatch, + results: tuple[object, ...], +) -> tuple[tuple[object, ...], tuple[bool, ...]] | None: + frozen = None + try: + frozen_results, freeze_invalid = _freeze_concurrent_results(results, deepcopy) + except CancelledError: + # Cancellation after dispatch makes the whole batch indeterminate. + _settle_indeterminate_concurrent_batch( + prepared.settlement, + code="runtime.participant-concurrent-batch-cancelled", + message="Concurrent participant result isolation was cancelled after dispatch.", + ) + else: + frozen = (frozen_results, freeze_invalid) + return frozen + + +def _materialize_prepared_concurrent_batch( + prepared: _PreparedConcurrentBatch, + *, + batch_failed: bool, +) -> bool: + run = prepared.settlement.run + materialized = True + if prepared.materialize or batch_failed or run.failure is not None: + try: + run.working = _materialize_concurrent_snapshot(run.working) + except (Exception, CancelledError): # NOSONAR - the commit boundary must fail closed + _set_concurrent_failure( + run, + Diagnostic( + code="runtime.participant-concurrent-commit-invalid", + domain="participant", + address=prepared.settlement.policy_address, + message="Concurrent participant batch state failed final invariant validation.", + ), + ) + materialized = False + return materialized + + +def _commit_prepared_concurrent_results( + prepared: _PreparedConcurrentBatch, + frozen_results: tuple[object, ...], + freeze_invalid: tuple[bool, ...], +) -> None: + settlement = prepared.settlement + protocol_invalid = tuple( + invalid + or _concurrent_result_protocol_invalid( + request, + result, + episode_id=state.episode_id, + predecessor=prepared.base, + ) + for state, request, result, invalid in zip( + settlement.states, + settlement.requests, + frozen_results, + freeze_invalid, + strict=True, + ) + ) + batch_failed = False + for context, state, request, result, invalid in zip( + settlement.contexts, + settlement.states, + settlement.requests, + frozen_results, + protocol_invalid, + strict=True, + ): + # Every peer was already dispatched. Settle every peer before honoring + # stop semantics so valid native outcomes are never silently dropped. + batch_failed = ( + _commit_concurrent_result( + context, + state, + request, + result, + invalid, + prepared.base, + settlement.run, + ) + or batch_failed + ) + service_settled = _settle_concurrent_service_state( + settlement.run, + policy_address=settlement.policy_address, + completed_count=len(settlement.requests), + pre_batch=settlement.pre_batch, + ) + materialized = _materialize_prepared_concurrent_batch(prepared, batch_failed=batch_failed) + if materialized and batch_failed and service_settled: + _set_concurrent_failure(settlement.run) + + +def _execute_prepared_concurrent_batch(prepared: _PreparedConcurrentBatch) -> None: + results = _dispatch_concurrent_results(prepared) + if results is not None: + frozen = _freeze_dispatched_concurrent_results(prepared, results) + if frozen is not None: + _commit_prepared_concurrent_results(prepared, *frozen) + + +def _execute_concurrent_batch(batch: _ConcurrentBatch) -> None: + batch_method = getattr(batch.participant_runtime, "admit_actions_concurrently", None) + if callable(batch_method): + prepared = _prepare_concurrent_batch(batch, cast(_ConcurrentBatchMethod, batch_method)) + if prepared is not None: + _execute_prepared_concurrent_batch(prepared) + else: + _unsupported_concurrency_failure(batch.policy, batch.run) + + +__all__ = ( + "_CONCURRENT_SNAPSHOT_ISOLATION_FAILED", + "_ConcurrentBatch", + "_execute_concurrent_batch", + "_unsupported_concurrency_failure", +) diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_settlement.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_settlement.py new file mode 100644 index 00000000..5aa6e174 --- /dev/null +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_settlement.py @@ -0,0 +1,184 @@ +"""Failure settlement for bounded concurrent participant dispatch.""" + +from __future__ import annotations + +from asyncio import CancelledError +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from raes_contracts.contracts import ParticipantAutonomousExecutionStateModel +from raes_contracts.diagnostics import Diagnostic +from raes_contracts.participant_binding import ParticipantActionAdmissionRequest +from raes_contracts.runtime_state import ApplyResult, RuntimeSnapshot + +from .participant_scheduler_concurrent_state import ( + _finish_concurrent_service_state, + _with_concurrent_scheduler_updates, +) + +if TYPE_CHECKING: + from .participant_scheduler_types import SchedulerRunState, _DueActionContext + + +@dataclass(frozen=True) +class _ConcurrentBatchSettlement: + """Authoritative scheduler state needed to settle one dispatched batch.""" + + run: SchedulerRunState + policy_address: str + contexts: tuple[_DueActionContext, ...] + states: tuple[ParticipantAutonomousExecutionStateModel, ...] + requests: tuple[ParticipantActionAdmissionRequest, ...] + pre_batch: RuntimeSnapshot + + +def _set_concurrent_failure(run: SchedulerRunState, diagnostic: Diagnostic | None = None) -> None: + if diagnostic is not None: + run.diagnostics.append(diagnostic) + run.failure = ApplyResult( + success=False, + snapshot=run.working, + diagnostics=run.diagnostics, + changed_addresses=list(dict.fromkeys(run.changed)), + ) + + +def _fail_concurrent_batch_before_dispatch( + run: SchedulerRunState, + pre_batch: RuntimeSnapshot, + *, + policy_address: str, + code: str, + message: str, +) -> None: + """Restore scheduler state only when no native action could have run.""" + + run.working = pre_batch + _set_concurrent_failure( + run, + Diagnostic( + code=code, + domain="participant", + address=policy_address, + message=message, + ), + ) + + +def _settle_failed_concurrent_occurrence( + context: _DueActionContext, + state: ParticipantAutonomousExecutionStateModel, + request: ParticipantActionAdmissionRequest, + run: SchedulerRunState, + *, + diagnostic: Diagnostic | None = None, +) -> bool: + """Settle one dispatched occurrence without committing its native snapshot.""" + + from .participant_scheduler_operations import _next_action_state + + if diagnostic is not None: + run.diagnostics.append(diagnostic) + next_state = _next_action_state( + context, + state, + request, + action_succeeded=False, + protocol_failure=True, + ).model_copy(update={"in_flight": state.in_flight}) + scheduler_states = dict(run.working.participant_autonomous_execution_states) + scheduler_states[context.key] = next_state.model_dump(mode="json") + run.working = _with_concurrent_scheduler_updates( + run.working, + states=scheduler_states, + ) + run.changed.append(context.key) + return True + + +def _settle_concurrent_service_state( + run: SchedulerRunState, + *, + policy_address: str, + completed_count: int, + pre_batch: RuntimeSnapshot, +) -> bool: + """Release this batch or restore its exact prior service counters.""" + + try: + _finish_concurrent_service_state(run, policy_address, completed_count) + prior = pre_batch.participant_execution_services.get(policy_address) + settled = run.working.participant_execution_services.get(policy_address) + if settled != prior: + raise ValueError("concurrent participant service settlement did not restore prior accounting") + except (Exception, CancelledError): # NOSONAR - settlement must not leak a backend-facing exception + services = dict(run.working.participant_execution_services) + prior = pre_batch.participant_execution_services.get(policy_address) + if prior is None: + services.pop(policy_address, None) + else: + services[policy_address] = prior + run.working = _with_concurrent_scheduler_updates( + run.working, + services=services, + ) + _set_concurrent_failure( + run, + Diagnostic( + code="runtime.participant-concurrent-service-settlement-failed", + domain="participant", + address=policy_address, + message="Concurrent participant service accounting could not be settled normally and was restored.", + ), + ) + return False + return True + + +def _settle_indeterminate_concurrent_batch( + settlement: _ConcurrentBatchSettlement, + *, + code: str, + message: str, +) -> None: + """Fence every submitted action after an unpairable dispatch outcome. + + Once the backend dispatch boundary is entered, an exception, cancellation, + or malformed result collection cannot prove that any native side effect did + not occur. Every submitted action is therefore settled as a non-retryable + protocol failure while only this batch's service-accounting delta is + released. + """ + + settlement.run.diagnostics.append( + Diagnostic( + code=code, + domain="participant", + address=settlement.policy_address, + message=message, + ) + ) + for context, state, request in zip( + settlement.contexts, + settlement.states, + settlement.requests, + strict=True, + ): + _settle_failed_concurrent_occurrence(context, state, request, settlement.run) + if _settle_concurrent_service_state( + settlement.run, + policy_address=settlement.policy_address, + completed_count=len(settlement.requests), + pre_batch=settlement.pre_batch, + ): + _set_concurrent_failure(settlement.run) + + +__all__ = ( + "_ConcurrentBatchSettlement", + "_fail_concurrent_batch_before_dispatch", + "_set_concurrent_failure", + "_settle_concurrent_service_state", + "_settle_failed_concurrent_occurrence", + "_settle_indeterminate_concurrent_batch", +) diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_state.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_state.py new file mode 100644 index 00000000..1c16e826 --- /dev/null +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrent_state.py @@ -0,0 +1,322 @@ +"""Revision-safe state updates for bounded concurrent participant execution.""" + +from __future__ import annotations + +from collections.abc import Callable +from copy import copy, deepcopy +from dataclasses import fields +from typing import TYPE_CHECKING + +from raes_contracts.contracts import ParticipantAutonomousExecutionStateModel +from raes_contracts.contracts.participant_execution import ParticipantExecutionServiceStateModel +from raes_contracts.runtime_state import RuntimeSnapshot +from raes_processor.models import ParticipantAutonomousExecutionRuntime + +from .participant_action_validation import _PROTECTED_SCHEDULER_SNAPSHOT_FIELDS + +if TYPE_CHECKING: + from .participant_scheduler_types import SchedulerRunState, _DueActionContext + + +_MISSING = object() + +# Native participant execution receives the complete portable snapshot, and the +# serial scheduler commits that complete validated result. Concurrent commit +# therefore needs an explicit owner for every RuntimeSnapshot field: silently +# ignoring a newly added field would make serial and concurrent execution mean +# different things. Scheduler state and execution-service accounting are the +# only protected fields; a native result must carry their reserved predecessor +# unchanged and the serialized scheduler applies their deltas itself. +_PROTECTED_SCHEDULER_FIELDS = frozenset(_PROTECTED_SCHEDULER_SNAPSHOT_FIELDS) +_BACKEND_MAPPING_FIELDS = ( + "entries", + "orchestration_results", + "orchestration_history", + "evaluation_results", + "evaluation_history", + "proposition_truth_results", + "participant_episode_results", + "participant_episode_history", + "participant_episode_closure_records", + "participant_behavior_history", + "participant_control_history", + "participant_crossing_history", + "information_state_history", + "participant_resource_budget_states", + "participant_resource_pool_states", + "participant_resource_budget_events", + "shared_state_records", + "shared_state_history", + "joint_action_records", + "time_management_contexts", + "metadata", +) +_BACKEND_VALUE_FIELDS = ( + "time_model_state", + "realization_provenance", + "realization_observations", + "realization_envelope", +) +_OWNED_SNAPSHOT_FIELDS = frozenset((*_PROTECTED_SCHEDULER_FIELDS, *_BACKEND_MAPPING_FIELDS, *_BACKEND_VALUE_FIELDS)) +_DECLARED_SNAPSHOT_FIELDS = frozenset(field.name for field in fields(RuntimeSnapshot)) + + +def _assert_snapshot_field_ownership( + declared_fields: frozenset[str], + owned_fields: frozenset[str], +) -> None: + """Fail when a snapshot field has no explicit concurrent owner.""" + + missing = sorted(declared_fields - owned_fields) + stale = sorted(owned_fields - declared_fields) + if not missing and not stale: + return + raise RuntimeError( + f"concurrent participant snapshot ownership is incomplete (missing={missing!r}, stale={stale!r})" + ) + + +_assert_snapshot_field_ownership(_DECLARED_SNAPSHOT_FIELDS, _OWNED_SNAPSHOT_FIELDS) + + +def _changed_mapping( + base: dict[str, object], + incoming: dict[str, object], +) -> dict[str, object]: + changed: dict[str, object] = {} + for key in base.keys() | incoming.keys(): + base_value = base.get(key, _MISSING) + incoming_value = incoming.get(key, _MISSING) + if base_value != incoming_value: + changed[key] = incoming_value + return changed + + +def _merge_mapping_revision_checked( + *, + base: dict[str, object], + current: dict[str, object], + incoming: dict[str, object], + field_name: str, +) -> dict[str, object]: + # The frozen result envelope is already detached from the backend, but the + # committed snapshot must not retain aliases to either the result or an + # older authoritative snapshot. Deep-copying the selected values makes the + # commit a true ownership transfer rather than a shallow dict replacement. + merged = deepcopy(current) + changed = _changed_mapping(base, incoming) + for key in sorted(changed): + value = changed[key] + current_value = current.get(key, _MISSING) + base_value = base.get(key, _MISSING) + if current_value != base_value and current_value != value: + # Mapping keys are backend-controlled. Keep diagnostics stable and + # never copy a key (which may contain participant data) into them. + raise ValueError(f"concurrent participant commit conflict in {field_name}") + if value is _MISSING: + merged.pop(key, None) + else: + merged[key] = deepcopy(value) + return merged + + +def _merge_value_revision_checked( + *, + base: object, + current: object, + incoming: object, + field_name: str, +) -> object: + if incoming == base: + return deepcopy(current) + if current != base and current != incoming: + raise ValueError(f"concurrent participant commit conflict in {field_name}") + return deepcopy(incoming) + + +def _require_protected_fields_unchanged( + base: RuntimeSnapshot, + incoming: RuntimeSnapshot, +) -> None: + for field_name in sorted(_PROTECTED_SCHEDULER_FIELDS): + if getattr(incoming, field_name) != getattr(base, field_name): + raise ValueError(f"concurrent participant result changed protected field {field_name}") + + +def _merge_concurrent_action_snapshot( + base: RuntimeSnapshot, + current: RuntimeSnapshot, + incoming: RuntimeSnapshot, +) -> RuntimeSnapshot: + """Three-way merge every backend-owned field without replacing scheduler state.""" + + staged = _stage_concurrent_action_snapshot(base, current, incoming) + return _materialize_concurrent_snapshot(staged) + + +def _stage_concurrent_action_snapshot( + base: RuntimeSnapshot, + current: RuntimeSnapshot, + incoming: RuntimeSnapshot, +) -> RuntimeSnapshot: + """Merge backend-owned fields while deferring the full scheduler-state scan.""" + + _require_protected_fields_unchanged(base, incoming) + staged = copy(current) + for field_name in _BACKEND_MAPPING_FIELDS: + base_mapping = getattr(base, field_name) + incoming_mapping = getattr(incoming, field_name) + merged = getattr(current, field_name) + if incoming_mapping != base_mapping: + merged = _merge_mapping_revision_checked( + base=dict(base_mapping), + current=dict(merged), + incoming=dict(incoming_mapping), + field_name=field_name, + ) + setattr(staged, field_name, merged) + for field_name in _BACKEND_VALUE_FIELDS: + base_value = getattr(base, field_name) + incoming_value = getattr(incoming, field_name) + merged_value = getattr(current, field_name) + if incoming_value != base_value: + merged_value = _merge_value_revision_checked( + base=base_value, + current=merged_value, + incoming=incoming_value, + field_name=field_name, + ) + setattr(staged, field_name, merged_value) + # Validate every backend-owned RuntimeSnapshot invariant for this result, + # but omit the protected participant maps already checked above. The full + # participant-state oracle runs once when the completed batch materializes. + staged.with_entries( + dict(staged.entries), + participant_autonomous_execution_states={}, + participant_execution_services={}, + ) + return staged + + +def _with_concurrent_scheduler_updates( + snapshot: RuntimeSnapshot, + *, + states: dict[str, dict[str, object]] | None = None, + services: dict[str, dict[str, object]] | None = None, +) -> RuntimeSnapshot: + """Apply already model-validated scheduler updates without a global rescan.""" + + staged = copy(snapshot) + if states is not None: + staged.participant_autonomous_execution_states = states + if services is not None: + staged.participant_execution_services = services + return staged + + +def _materialize_concurrent_snapshot(snapshot: RuntimeSnapshot) -> RuntimeSnapshot: + """Detach and fully validate a completed concurrent batch.""" + + return snapshot.with_entries( + dict(snapshot.entries), + participant_autonomous_execution_states=dict(snapshot.participant_autonomous_execution_states), + participant_execution_services=dict(snapshot.participant_execution_services), + ) + + +def _freeze_concurrent_results( + results: tuple[object, ...], + copier: Callable[[object], object], +) -> tuple[list[object], list[bool]]: + """Detach result envelopes while preserving repeated-object identity.""" + + frozen_results: list[object] = [] + freeze_invalid: list[bool] = [] + frozen_by_identity: dict[int, tuple[object, object]] = {} + for result in results: + cached = frozen_by_identity.get(id(result)) + if cached is not None and cached[0] is result: + frozen_results.append(cached[1]) + freeze_invalid.append(False) + continue + try: + frozen = copier(result) + frozen_results.append(frozen) + freeze_invalid.append(False) + frozen_by_identity[id(result)] = (result, frozen) + except Exception: # NOSONAR - reject only the unfreezable paired result + frozen_results.append(None) + freeze_invalid.append(True) + return frozen_results, freeze_invalid + + +def _reserve_concurrent_actions( + run: SchedulerRunState, + contexts: tuple[_DueActionContext, ...], +) -> None: + states = dict(run.working.participant_autonomous_execution_states) + for context in contexts: + state = ParticipantAutonomousExecutionStateModel.model_validate(states[context.key]) + states[context.key] = state.model_copy( + update={ + "attempted_actions": state.attempted_actions + 1, + "in_flight": state.in_flight + 1, + } + ).model_dump(mode="json") + services = dict(run.working.participant_execution_services) + for policy_address in {context.policy.address for context in contexts}: + payload = services.get(policy_address) + if payload is None: + continue + service = ParticipantExecutionServiceStateModel.model_validate(payload) + count = sum(1 for context in contexts if context.policy.address == policy_address) + available = service.capacity - service.reserved - service.in_flight + if count > available: + raise ValueError("concurrent participant batch exceeds available execution-service capacity") + services[policy_address] = service.model_copy( + update={ + "in_flight": service.in_flight + count, + "quiescent": False, + } + ).model_dump(mode="json") + run.working = _with_concurrent_scheduler_updates( + run.working, + states=states, + services=services, + ) + + +def _finish_concurrent_service_state( + run: SchedulerRunState, + policy_address: str, + completed_count: int, +) -> None: + services = dict(run.working.participant_execution_services) + payload = services.get(policy_address) + if payload is None: + return + service = ParticipantExecutionServiceStateModel.model_validate(payload) + if completed_count > service.in_flight: + raise ValueError("concurrent participant completion exceeds execution-service in-flight work") + remaining = service.in_flight - completed_count + services[policy_address] = service.model_copy( + update={ + "in_flight": remaining, + "quiescent": service.reserved == 0 and remaining == 0, + } + ).model_dump(mode="json") + run.working = _with_concurrent_scheduler_updates( + run.working, + services=services, + ) + + +def _available_concurrent_capacity( + policy: ParticipantAutonomousExecutionRuntime, + run: SchedulerRunState, +) -> int: + payload = run.working.participant_execution_services.get(policy.address) + if payload is None: + return 0 + service = ParticipantExecutionServiceStateModel.model_validate(payload) + return min(policy.max_in_flight, max(0, service.capacity - service.reserved - service.in_flight)) diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_operations.py b/implementations/python/packages/raes_runtime/participant_scheduler_operations.py index 67ab20a6..ba2eef8e 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_operations.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_operations.py @@ -31,7 +31,8 @@ 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_concurrency import run_policy_due_concurrently +from .participant_scheduler_concurrent_commit import participant_generation_commit_diagnostic from .participant_scheduler_resources import ( commit_activity_resources, measurement_requirements, @@ -296,6 +297,7 @@ def _activity_action_is_due( state.lifecycle_state == "running", state.next_tick == context.current_tick, state.attempted_actions < context.policy.max_action_attempts, + state.in_flight == 0, run.failure is None, ) ) @@ -396,6 +398,7 @@ def _legacy_action_is_due( state.lifecycle_state == "running", state.next_tick == context.current_tick, state.attempted_actions < context.policy.max_action_attempts, + state.in_flight == 0, run.failure is None, ) ) diff --git a/implementations/python/tests/test_issue_898_participant_execution_control.py b/implementations/python/tests/test_issue_898_participant_execution_control.py index ebd423e4..8363b583 100644 --- a/implementations/python/tests/test_issue_898_participant_execution_control.py +++ b/implementations/python/tests/test_issue_898_participant_execution_control.py @@ -15,6 +15,7 @@ ) from implementations.python.tests.test_runtime_control_plane_api import _test_security from raes import parse_sdl +from raes.participant_behavior import ParticipantFailureClass from raes_backend_protocols.capability_admission import ( participant_autonomous_execution_capability_gaps, ) @@ -26,19 +27,26 @@ from raes_backend_stubs.stubs import create_stub_target from raes_conformance.conformance.profiles import BackendCapabilityProfile from raes_conformance.conformance.target_probes import _target_adapter_cases -from raes_contracts.contracts import ParticipantTemporalRuntimeContextModel +from raes_contracts.contracts import ( + ParticipantAutonomousExecutionStateModel, + ParticipantTemporalRuntimeContextModel, +) from raes_contracts.contracts.participant_execution import ( ParticipantExecutionBindingModel, ParticipantExecutionControlRequestModel, ParticipantExecutionServiceStateModel, ) +from raes_contracts.participant_binding import ( + ParticipantActionAdmissionRequest, + ParticipantNativeActionExecution, +) from raes_contracts.participant_episode import ParticipantEpisodeInitializeRequest from raes_contracts.runtime_state import ApplyResult, RuntimeSnapshot from raes_processor.compiler import compile_runtime_model from raes_runtime.control_plane import RuntimeControlPlane from raes_runtime.control_plane_api import create_control_plane_app from raes_runtime.manager import RuntimeManager -from raes_runtime.participant_scheduler_concurrency import ( +from raes_runtime.participant_scheduler_concurrent_commit import ( participant_generation_commit_diagnostic, ) from starlette.testclient import TestClient @@ -483,17 +491,123 @@ def _model_action(self, request, snapshot, *, episode_id): self.peak_active = max(self.peak_active, self._active) try: self._barrier.wait(timeout=2) - return super()._model_action( + execution = super()._model_action( request, snapshot, episode_id=episode_id, ) + metadata = dict(execution.apply_result.snapshot.metadata) + action_instance_id = metadata.pop("last_native_action") + metadata[f"last_native_action:{request.participant_address}"] = action_instance_id + return replace( + execution, + apply_result=replace( + execution.apply_result, + snapshot=execution.apply_result.snapshot.with_entries( + dict(execution.apply_result.snapshot.entries), + metadata=metadata, + ), + ), + ) finally: with self._active_lock: self._active -= 1 -def _two_green_participant_scenario(): +class _OneFailedOverlappingParticipantRuntime(_OverlappingParticipantRuntime): + def _model_action( + self, + request: ParticipantActionAdmissionRequest, + snapshot: RuntimeSnapshot, + *, + episode_id: str, + ) -> ParticipantNativeActionExecution: + execution = super()._model_action(request, snapshot, episode_id=episode_id) + if request.participant_address.endswith("participant-agent"): + assert execution.action_result is not None + return replace( + execution, + action_result=execution.action_result.model_copy( + update={ + "status": "failed", + "failure_class": ParticipantFailureClass.TARGET_UNAVAILABLE, + } + ), + ) + return execution + + +class _SnapshotProjectionParticipantRuntime(_NativeParticipantRuntime): + """Backend that writes a portable field outside the historical merge allowlist.""" + + def __init__(self) -> None: + super().__init__() + self._retained_results: list[object] = [] + + def _model_action( + self, + request: ParticipantActionAdmissionRequest, + snapshot: RuntimeSnapshot, + *, + episode_id: str, + ) -> ParticipantNativeActionExecution: + execution = super()._model_action(request, snapshot, episode_id=episode_id) + result_address = f"evaluation.result.native-{request.participant_address.rsplit('.', 1)[-1]}" + metadata = dict(execution.apply_result.snapshot.metadata) + action_instance_id = metadata.pop("last_native_action") + metadata[f"last_native_action:{request.participant_address}"] = action_instance_id + evaluation_results = { + **execution.apply_result.snapshot.evaluation_results, + result_address: { + "participant_address": request.participant_address, + "nested": {"state": "committed"}, + }, + } + modeled = replace( + execution.apply_result, + snapshot=execution.apply_result.snapshot.with_entries( + dict(execution.apply_result.snapshot.entries), + evaluation_results=evaluation_results, + metadata=metadata, + ), + changed_addresses=[*execution.apply_result.changed_addresses, result_address], + ) + return replace(execution, apply_result=modeled) + + def admit_action(self, request, snapshot): + result = super().admit_action(request, snapshot) + self._retained_results.append(result) + return result + + +class _ProtectedStateMutationParticipantRuntime(_NativeParticipantRuntime): + """Backend that attempts to write serialized scheduler-owned state.""" + + def _model_action( + self, + request: ParticipantActionAdmissionRequest, + snapshot: RuntimeSnapshot, + *, + episode_id: str, + ) -> ParticipantNativeActionExecution: + execution = super()._model_action(request, snapshot, episode_id=episode_id) + services = { + **execution.apply_result.snapshot.participant_execution_services, + "participant.autonomous-execution.injected": {}, + } + return replace( + execution, + apply_result=replace( + execution.apply_result, + snapshot=execution.apply_result.snapshot.with_entries( + dict(execution.apply_result.snapshot.entries), + participant_execution_services=services, + ), + ), + ) + + +def _two_green_participant_scenario(*, max_in_flight: int = 2): payload = yaml.safe_load(_scenario_yaml()) payload["entities"]["enterprise-participant-2"] = { **payload["entities"]["enterprise-participant"], @@ -506,7 +620,7 @@ def _two_green_participant_scenario(): } specification = payload["behavior_specifications"]["participant-behavior"] specification["participant_refs"].append("participant-agent-2") - specification["autonomous_execution"]["max_in_flight"] = 2 + specification["autonomous_execution"]["max_in_flight"] = max_in_flight return parse_sdl(yaml.safe_dump(payload, sort_keys=False)) @@ -536,6 +650,99 @@ def test_scheduler_executes_two_due_green_participants_with_bounded_overlap() -> assert states[0].in_flight == 0 +def test_serial_and_concurrent_actions_commit_the_same_backend_owned_projection() -> None: + def _apply(max_in_flight: int): + scenario = _two_green_participant_scenario(max_in_flight=max_in_flight) + runtime_model = compile_runtime_model(scenario) + participant_runtime = _SnapshotProjectionParticipantRuntime() + target = replace( + create_stub_target(), + manifest=_autonomous_manifest(runtime_model), + participant_runtime=participant_runtime, + ) + manager = RuntimeManager(target) + return manager.apply(manager.plan(scenario)), participant_runtime + + serial, _serial_runtime = _apply(1) + concurrent, concurrent_runtime = _apply(2) + prefix = "evaluation.result.native-" + serial_projection = { + key: value for key, value in serial.snapshot.evaluation_results.items() if key.startswith(prefix) + } + concurrent_projection = { + key: value for key, value in concurrent.snapshot.evaluation_results.items() if key.startswith(prefix) + } + + assert serial.success is True + assert concurrent.success is True + assert concurrent_projection == serial_projection + assert len(concurrent_projection) == 2 + assert set(concurrent_projection) <= set(concurrent.changed_addresses) + + retained = next( + result + for result in concurrent_runtime._retained_results + if any(key.startswith(prefix) for key in result.snapshot.evaluation_results) + ) + retained_key = next(key for key in retained.snapshot.evaluation_results if key.startswith(prefix)) + retained.snapshot.evaluation_results[retained_key]["nested"]["state"] = "mutated-after-return" + assert concurrent.snapshot.evaluation_results[retained_key]["nested"]["state"] == "committed" + + +def test_serial_action_rejects_backend_mutation_of_scheduler_owned_state() -> None: + scenario = _two_green_participant_scenario(max_in_flight=1) + runtime_model = compile_runtime_model(scenario) + participant_runtime = _ProtectedStateMutationParticipantRuntime() + target = replace( + create_stub_target(), + manifest=_autonomous_manifest(runtime_model), + participant_runtime=participant_runtime, + ) + manager = RuntimeManager(target) + + applied = manager.apply(manager.plan(scenario)) + + assert applied.success is False + assert any( + diagnostic.code == "runtime.participant-autonomous-action-protocol-invalid" + for diagnostic in applied.diagnostics + ) + assert "participant.autonomous-execution.injected" not in applied.snapshot.participant_execution_services + + +def test_stop_policy_settles_every_already_dispatched_peer() -> None: + scenario = _two_green_participant_scenario() + runtime_model = compile_runtime_model(scenario) + participant_runtime = _OneFailedOverlappingParticipantRuntime() + target = replace( + create_stub_target(), + manifest=_autonomous_manifest(runtime_model), + participant_runtime=participant_runtime, + ) + manager = RuntimeManager(target) + + applied = manager.apply(manager.plan(scenario)) + + assert applied.success is False + assert len(participant_runtime.native_actions) == 2 + scheduler_states = { + state.participant_address: state + for state in ( + ParticipantAutonomousExecutionStateModel.model_validate(payload) + for payload in applied.snapshot.participant_autonomous_execution_states.values() + ) + } + failed = scheduler_states["participant.behavior.participant-agent"] + succeeded = scheduler_states["participant.behavior.participant-agent-2"] + assert (failed.lifecycle_state, failed.failed_actions, failed.in_flight) == ("failed", 1, 0) + assert (succeeded.succeeded_actions, succeeded.in_flight) == (1, 0) + assert set(applied.snapshot.participant_behavior_history) == set(scheduler_states) + service = ParticipantExecutionServiceStateModel.model_validate( + applied.snapshot.participant_execution_services["participant.autonomous-execution.participant-behavior"] + ) + assert (service.reserved, service.in_flight, service.quiescent) == (0, 0, True) + + def test_control_plane_exposes_authenticated_generation_bound_execution_control() -> None: scope = "participant.autonomous-execution.green-activity" initial_snapshot = RuntimeSnapshot(participant_execution_services={scope: _service_state().model_dump(mode="json")}) diff --git a/implementations/python/tests/test_participant_concurrent_batch_reservations.py b/implementations/python/tests/test_participant_concurrent_batch_reservations.py new file mode 100644 index 00000000..5fa6dd84 --- /dev/null +++ b/implementations/python/tests/test_participant_concurrent_batch_reservations.py @@ -0,0 +1,1271 @@ +"""Reservation release when a backend concurrent participant batch misbehaves.""" + +from __future__ import annotations + +import threading +from asyncio import CancelledError +from copy import deepcopy +from dataclasses import dataclass +from dataclasses import fields as dataclass_fields +from types import SimpleNamespace + +import pytest +import raes_contracts.runtime_state as runtime_state_contracts +import raes_runtime.participant_scheduler as participant_scheduler +import raes_runtime.participant_scheduler_concurrency as scheduler_concurrency +import raes_runtime.participant_scheduler_concurrent_commit as scheduler_commit +import raes_runtime.participant_scheduler_concurrent_dispatch as scheduler_dispatch +import raes_runtime.participant_scheduler_concurrent_settlement as scheduler_settlement +import raes_runtime.participant_scheduler_operations as scheduler_operations +from raes.explicitness import ExplicitnessClass, ExplicitnessProvenance +from raes_backend_protocols.participant_execution_runtime import ParticipantExecutionRuntimeMixin +from raes_contracts.contracts import ParticipantAutonomousExecutionStateModel +from raes_contracts.contracts.participant_execution import ParticipantExecutionServiceStateModel +from raes_contracts.participant_binding import ParticipantActionApplyResult +from raes_contracts.runtime_state import RealizationProvenanceEntry, RuntimeSnapshot +from raes_runtime.participant_clock_driver import ParticipantClockDriver +from raes_runtime.participant_scheduler_concurrency import ( + _execute_concurrent_batch, + run_policy_due_concurrently, +) +from raes_runtime.participant_scheduler_concurrent_commit import ( + _concurrent_result_protocol_invalid, +) +from raes_runtime.participant_scheduler_concurrent_state import ( + _BACKEND_MAPPING_FIELDS, + _BACKEND_VALUE_FIELDS, + _OWNED_SNAPSHOT_FIELDS, + _PROTECTED_SCHEDULER_FIELDS, + _assert_snapshot_field_ownership, + _available_concurrent_capacity, + _finish_concurrent_service_state, + _merge_concurrent_action_snapshot, + _merge_mapping_revision_checked, + _merge_value_revision_checked, + _reserve_concurrent_actions, +) +from raes_runtime.participant_scheduler_types import SchedulerRunState + +_POLICY_ADDRESS = "participant.autonomous-execution.green-users" +_IMPLEMENTATION_REF = "participant-implementation-manifests.green-worker.v1" +_PARTICIPANTS = ( + "participant.behavior.green-user-a", + "participant.behavior.green-user-b", +) + + +@dataclass(frozen=True) +class _StubContext: + """Minimal stand-in for `_DueActionContext` for the batch-failure paths.""" + + policy: object + participant_address: str + key: str + cadence_ticks: int = 1 + + +def _execution_state( + participant_address: str, + *, + in_flight: int = 0, +) -> ParticipantAutonomousExecutionStateModel: + return ParticipantAutonomousExecutionStateModel( + policy_address=_POLICY_ADDRESS, + policy_digest="sha256:" + "0" * 64, + participant_address=participant_address, + episode_id=f"{participant_address}-autonomous-0", + participant_implementation_ref=_IMPLEMENTATION_REF, + clock_address="time.clock.scenario-clock", + time_segment=0, + lifecycle_state="running", + next_tick=0, + next_action_index=0, + # attempted_actions == succeeded + failed + in_flight is a snapshot invariant. + attempted_actions=in_flight, + succeeded_actions=0, + failed_actions=0, + in_flight=in_flight, + ) + + +def _service_state( + *, + capacity: int = 2, + reserved: int = 0, + in_flight: int = 0, +) -> ParticipantExecutionServiceStateModel: + digest = "sha256:" + "0" * 64 + return ParticipantExecutionServiceStateModel( + execution_scope_ref="participant.execution-scope.green", + policy_address=_POLICY_ADDRESS, + desired_lifecycle="running", + observed_lifecycle="running", + generation=1, + observed_generation=1, + health="healthy", + readiness="ready", + accepting_new_work=True, + draining=False, + quiescent=in_flight == 0, + resources_released=False, + policy_digest=digest, + binding_digest=digest, + time_declaration_digest=digest, + capacity=capacity, + reserved=reserved, + in_flight=in_flight, + last_transition_ref=f"operation:{_POLICY_ADDRESS}:start:generation-1", + evidence_refs=("evidence.green-login.native-action",), + ) + + +def _state_key(participant_address: str) -> str: + return f"{_POLICY_ADDRESS}.state.{participant_address}" + + +def _batch(participant_runtime: object, *, in_flight: int = 0) -> SimpleNamespace: + policy = SimpleNamespace( + address=_POLICY_ADDRESS, + max_in_flight=2, + max_action_attempts=2, + action_contract_addresses=("participant.action-contract.green-action",), + failure_policy="stop", + ) + states = [_execution_state(address, in_flight=in_flight) for address in _PARTICIPANTS] + snapshot = RuntimeSnapshot( + participant_autonomous_execution_states={ + _state_key(address): state.model_dump(mode="json") + for address, state in zip(_PARTICIPANTS, states, strict=True) + }, + participant_execution_services={_POLICY_ADDRESS: _service_state(in_flight=in_flight).model_dump(mode="json")}, + ) + run = SchedulerRunState(working=snapshot, diagnostics=[], changed=[]) + contexts = [ + _StubContext(policy=policy, participant_address=address, key=_state_key(address)) for address in _PARTICIPANTS + ] + return SimpleNamespace( + policy=policy, + time_model=None, + participant_runtime=participant_runtime, + current_tick=0, + cadence_ticks=1, + run=run, + contexts=contexts, + states=states, + ) + + +@pytest.fixture(autouse=True) +def _stub_request_binding(monkeypatch: pytest.MonkeyPatch) -> None: + """Request construction is irrelevant to the batch-failure paths under test.""" + + monkeypatch.setattr( + scheduler_operations, + "_bound_action_request", + lambda context, working, state: SimpleNamespace( + participant_address=context.participant_address, + execution_scope_ref=None, + action_instance_id=f"{context.participant_address}:attempt-{state.attempted_actions}", + ), + ) + + +def _assert_indeterminate_batch_settled(run: SchedulerRunState, before: RuntimeSnapshot) -> None: + """Dispatched work is fenced while only this batch's service delta is released.""" + + result = run.result() + assert result.success is False + for address in _PARTICIPANTS: + prior = ParticipantAutonomousExecutionStateModel.model_validate( + before.participant_autonomous_execution_states[_state_key(address)] + ) + settled = ParticipantAutonomousExecutionStateModel.model_validate( + result.snapshot.participant_autonomous_execution_states[_state_key(address)] + ) + assert settled.lifecycle_state == "failed" + assert settled.attempted_actions == prior.attempted_actions + 1 + assert settled.failed_actions == prior.failed_actions + 1 + assert settled.in_flight == prior.in_flight + assert settled.last_action_instance_id == f"{address}:attempt-{prior.attempted_actions}" + prior_service = ParticipantExecutionServiceStateModel.model_validate( + before.participant_execution_services[_POLICY_ADDRESS] + ) + settled_service = ParticipantExecutionServiceStateModel.model_validate( + result.snapshot.participant_execution_services[_POLICY_ADDRESS] + ) + assert settled_service.reserved == prior_service.reserved + assert settled_service.in_flight == prior_service.in_flight + assert settled_service.quiescent == (prior_service.reserved == 0 and prior_service.in_flight == 0) + + +def _raise_service_settlement(run, policy_address, completed_count): + del run, policy_address, completed_count + raise RuntimeError("settlement failed") + + +def test_batch_without_concurrent_method_reports_unsupported_backend() -> None: + batch = _batch(object()) + + _execute_concurrent_batch(batch) + + assert batch.run.failure is not None + assert batch.run.failure.diagnostics[0].code == "runtime.participant-concurrency-unsupported" + + +def test_miscounted_backend_batch_is_reported_and_releases_reservations(): + """A wrong result count fences potentially executed actions. + + Restoring the pre-batch attempt counters would make the same action ids due + again even though the backend may have performed their native side effects. + """ + batch = _batch(SimpleNamespace(admit_actions_concurrently=lambda requests, snapshot, workers: ())) + before = batch.run.working + + _execute_concurrent_batch(batch) + + assert batch.run.failure is not None + codes = [diagnostic.code for diagnostic in batch.run.failure.diagnostics] + assert "runtime.participant-concurrent-result-count-invalid" in codes + _assert_indeterminate_batch_settled(batch.run, before) + + +def test_raising_backend_batch_is_reported_and_releases_reservations(): + """A raising backend is a non-retryable indeterminate dispatch.""" + + def _explode(requests, snapshot, workers): + raise RuntimeError("backend exploded") + + batch = _batch(SimpleNamespace(admit_actions_concurrently=_explode)) + before = batch.run.working + + _execute_concurrent_batch(batch) + + assert batch.run.failure is not None + codes = [diagnostic.code for diagnostic in batch.run.failure.diagnostics] + assert "runtime.participant-concurrent-batch-failed" in codes + # Neither the exception text nor its backend-specific type crosses the boundary. + message = next( + d.message for d in batch.run.failure.diagnostics if d.code == "runtime.participant-concurrent-batch-failed" + ) + assert "backend exploded" not in message + assert "RuntimeError" not in message + _assert_indeterminate_batch_settled(batch.run, before) + + +def test_rollback_isolated_from_in_place_backend_mutation(): + """Indeterminate settlement must not commit in-place backend mutation.""" + + def _mutate_then_raise(requests, snapshot, workers): + del requests, workers + snapshot.participant_episode_results["participant.behavior.external"]["nested"]["value"] = "mutated" + raise RuntimeError("backend failed after mutation") + + batch = _batch(SimpleNamespace(admit_actions_concurrently=_mutate_then_raise)) + batch.run.working = batch.run.working.with_entries( + dict(batch.run.working.entries), + participant_episode_results={ + "participant.behavior.external": {"nested": {"value": "before"}}, + }, + ) + before = deepcopy(batch.run.working) + + _execute_concurrent_batch(batch) + + assert batch.run.failure is not None + _assert_indeterminate_batch_settled(batch.run, before) + assert ( + batch.run.result().snapshot.participant_episode_results["participant.behavior.external"]["nested"]["value"] + == "before" + ) + + +def test_success_snapshot_is_detached_from_caller_predecessor(monkeypatch: pytest.MonkeyPatch) -> None: + def _complete(requests, snapshot, workers): + del workers + result = ParticipantActionApplyResult( + success=True, + snapshot=snapshot, + action_result=SimpleNamespace(status="succeeded"), + ) + return tuple(result for _request in requests) + + monkeypatch.setattr(scheduler_commit, "autonomous_action_result_violation", lambda *args, **kwargs: None) + batch = _batch(SimpleNamespace(admit_actions_concurrently=_complete)) + predecessor = batch.run.working.with_entries({}, metadata={"nested": {"value": "before"}}) + batch.run.working = predecessor + + _execute_concurrent_batch(batch) + result = batch.run.result() + result.snapshot.metadata["nested"]["value"] = "after" + + assert result.success is True + assert predecessor.metadata == {"nested": {"value": "before"}} + + +def test_cancelled_backend_batch_settles_every_submitted_action(): + def _cancel(requests, snapshot, workers): + del requests, snapshot, workers + raise CancelledError + + batch = _batch(SimpleNamespace(admit_actions_concurrently=_cancel)) + before = deepcopy(batch.run.working) + + _execute_concurrent_batch(batch) + + assert batch.run.failure is not None + assert batch.run.failure.diagnostics[0].code == "runtime.participant-concurrent-batch-failed" + _assert_indeterminate_batch_settled(batch.run, before) + + +def test_binding_failure_before_dispatch_restores_exact_snapshot(monkeypatch: pytest.MonkeyPatch): + called = False + + def _backend(requests, snapshot, workers): + nonlocal called + del requests, snapshot, workers + called = True + return () + + def _binding_failure(context, working, state): + del context, working, state + raise ValueError("binding failed") + + monkeypatch.setattr(scheduler_operations, "_bound_action_request", _binding_failure) + batch = _batch(SimpleNamespace(admit_actions_concurrently=_backend)) + before = deepcopy(batch.run.working) + + _execute_concurrent_batch(batch) + + assert called is False + assert batch.run.failure is not None + assert batch.run.failure.diagnostics[0].code == "runtime.participant-concurrent-binding-failed" + assert batch.run.failure.snapshot == before + + +@pytest.mark.parametrize("failure_call", [1, 3]) +def test_snapshot_isolation_failure_before_dispatch_restores_exact_snapshot( + monkeypatch: pytest.MonkeyPatch, + failure_call: int, +): + called = False + copy_calls = 0 + + def _backend(requests, snapshot, workers): + nonlocal called + del requests, snapshot, workers + called = True + return () + + def _failing_copy(value): + nonlocal copy_calls + copy_calls += 1 + if copy_calls == failure_call: + raise RuntimeError("snapshot copy failed") + return deepcopy(value) + + monkeypatch.setattr(scheduler_dispatch, "deepcopy", _failing_copy) + batch = _batch(SimpleNamespace(admit_actions_concurrently=_backend)) + before = deepcopy(batch.run.working) + + _execute_concurrent_batch(batch) + + assert called is False + assert batch.run.failure is not None + assert batch.run.failure.diagnostics[0].code == "runtime.participant-concurrent-snapshot-isolation-failed" + assert batch.run.failure.snapshot == before + + +def test_iterative_pass_snapshot_isolation_failure_prevents_dispatch(monkeypatch: pytest.MonkeyPatch) -> None: + called = False + + def _backend(requests, snapshot, workers): + nonlocal called + del requests, snapshot, workers + called = True + return () + + def _copy_failure(value): + del value + raise RuntimeError("snapshot copy failed") + + batch = _batch(SimpleNamespace(admit_actions_concurrently=_backend)) + policy = SimpleNamespace( + **vars(batch.policy), + profile="participant-autonomous-execution/v1", + participant_addresses=_PARTICIPANTS, + ) + monkeypatch.setattr(scheduler_concurrency, "deepcopy", _copy_failure) + + assert run_policy_due_concurrently(policy, None, batch.participant_runtime, 0, 1, batch.run) is True + assert called is False + assert batch.run.failure is not None + assert batch.run.failure.diagnostics[0].code == "runtime.participant-concurrent-snapshot-isolation-failed" + + +def test_reservation_failure_before_dispatch_restores_exact_snapshot(): + called = False + + def _backend(requests, snapshot, workers): + nonlocal called + del requests, snapshot, workers + called = True + return () + + batch = _batch(SimpleNamespace(admit_actions_concurrently=_backend)) + services = dict(batch.run.working.participant_execution_services) + services[_POLICY_ADDRESS] = _service_state(capacity=1).model_dump(mode="json") + batch.run.working = batch.run.working.with_entries( + dict(batch.run.working.entries), + participant_execution_services=services, + ) + before = deepcopy(batch.run.working) + + _execute_concurrent_batch(batch) + + assert called is False + assert batch.run.failure is not None + assert batch.run.failure.diagnostics[0].code == "runtime.participant-concurrent-reservation-failed" + assert batch.run.failure.snapshot == before + + +def test_cancelled_result_freeze_settles_every_submitted_action(): + class _CancellationDuringCopy: + def __deepcopy__(self, memo): + del memo + raise CancelledError + + batch = _batch( + SimpleNamespace( + admit_actions_concurrently=lambda requests, snapshot, workers: ( + _CancellationDuringCopy(), + _CancellationDuringCopy(), + ) + ) + ) + before = deepcopy(batch.run.working) + + _execute_concurrent_batch(batch) + + assert batch.run.failure is not None + assert batch.run.failure.diagnostics[0].code == "runtime.participant-concurrent-batch-cancelled" + _assert_indeterminate_batch_settled(batch.run, before) + + +def test_unfreezable_paired_results_fail_closed_and_settle_every_peer(): + class _UnfreezableEnvelope: + def __deepcopy__(self, memo): + del memo + raise RuntimeError("copy failed") + + batch = _batch( + SimpleNamespace( + admit_actions_concurrently=lambda requests, snapshot, workers: ( + _UnfreezableEnvelope(), + _UnfreezableEnvelope(), + ) + ) + ) + before = deepcopy(batch.run.working) + + _execute_concurrent_batch(batch) + + result = batch.run.result() + assert result.success is False + assert [diagnostic.code for diagnostic in result.diagnostics].count( + "runtime.participant-autonomous-action-protocol-invalid" + ) == 2 + _assert_indeterminate_batch_settled(batch.run, before) + + +def test_indeterminate_batch_action_ids_cannot_be_redispatched(): + calls = 0 + + def _explode(requests, snapshot, workers): + nonlocal calls + del requests, snapshot, workers + calls += 1 + raise RuntimeError("transport lost after dispatch") + + batch = _batch(SimpleNamespace(admit_actions_concurrently=_explode)) + _execute_concurrent_batch(batch) + first = batch.run.result() + policy = SimpleNamespace( + **vars(batch.policy), + profile="participant-autonomous-execution/v1", + participant_addresses=_PARTICIPANTS, + ) + repeated_run = SchedulerRunState(working=first.snapshot, diagnostics=[], changed=[]) + + handled = run_policy_due_concurrently(policy, None, batch.participant_runtime, 0, 1, repeated_run) + + assert handled is False + assert calls == 1 + + +def test_correct_length_untyped_results_fail_closed_and_settle_every_peer(): + batch = _batch(SimpleNamespace(admit_actions_concurrently=lambda requests, snapshot, workers: (object(), object()))) + + _execute_concurrent_batch(batch) + + result = batch.run.result() + assert result.success is False + assert [diagnostic.code for diagnostic in result.diagnostics].count( + "runtime.participant-autonomous-action-protocol-invalid" + ) == 2 + for address in _PARTICIPANTS: + state = ParticipantAutonomousExecutionStateModel.model_validate( + result.snapshot.participant_autonomous_execution_states[_state_key(address)] + ) + assert (state.lifecycle_state, state.attempted_actions, state.failed_actions, state.in_flight) == ( + "failed", + 1, + 1, + 0, + ) + service = ParticipantExecutionServiceStateModel.model_validate( + result.snapshot.participant_execution_services[_POLICY_ADDRESS] + ) + assert (service.reserved, service.in_flight, service.quiescent) == (0, 0, True) + + +def test_malformed_changed_address_fails_closed_at_result_boundary(): + result = ParticipantActionApplyResult(success=True, snapshot=RuntimeSnapshot()) + result.changed_addresses.append("not a compiled address") + + assert ( + _concurrent_result_protocol_invalid( + SimpleNamespace(), + result, + episode_id="episode.green", + predecessor=RuntimeSnapshot(), + ) + is True + ) + + +@pytest.mark.parametrize( + ("field_name", "value"), + [ + ("snapshot", object()), + ("success", 1), + ("diagnostics", ()), + ("diagnostics", [object()]), + ("changed_addresses", ()), + ( + "changed_addresses", + ["participant.behavior.green-user-a", "participant.behavior.green-user-a"], + ), + ], +) +def test_malformed_concurrent_result_envelope_fails_closed(field_name: str, value: object) -> None: + result = ParticipantActionApplyResult(success=True, snapshot=RuntimeSnapshot()) + object.__setattr__(result, field_name, value) + + assert ( + _concurrent_result_protocol_invalid( + SimpleNamespace(), + result, + episode_id="episode.green", + predecessor=RuntimeSnapshot(), + ) + is True + ) + + +def test_stale_generation_settles_dispatched_actions_without_committing_results( + monkeypatch: pytest.MonkeyPatch, +): + def _stale_binding(context, working, state): + del working + return SimpleNamespace( + participant_address=context.participant_address, + execution_scope_ref=_POLICY_ADDRESS, + execution_generation=0, + action_instance_id=f"{context.participant_address}:attempt-{state.attempted_actions}", + ) + + def _complete(requests, snapshot, workers): + del workers + result = ParticipantActionApplyResult( + success=True, + snapshot=snapshot, + action_result=SimpleNamespace(status="succeeded"), + ) + return tuple(result for _request in requests) + + monkeypatch.setattr(scheduler_operations, "_bound_action_request", _stale_binding) + monkeypatch.setattr(scheduler_commit, "autonomous_action_result_violation", lambda *args, **kwargs: None) + batch = _batch(SimpleNamespace(admit_actions_concurrently=_complete)) + before = deepcopy(batch.run.working) + + _execute_concurrent_batch(batch) + + result = batch.run.result() + assert [diagnostic.code for diagnostic in result.diagnostics].count( + "runtime.participant-execution-stale-completion" + ) == 2 + _assert_indeterminate_batch_settled(batch.run, before) + + +def test_missing_execution_service_fences_concurrent_completion() -> None: + request = SimpleNamespace( + execution_scope_ref="participant.execution-service.missing", + execution_generation=1, + participant_address=_PARTICIPANTS[0], + ) + + diagnostic = scheduler_commit.participant_generation_commit_diagnostic(request, RuntimeSnapshot()) + + assert diagnostic is not None + assert diagnostic.code == "runtime.participant-execution-stale-completion" + + +def test_protocol_invalid_snapshot_is_not_merged_before_validation(): + def _invalid_results(requests, snapshot, workers): + del workers + poisoned = snapshot.with_entries( + dict(snapshot.entries), + metadata={**snapshot.metadata, "backend_secret": "must-not-commit"}, + ) + result = ParticipantActionApplyResult(success=True, snapshot=poisoned, action_result=None) + return tuple(result for _request in requests) + + batch = _batch(SimpleNamespace(admit_actions_concurrently=_invalid_results)) + + _execute_concurrent_batch(batch) + + assert batch.run.result().success is False + assert "backend_secret" not in batch.run.result().snapshot.metadata + + +def test_service_accounting_changes_only_this_batch_delta(): + batch = _batch(SimpleNamespace()) + services = dict(batch.run.working.participant_execution_services) + services[_POLICY_ADDRESS] = _service_state(capacity=4, reserved=1, in_flight=1).model_dump(mode="json") + batch.run.working = batch.run.working.with_entries( + dict(batch.run.working.entries), + participant_execution_services=services, + ) + + _reserve_concurrent_actions(batch.run, tuple(batch.contexts[:1])) + reserved = ParticipantExecutionServiceStateModel.model_validate( + batch.run.working.participant_execution_services[_POLICY_ADDRESS] + ) + assert (reserved.reserved, reserved.in_flight, reserved.quiescent) == (1, 2, False) + + _finish_concurrent_service_state(batch.run, _POLICY_ADDRESS, 1) + finished = ParticipantExecutionServiceStateModel.model_validate( + batch.run.working.participant_execution_services[_POLICY_ADDRESS] + ) + assert (finished.reserved, finished.in_flight, finished.quiescent) == (1, 1, False) + + +def test_concurrent_mapping_merge_applies_backend_deletion_without_aliasing() -> None: + base = {"removed": {"secret": [1]}, "retained": {"value": [2]}} + current = deepcopy(base) + incoming = {"retained": {"value": [2]}} + + merged = _merge_mapping_revision_checked( + base=base, + current=current, + incoming=incoming, + field_name="metadata", + ) + + assert merged == {"retained": {"value": [2]}} + assert merged["retained"] is not current["retained"] + + +def test_concurrent_service_helpers_handle_absent_service_and_reject_overcompletion() -> None: + batch = _batch(SimpleNamespace()) + batch.run.working = batch.run.working.with_entries( + dict(batch.run.working.entries), + participant_execution_services={}, + ) + + _reserve_concurrent_actions(batch.run, tuple(batch.contexts[:1])) + _finish_concurrent_service_state(batch.run, _POLICY_ADDRESS, 1) + assert _available_concurrent_capacity(batch.policy, batch.run) == 0 + state = ParticipantAutonomousExecutionStateModel.model_validate( + batch.run.working.participant_autonomous_execution_states[batch.contexts[0].key] + ) + assert (state.attempted_actions, state.in_flight) == (1, 1) + + with_service = _batch(SimpleNamespace()) + with pytest.raises(ValueError, match="completion exceeds execution-service in-flight work"): + _finish_concurrent_service_state(with_service.run, _POLICY_ADDRESS, 1) + + +def test_due_scan_excludes_a_participant_with_existing_in_flight_work(): + participant = SimpleNamespace() + batch = _batch(participant) + policy = SimpleNamespace( + **vars(batch.policy), + profile="participant-autonomous-execution/v1", + participant_addresses=_PARTICIPANTS, + ) + states = dict(batch.run.working.participant_autonomous_execution_states) + states[_state_key(_PARTICIPANTS[0])] = _execution_state(_PARTICIPANTS[0], in_flight=1).model_dump(mode="json") + batch.run.working = batch.run.working.with_entries( + dict(batch.run.working.entries), + participant_autonomous_execution_states=states, + ) + + contexts, _ = scheduler_concurrency._due_contexts(policy, None, participant, 0, 1, batch.run) + + assert [context.participant_address for context in contexts] == [_PARTICIPANTS[1]] + + +def test_capacity_exhaustion_returns_explicit_failure_instead_of_silent_success( + monkeypatch: pytest.MonkeyPatch, +): + batch = _batch(object()) + services = dict(batch.run.working.participant_execution_services) + services[_POLICY_ADDRESS] = _service_state(capacity=2, in_flight=2).model_dump(mode="json") + batch.run.working = batch.run.working.with_entries( + dict(batch.run.working.entries), + participant_execution_services=services, + ) + before = deepcopy(batch.run.working) + policy = SimpleNamespace( + **vars(batch.policy), + profile="participant-autonomous-execution/v1", + participant_addresses=_PARTICIPANTS, + clock_address="time.clock.scenario-clock", + ) + monkeypatch.setattr(participant_scheduler, "_cadence", lambda policy, time_model: (0, 1)) + monkeypatch.setattr(participant_scheduler, "_clock_tick", lambda snapshot, clock_address: 0) + + participant_scheduler._run_due_policy(policy, None, object(), {}, batch.run) + + assert batch.run.failure is not None + assert batch.run.failure.diagnostics[0].code == "runtime.participant-execution-capacity-blocked" + assert batch.run.failure.snapshot == before + + +def test_concurrent_due_scan_preserves_preexisting_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + batch = _batch(object()) + policy = SimpleNamespace( + **vars(batch.policy), + profile="participant-autonomous-execution/v1", + participant_addresses=_PARTICIPANTS, + ) + scheduler_dispatch._unsupported_concurrency_failure(policy, batch.run) + original_failure = batch.run.failure + monkeypatch.setattr(scheduler_concurrency, "_due_contexts", lambda *_args: ([], [])) + + assert run_policy_due_concurrently(policy, None, object(), 0, 1, batch.run) is True + assert batch.run.failure is original_failure + + +def test_non_v1_policy_is_left_for_the_serial_scheduler() -> None: + batch = _batch(object()) + policy = SimpleNamespace(profile="participant-autonomous-execution/v2") + predecessor = batch.run.working + + handled = run_policy_due_concurrently(policy, None, object(), 0, 1, batch.run) + + assert handled is False + assert batch.run.working is predecessor + assert batch.run.failure is None + + +def test_single_available_slot_falls_back_to_serial_and_stops_after_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + batch = _batch(object()) + policy = SimpleNamespace( + **vars(batch.policy), + profile="participant-autonomous-execution/v1", + participant_addresses=_PARTICIPANTS, + ) + serial_calls: list[str] = [] + + def fail_first(context: _StubContext, run: SchedulerRunState) -> None: + serial_calls.append(context.participant_address) + scheduler_dispatch._unsupported_concurrency_failure(policy, run) + + monkeypatch.setattr( + scheduler_concurrency, + "_due_contexts", + lambda *_args: (list(batch.contexts), list(batch.states)), + ) + monkeypatch.setattr(scheduler_concurrency, "_available_concurrent_capacity", lambda *_args: 1) + monkeypatch.setattr(scheduler_operations, "run_participant_due", fail_first) + + assert run_policy_due_concurrently(policy, None, object(), 0, 1, batch.run) is True + assert serial_calls == [_PARTICIPANTS[0]] + assert batch.run.failure is not None + + +def test_single_available_slot_processes_every_due_context_serially( + monkeypatch: pytest.MonkeyPatch, +) -> None: + batch = _batch(object()) + policy = SimpleNamespace( + **vars(batch.policy), + profile="participant-autonomous-execution/v1", + participant_addresses=_PARTICIPANTS, + ) + serial_calls: list[str] = [] + monkeypatch.setattr( + scheduler_concurrency, + "_due_contexts", + lambda *_args: (list(batch.contexts), list(batch.states)), + ) + monkeypatch.setattr(scheduler_concurrency, "_available_concurrent_capacity", lambda *_args: 1) + monkeypatch.setattr( + scheduler_operations, + "run_participant_due", + lambda context, _run: serial_calls.append(context.participant_address), + ) + + assert run_policy_due_concurrently(policy, None, object(), 0, 1, batch.run) is True + assert serial_calls == list(_PARTICIPANTS) + assert batch.run.failure is None + + +def test_capacity_exhaustion_preserves_missed_cadence_failure( + monkeypatch: pytest.MonkeyPatch, +): + batch = _batch(object()) + services = dict(batch.run.working.participant_execution_services) + services[_POLICY_ADDRESS] = _service_state(capacity=2, in_flight=2).model_dump(mode="json") + batch.run.working = batch.run.working.with_entries( + dict(batch.run.working.entries), + participant_execution_services=services, + ) + policy = SimpleNamespace( + **vars(batch.policy), + profile="participant-autonomous-execution/v1", + participant_addresses=_PARTICIPANTS, + clock_address="time.clock.scenario-clock", + ) + monkeypatch.setattr(participant_scheduler, "_cadence", lambda policy, time_model: (0, 1)) + monkeypatch.setattr(participant_scheduler, "_clock_tick", lambda snapshot, clock_address: 1) + + participant_scheduler._run_due_policy(policy, None, object(), {}, batch.run) + + assert batch.run.failure is not None + assert batch.run.failure.diagnostics[0].code == "runtime.participant-autonomous-cadence-missed" + + +def test_capacity_exhaustion_is_not_a_new_failure_for_already_in_flight_work( + monkeypatch: pytest.MonkeyPatch, +): + batch = _batch(object(), in_flight=1) + services = dict(batch.run.working.participant_execution_services) + services[_POLICY_ADDRESS] = _service_state(capacity=2, in_flight=2).model_dump(mode="json") + batch.run.working = batch.run.working.with_entries( + dict(batch.run.working.entries), + participant_execution_services=services, + ) + policy = SimpleNamespace( + **vars(batch.policy), + profile="participant-autonomous-execution/v1", + participant_addresses=_PARTICIPANTS, + clock_address="time.clock.scenario-clock", + ) + monkeypatch.setattr(participant_scheduler, "_cadence", lambda policy, time_model: (0, 1)) + monkeypatch.setattr(participant_scheduler, "_clock_tick", lambda snapshot, clock_address: 0) + + participant_scheduler._run_due_policy(policy, None, object(), {}, batch.run) + + assert batch.run.failure is None + + +def test_concurrent_entrypoint_reports_zero_capacity_as_backpressure(): + batch = _batch(object()) + services = dict(batch.run.working.participant_execution_services) + services[_POLICY_ADDRESS] = _service_state(capacity=2, in_flight=2).model_dump(mode="json") + batch.run.working = batch.run.working.with_entries( + dict(batch.run.working.entries), + participant_execution_services=services, + ) + policy = SimpleNamespace( + **vars(batch.policy), + profile="participant-autonomous-execution/v1", + participant_addresses=_PARTICIPANTS, + ) + + handled = run_policy_due_concurrently(policy, None, object(), 0, 1, batch.run) + + assert handled is True + assert batch.run.failure is not None + assert batch.run.failure.diagnostics[0].code == "runtime.participant-execution-capacity-blocked" + + +def test_clock_driver_does_not_spin_on_already_in_flight_participants(): + states = { + _state_key(address): _execution_state(address, in_flight=1).model_dump(mode="json") for address in _PARTICIPANTS + } + snapshot = RuntimeSnapshot(participant_autonomous_execution_states=states) + + assert ParticipantClockDriver._next_participant_tick(snapshot, "time.clock.scenario-clock") is None + + +def test_metadata_three_way_merge_preserves_nonconflicting_prior_commit(): + base = RuntimeSnapshot(metadata={"shared": "base"}) + first = base.with_entries({}, metadata={"shared": "first"}) + current = _merge_concurrent_action_snapshot(base, base, first) + second = base.with_entries({}, metadata={"shared": "base", "second": True}) + + merged = _merge_concurrent_action_snapshot(base, current, second) + + assert merged.metadata == {"shared": "first", "second": True} + + +def test_value_three_way_merge_preserves_prior_commit_and_rejects_conflict(): + assert ( + _merge_value_revision_checked( + base="base", + current="current", + incoming="base", + field_name="time_model_state", + ) + == "current" + ) + assert ( + _merge_value_revision_checked( + base="base", + current="base", + incoming="incoming", + field_name="time_model_state", + ) + == "incoming" + ) + with pytest.raises(ValueError, match="time_model_state"): + _merge_value_revision_checked( + base="base", + current="first", + incoming="second", + field_name="time_model_state", + ) + provenance = RealizationProvenanceEntry( + address="node.green", + field_path="nodes.green.os", + domain="runtime-realization", + requirement_kind="os-family", + explicitness=ExplicitnessClass.EXACT, + provenance=ExplicitnessProvenance.AUTHOR_DECLARED, + ) + base = RuntimeSnapshot() + merged = _merge_concurrent_action_snapshot( + base, + base, + base.with_entries({}, realization_provenance=(provenance,)), + ) + assert merged.realization_provenance == (provenance,) + + +def test_metadata_conflict_diagnostic_does_not_include_backend_key(): + secret_key = "credential-value-must-not-leak" + base = RuntimeSnapshot(metadata={secret_key: "base"}) + current = _merge_concurrent_action_snapshot( + base, + base, + base.with_entries({}, metadata={secret_key: "first"}), + ) + conflicting = base.with_entries({}, metadata={secret_key: "second"}) + + with pytest.raises(ValueError) as exc_info: + _merge_concurrent_action_snapshot(base, current, conflicting) + + assert secret_key not in str(exc_info.value) + + +def test_rejected_conflicting_result_contributes_no_backend_changed_address( + monkeypatch: pytest.MonkeyPatch, +): + accepted_address = "evaluation.result.accepted" + rejected_address = "evaluation.result.rejected" + + def _conflicting_results(requests, snapshot, workers): + del requests, workers + action_result = SimpleNamespace(status="succeeded") + return ( + ParticipantActionApplyResult( + success=True, + snapshot=snapshot.with_entries({}, metadata={"shared": "first"}), + action_result=action_result, + changed_addresses=[accepted_address], + ), + ParticipantActionApplyResult( + success=True, + snapshot=snapshot.with_entries({}, metadata={"shared": "second"}), + action_result=action_result, + changed_addresses=[rejected_address], + ), + ) + + monkeypatch.setattr(scheduler_commit, "autonomous_action_result_violation", lambda *args, **kwargs: None) + batch = _batch(SimpleNamespace(admit_actions_concurrently=_conflicting_results)) + + _execute_concurrent_batch(batch) + + result = batch.run.result() + assert result.success is False + assert accepted_address in result.changed_addresses + assert rejected_address not in result.changed_addresses + + +def test_final_concurrent_materialization_failure_is_normalized(monkeypatch: pytest.MonkeyPatch) -> None: + def _complete(requests, snapshot, workers): + del workers + result = ParticipantActionApplyResult( + success=True, + snapshot=snapshot, + action_result=SimpleNamespace(status="succeeded"), + ) + return tuple(result for _request in requests) + + def _materialization_failure(snapshot): + del snapshot + raise ValueError("final validation failed") + + monkeypatch.setattr(scheduler_commit, "autonomous_action_result_violation", lambda *args, **kwargs: None) + monkeypatch.setattr(scheduler_dispatch, "_materialize_concurrent_snapshot", _materialization_failure) + batch = _batch(SimpleNamespace(admit_actions_concurrently=_complete)) + + _execute_concurrent_batch(batch) + + result = batch.run.result() + assert result.success is False + assert result.diagnostics[-1].code == "runtime.participant-concurrent-commit-invalid" + + +def test_snapshot_ownership_exhaustively_classifies_every_runtime_field(): + classified = { + *_BACKEND_MAPPING_FIELDS, + *_BACKEND_VALUE_FIELDS, + *_PROTECTED_SCHEDULER_FIELDS, + } + + assert classified == {field.name for field in dataclass_fields(RuntimeSnapshot)} + actual_protected_fields = _PROTECTED_SCHEDULER_FIELDS + assert actual_protected_fields == { + "participant_autonomous_execution_states", + "participant_execution_services", + } + + +def test_snapshot_ownership_guard_reports_missing_and_stale_fields(): + with pytest.raises(RuntimeError, match=r"missing=\['new_snapshot_field'\], stale=\[\]"): + _assert_snapshot_field_ownership( + _OWNED_SNAPSHOT_FIELDS | {"new_snapshot_field"}, + _OWNED_SNAPSHOT_FIELDS, + ) + with pytest.raises(RuntimeError, match=r"missing=\[\], stale=\['removed_snapshot_field'\]"): + _assert_snapshot_field_ownership( + _OWNED_SNAPSHOT_FIELDS, + _OWNED_SNAPSHOT_FIELDS | {"removed_snapshot_field"}, + ) + + +@pytest.mark.parametrize( + "field_name", + ["participant_autonomous_execution_states", "participant_execution_services"], +) +def test_backend_cannot_change_scheduler_owned_snapshot_fields(field_name: str): + batch = _batch(SimpleNamespace()) + base = batch.run.working + protected = deepcopy(getattr(base, field_name)) + first_key = next(iter(protected)) + changed_field = "lifecycle_state" if field_name == "participant_autonomous_execution_states" else "health" + changed_value = "paused" if field_name == "participant_autonomous_execution_states" else "degraded" + protected[first_key] = {**protected[first_key], changed_field: changed_value} + incoming = base.with_entries( + dict(base.entries), + **{field_name: protected}, + ) + + with pytest.raises(ValueError, match="protected field"): + _merge_concurrent_action_snapshot(base, base, incoming) + + +def test_default_concurrent_runtime_gives_each_worker_an_isolated_predecessor(): + class _MutatingRuntime(ParticipantExecutionRuntimeMixin): + def __init__(self) -> None: + self._barrier = threading.Barrier(2) + self._snapshot_ids: list[int] = [] + + def admit_action(self, request, snapshot): + self._snapshot_ids.append(id(snapshot)) + snapshot.metadata["seen"].append(request) + self._barrier.wait(timeout=1) + return ParticipantActionApplyResult(success=True, snapshot=snapshot) + + runtime = _MutatingRuntime() + predecessor = RuntimeSnapshot(metadata={"seen": []}) + + results = runtime.admit_actions_concurrently(("first", "second"), predecessor, 2) + + assert len(set(runtime._snapshot_ids)) == 2 + assert predecessor.metadata == {"seen": []} + assert sorted(tuple(result.snapshot.metadata["seen"]) for result in results) == [("first",), ("second",)] + + +@pytest.mark.parametrize("fault", ["raises", "leaves-live-counters"]) +def test_service_settlement_failure_is_normalized_and_restores_counters( + monkeypatch: pytest.MonkeyPatch, + fault: str, +): + def _settlement_failure(run, policy_address, completed_count): + del run, policy_address, completed_count + if fault == "raises": + raise RuntimeError("settlement failed") + + monkeypatch.setattr(scheduler_settlement, "_finish_concurrent_service_state", _settlement_failure) + batch = _batch(SimpleNamespace(admit_actions_concurrently=lambda requests, snapshot, workers: (object(), object()))) + before = deepcopy(batch.run.working) + + _execute_concurrent_batch(batch) + + result = batch.run.result() + assert result.success is False + assert any( + diagnostic.code == "runtime.participant-concurrent-service-settlement-failed" + for diagnostic in result.diagnostics + ) + _assert_indeterminate_batch_settled(batch.run, before) + + +def test_indeterminate_service_settlement_failure_is_normalized( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setattr(scheduler_settlement, "_finish_concurrent_service_state", _raise_service_settlement) + batch = _batch(SimpleNamespace(admit_actions_concurrently=lambda requests, snapshot, workers: ())) + before = deepcopy(batch.run.working) + + _execute_concurrent_batch(batch) + + result = batch.run.result() + assert {diagnostic.code for diagnostic in result.diagnostics} >= { + "runtime.participant-concurrent-result-count-invalid", + "runtime.participant-concurrent-service-settlement-failed", + } + _assert_indeterminate_batch_settled(batch.run, before) + + +def test_service_settlement_failure_removes_state_absent_before_batch( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setattr(scheduler_settlement, "_finish_concurrent_service_state", _raise_service_settlement) + run = SchedulerRunState( + working=RuntimeSnapshot( + participant_execution_services={ + _POLICY_ADDRESS: _service_state().model_dump(mode="json"), + } + ), + diagnostics=[], + changed=[], + ) + + settled = scheduler_settlement._settle_concurrent_service_state( + run, + policy_address=_POLICY_ADDRESS, + completed_count=1, + pre_batch=RuntimeSnapshot(), + ) + + assert settled is False + assert _POLICY_ADDRESS not in run.result().snapshot.participant_execution_services + + +def test_many_participants_use_one_iterative_due_scan_and_real_settlement(monkeypatch: pytest.MonkeyPatch): + participant_count = 800 + participants = tuple(f"participant.behavior.scale-{index:04d}" for index in range(participant_count)) + policy = SimpleNamespace( + address=_POLICY_ADDRESS, + profile="participant-autonomous-execution/v1", + participant_addresses=participants, + max_in_flight=2, + max_action_attempts=1, + action_contract_addresses=("participant.action-contract.green-action",), + failure_policy="continue", + clock_address="time.clock.scenario-clock", + ) + states = {_state_key(address): _execution_state(address).model_dump(mode="json") for address in participants} + run = SchedulerRunState( + working=RuntimeSnapshot( + participant_autonomous_execution_states=states, + participant_execution_services={ + _POLICY_ADDRESS: _service_state().model_dump(mode="json"), + }, + ), + diagnostics=[], + changed=[], + ) + due_scans = 0 + batch_sizes: list[int] = [] + snapshot_copy_calls = 0 + validated_state_entries = 0 + original_due_contexts = scheduler_concurrency._due_contexts + original_state_validation = runtime_state_contracts.require_participant_autonomous_state_snapshot + + def _counted_due_contexts(*args, **kwargs): + nonlocal due_scans + due_scans += 1 + return original_due_contexts(*args, **kwargs) + + def _complete_batch(requests, snapshot, workers): + assert workers == 2 + batch_sizes.append(len(requests)) + result = ParticipantActionApplyResult( + success=True, + snapshot=snapshot, + action_result=SimpleNamespace(status="succeeded"), + ) + return tuple(result for _request in requests) + + def _counted_copy(value): + nonlocal snapshot_copy_calls + if isinstance(value, RuntimeSnapshot): + snapshot_copy_calls += 1 + return deepcopy(value) + + def _counted_state_validation(states): + nonlocal validated_state_entries + validated_state_entries += len(states) + return original_state_validation(states) + + monkeypatch.setattr(scheduler_concurrency, "_due_contexts", _counted_due_contexts) + monkeypatch.setattr(scheduler_concurrency, "deepcopy", _counted_copy) + monkeypatch.setattr(scheduler_dispatch, "deepcopy", _counted_copy) + monkeypatch.setattr(scheduler_commit, "autonomous_action_result_violation", lambda *args, **kwargs: None) + monkeypatch.setattr( + runtime_state_contracts, + "require_participant_autonomous_state_snapshot", + _counted_state_validation, + ) + + handled = run_policy_due_concurrently( + policy, + None, + SimpleNamespace(admit_actions_concurrently=_complete_batch), + 0, + 1, + run, + ) + + assert handled is True + assert due_scans == 1 + assert batch_sizes == [2] * (participant_count // 2) + assert snapshot_copy_calls == 1 + 2 * len(batch_sizes) + assert validated_state_entries == participant_count + assert all( + ( + payload["lifecycle_state"], + payload["attempted_actions"], + payload["succeeded_actions"], + payload["in_flight"], + ) + == ("completed", 1, 1, 0) + for payload in run.result().snapshot.participant_autonomous_execution_states.values() + ) + service = ParticipantExecutionServiceStateModel.model_validate( + run.result().snapshot.participant_execution_services[_POLICY_ADDRESS] + ) + assert (service.reserved, service.in_flight, service.quiescent) == (0, 0, True)