fix(runtime): settle concurrent participant batches reliably

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.

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)

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)

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,

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
 

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)