From 767b70f52ea144f6128dfd148818474c028a7a2d Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 17:05:04 -0700 Subject: [PATCH 01/13] fix(runtime): close control-plane request-size and bearer-auth gaps Three defects in the HTTP/JSON control-plane adapter: - The request-size guard buffered the whole body via `request.body()` before measuring it. A request with no declared `content-length` (e.g. `Transfer-Encoding: chunked`) is invisible to the content-length guard, so an unbounded upload could exhaust memory even though the response was still 413. The body is now accumulated incrementally and rejected as soon as the running total exceeds the cap; Starlette's body cache is seeded so downstream parsing is unchanged. - A presented-but-unresolvable bearer token fell through to the proxy-header path instead of failing. Where header identities are trusted, a revoked or bogus token kept authenticating and the rejected credential never reached the audit log. An unresolvable token is now a 401. - The bearer path returned its identity without the target-binding check the header path applies, so a token scoped to one target authenticated against another control plane. Both paths now share `_require_target_binding`. Also compare tokens in constant time over encoded bytes (a non-ASCII token previously raised instead of reporting unauthorized), and make `ControlPlaneSecurityConfig` mappings read-only so `strict_defaults()` cannot be granted principals or tokens after construction. Co-Authored-By: Claude Opus 5 (1M context) --- .../raes_runtime/control_plane_api/_auth.py | 30 ++- .../raes_runtime/control_plane_api_guards.py | 20 +- .../raes_runtime/control_plane_security.py | 14 +- .../tests/test_runtime_control_plane_api.py | 172 ++++++++++++++++++ 4 files changed, 227 insertions(+), 9 deletions(-) diff --git a/implementations/python/packages/raes_runtime/control_plane_api/_auth.py b/implementations/python/packages/raes_runtime/control_plane_api/_auth.py index e848ef0c6..de41f9100 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api/_auth.py +++ b/implementations/python/packages/raes_runtime/control_plane_api/_auth.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hmac from typing import Annotated from fastapi import Depends, HTTPException, Request @@ -54,9 +55,14 @@ def _authenticate_request(self, request: Request) -> ControlPlaneIdentity: authorization = request.headers.get("authorization", "") if authorization.lower().startswith("bearer "): token = authorization.split(" ", 1)[1].strip() - identity = self._security.bearer_tokens.get(token) - if identity is not None: - return identity + identity = self._resolve_bearer_identity(token) + # A presented-but-unresolvable token is a hard failure: falling through + # to the proxy-header path would let a revoked or bogus token keep + # working whenever header identities are trusted, and would leave the + # rejected credential out of the audit log entirely. + if identity is None: + raise HTTPException(status_code=401, detail="invalid bearer token") + return self._require_target_binding(identity) if not self._security.trust_proxy_identity_headers: raise HTTPException(status_code=401, detail="trusted proxy identity headers are not enabled") identity_name = request.headers.get(self._security.identity_header, "") @@ -66,6 +72,24 @@ def _authenticate_request(self, request: Request) -> ControlPlaneIdentity: identity = self._security.trusted_identities.get(identity_name) if identity is None: raise HTTPException(status_code=401, detail="unknown client identity") + return self._require_target_binding(identity) + + def _resolve_bearer_identity(self, token: str) -> ControlPlaneIdentity | None: + """Resolve a bearer token without leaking which token matched via timing.""" + + # Compare encoded bytes: ``compare_digest`` rejects non-ASCII ``str``, so a + # non-ASCII token would raise instead of being reported as unauthorized. + # The loop never breaks early, so timing does not reveal which token matched. + presented = token.encode("utf-8") + matched: ControlPlaneIdentity | None = None + for candidate, identity in self._security.bearer_tokens.items(): + if hmac.compare_digest(candidate.encode("utf-8"), presented): + matched = identity + return matched + + def _require_target_binding(self, identity: ControlPlaneIdentity) -> ControlPlaneIdentity: + """Reject an identity scoped to a different target than this control plane.""" + if identity.target_name and identity.target_name != self._control_plane.target_name: raise HTTPException(status_code=403, detail="identity is not authorized for this target") return identity diff --git a/implementations/python/packages/raes_runtime/control_plane_api_guards.py b/implementations/python/packages/raes_runtime/control_plane_api_guards.py index ced8ebf33..ea8207d04 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api_guards.py +++ b/implementations/python/packages/raes_runtime/control_plane_api_guards.py @@ -56,10 +56,22 @@ async def _body_size_guard_response( *, max_request_bytes: int, ) -> JSONResponse | None: - body = await request.body() - if len(body) > max_request_bytes: - return _request_too_large_response(control_plane, request) - request.state.raw_body = body + # Accumulate the body incrementally and stop as soon as the running total + # exceeds the limit. Buffering via ``request.body()`` would read the whole + # payload first, so a request without a declared ``content-length`` (e.g. + # ``Transfer-Encoding: chunked``) bypasses ``_content_length_guard_response`` + # and could exhaust memory before any size check runs. + body = bytearray() + async for chunk in request.stream(): + body.extend(chunk) + if len(body) > max_request_bytes: + return _request_too_large_response(control_plane, request) + accepted_body = bytes(body) + # Streaming consumes the receive channel, so seed Starlette's body cache the + # way ``Request.body()`` would. Route handlers and FastAPI's own body parsing + # then still see the payload instead of an exhausted stream. + request._body = accepted_body # noqa: SLF001 + request.state.raw_body = accepted_body return None diff --git a/implementations/python/packages/raes_runtime/control_plane_security.py b/implementations/python/packages/raes_runtime/control_plane_security.py index a2f08481c..100aed53d 100644 --- a/implementations/python/packages/raes_runtime/control_plane_security.py +++ b/implementations/python/packages/raes_runtime/control_plane_security.py @@ -2,8 +2,10 @@ from __future__ import annotations +from collections.abc import Mapping from dataclasses import dataclass, field from enum import Enum +from types import MappingProxyType class ControlPlaneRole(str, Enum): @@ -62,8 +64,16 @@ class ControlPlaneSecurityConfig: identity_header: str = "x-raes-client-identity" trust_proxy_identity_headers: bool = False max_request_bytes: int = 1_000_000 - trusted_identities: dict[str, ControlPlaneIdentity] = field(default_factory=dict) - bearer_tokens: dict[str, ControlPlaneIdentity] = field(default_factory=dict) + trusted_identities: Mapping[str, ControlPlaneIdentity] = field(default_factory=dict) + bearer_tokens: Mapping[str, ControlPlaneIdentity] = field(default_factory=dict) + + def __post_init__(self) -> None: + # ``frozen=True`` only blocks rebinding the attributes; a caller (or a + # later code path) could still mutate the underlying dicts and grant + # principals or tokens after construction, defeating ``strict_defaults``. + # Read-only proxies keep dict equality while blocking that mutation. + object.__setattr__(self, "trusted_identities", MappingProxyType(dict(self.trusted_identities))) + object.__setattr__(self, "bearer_tokens", MappingProxyType(dict(self.bearer_tokens))) @classmethod def strict_defaults(cls) -> ControlPlaneSecurityConfig: diff --git a/implementations/python/tests/test_runtime_control_plane_api.py b/implementations/python/tests/test_runtime_control_plane_api.py index f973d2b07..8aa190ea9 100644 --- a/implementations/python/tests/test_runtime_control_plane_api.py +++ b/implementations/python/tests/test_runtime_control_plane_api.py @@ -2,11 +2,13 @@ from __future__ import annotations +import asyncio import textwrap from pathlib import Path import pytest import raes_runtime.control_plane_store as control_plane_store_module +from fastapi import HTTPException from raes import parse_sdl from raes_backend_stubs.stubs import create_stub_target from raes_contracts.contracts import ( @@ -26,12 +28,15 @@ from raes_processor.planner import plan from raes_runtime.control_plane import RuntimeControlPlane from raes_runtime.control_plane_api import create_control_plane_app +from raes_runtime.control_plane_api._auth import _ControlPlaneApiAuth +from raes_runtime.control_plane_api_guards import request_size_guard_response from raes_runtime.control_plane_security import ( ControlPlaneIdentity, ControlPlaneRole, ControlPlaneSecurityConfig, ) from raes_runtime.control_plane_store import ControlPlaneOperationRecord, LocalControlPlaneStore +from starlette.requests import Request from starlette.testclient import TestClient @@ -574,6 +579,173 @@ def test_control_plane_api_rejects_invalid_content_length_header(): assert control_plane.audit_log()[-1].reason == "invalid content-length" +def test_control_plane_api_enforces_request_size_limit_without_content_length(): + """A chunked body (no content-length) must still be rejected with 413.""" + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + security = _test_security(target.name, max_request_bytes=32) + app = create_control_plane_app(control_plane, security=security) + headers = { + "x-raes-client-verified": "true", + "x-raes-client-identity": "backend-service", + "content-type": "application/json", + } + + def _chunked_body(): + for _ in range(100): + yield b"x" * 32 + + with TestClient(app) as client: + response = client.post( + "/operations/provisioning", + content=_chunked_body(), + headers=headers, + ) + + assert response.status_code == 413 + assert response.json() == {"detail": "request too large"} + assert control_plane.audit_log()[-1].reason == "request too large" + + +def test_control_plane_api_rejects_invalid_bearer_token_instead_of_trusting_headers(): + """An unresolvable bearer token must fail closed, not fall through to header identity.""" + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + app = create_control_plane_app( + control_plane, + security=_test_security(target.name), + ) + + with TestClient(app) as client: + response = client.get( + "/snapshot", + headers={ + "authorization": "Bearer revoked-token", + "x-raes-client-verified": "true", + "x-raes-client-identity": "backend-service", + }, + ) + + assert response.status_code == 401 + assert response.json() == {"detail": "invalid bearer token"} + assert control_plane.audit_log()[-1].reason == "invalid bearer token" + assert control_plane.audit_log()[-1].allowed is False + + +def test_control_plane_auth_rejects_non_ascii_bearer_token_as_unauthorized(): + """A non-ASCII token must be reported unauthorized, not crash the comparison. + + Starlette decodes header bytes as latin-1, so a raw request can deliver a + non-ASCII token string even though HTTP clients refuse to encode one. A + ``str``-based constant-time comparison would raise ``TypeError`` there and + surface as a 500 instead of a 401. + """ + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + auth = _ControlPlaneApiAuth(control_plane, _test_security(target.name)) + request = Request( + { + "type": "http", + "method": "GET", + "path": "/snapshot", + "headers": [(b"authorization", "Bearer token-\xf6\xe9".encode("latin-1"))], + "query_string": b"", + }, + ) + + with pytest.raises(HTTPException) as excinfo: + auth.read_identity(request) + + assert excinfo.value.status_code == 401 + assert excinfo.value.detail == "invalid bearer token" + + +def test_control_plane_api_rejects_bearer_token_bound_to_another_target(): + """A bearer token scoped to a different target must not authenticate here. + + The header-identity path already enforces this binding; the bearer path must + apply the same check rather than returning the identity unconditionally. + """ + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + security = ControlPlaneSecurityConfig( + trust_proxy_identity_headers=False, + bearer_tokens={ + "other-target-token": ControlPlaneIdentity( + identity="operator", + roles=frozenset({ControlPlaneRole.OPERATOR}), + target_name="some-other-target", + ), + }, + ) + app = create_control_plane_app(control_plane, security=security) + + with TestClient(app) as client: + response = client.get( + "/snapshot", + headers={"authorization": "Bearer other-target-token"}, + ) + + assert response.status_code == 403 + assert response.json() == {"detail": "identity is not authorized for this target"} + + +def test_control_plane_security_config_mappings_cannot_be_mutated_after_construction(): + """``strict_defaults`` must stay fail-closed; frozen=True alone does not stop dict mutation.""" + security = ControlPlaneSecurityConfig.strict_defaults() + intruder = ControlPlaneIdentity(identity="intruder", roles=frozenset({ControlPlaneRole.OPERATOR})) + + with pytest.raises(TypeError): + security.bearer_tokens["stolen"] = intruder # type: ignore[index] + with pytest.raises(TypeError): + security.trusted_identities["stolen"] = intruder # type: ignore[index] + + assert security.bearer_tokens == {} + assert security.trusted_identities == {} + + +def test_request_size_guard_stops_reading_an_oversized_chunked_body(): + """The body guard must cap while streaming, not buffer the whole payload first. + + A request without ``content-length`` (e.g. ``Transfer-Encoding: chunked``) is + invisible to the content-length guard, so the body guard is the only limit. If + it buffers the full body before measuring it, an unbounded chunked upload can + exhaust memory even though the response is still 413. + """ + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + chunk = b"x" * 32 + total_chunks = 1000 + delivered = 0 + + async def receive() -> dict[str, object]: + nonlocal delivered + if delivered >= total_chunks: + return {"type": "http.request", "body": b"", "more_body": False} + delivered += 1 + return {"type": "http.request", "body": chunk, "more_body": True} + + request = Request( + { + "type": "http", + "method": "POST", + "path": "/operations/provisioning", + "headers": [(b"content-type", b"application/json")], + "query_string": b"", + }, + receive=receive, + ) + + response = asyncio.run( + request_size_guard_response(control_plane, request, max_request_bytes=64), + ) + + assert response is not None + assert response.status_code == 413 + # Rejected after crossing the cap, not after draining all 1000 chunks. + assert delivered <= 3 + + def test_local_control_plane_store_saves_snapshot_with_atomic_replace( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, From ec2354795c02c202b8eb5e03e6f1dc9a77c35249 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 17:11:30 -0700 Subject: [PATCH 02/13] fix(runtime): stop swallowing timestamp errors in workflow timeout reconciliation `_workflow_has_timed_out` wrapped both timestamp parses in `except Exception: return False`, so any unparseable value reported "not timed out". A RUNNING workflow whose recorded `started_at` could not be parsed therefore had no derivable deadline and stayed RUNNING for the lifetime of the control plane: reconciliation could never reclaim it, even decades past a one-second timeout. The two timestamps have different scope, so they are now handled differently: - `submitted_at` is the caller's reconciliation clock and governs every workflow in the pass, so an unusable value raises instead of quietly disabling all timeouts. The HTTP adapter already maps `ValueError` to 409 for this route. - A per-workflow `started_at` that cannot be parsed no longer blocks reclamation; the workflow is timed out under a distinct terminal reason so it stays diagnosable rather than looking like an ordinary timeout. Co-Authored-By: Claude Opus 5 (1M context) --- .../raes_runtime/control_plane_timeouts.py | 40 +++++--- ...runtime_workflow_timeout_reconciliation.py | 91 +++++++++++++++++++ 2 files changed, 118 insertions(+), 13 deletions(-) create mode 100644 implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py diff --git a/implementations/python/packages/raes_runtime/control_plane_timeouts.py b/implementations/python/packages/raes_runtime/control_plane_timeouts.py index 6d9f0dfc1..ec1aaa537 100644 --- a/implementations/python/packages/raes_runtime/control_plane_timeouts.py +++ b/implementations/python/packages/raes_runtime/control_plane_timeouts.py @@ -14,6 +14,9 @@ from .control_plane_workflows import maybe_apply_compensation, parse_timestamp +TIMED_OUT_REASON = "workflow timed out" +UNPARSEABLE_START_REASON = "workflow timed out: started_at could not be parsed" + def workflow_timeout_update( snapshot: RuntimeSnapshot, @@ -26,11 +29,10 @@ def workflow_timeout_update( update = None timeout_seconds = _eligible_workflow_timeout_seconds(entry) normalized = _running_workflow_result(orchestration_results.get(workflow_address)) - if ( - timeout_seconds is not None - and normalized is not None - and _workflow_has_timed_out(normalized, timeout_seconds, submitted_at) - ): + terminal_reason = None + if timeout_seconds is not None and normalized is not None: + terminal_reason = _workflow_timeout_reason(normalized, timeout_seconds, submitted_at) + if terminal_reason is not None: update = _timed_out_workflow_update( snapshot, workflow_address, @@ -38,6 +40,7 @@ def workflow_timeout_update( timeout_seconds, orchestration_history, submitted_at, + terminal_reason, ) return update @@ -77,17 +80,26 @@ def _coerce_timeout_seconds(raw: object) -> int | None: return timeout -def _workflow_has_timed_out( +def _workflow_timeout_reason( normalized: WorkflowExecutionState, timeout_seconds: int, submitted_at: str, -) -> bool: +) -> str | None: + """Return the terminal reason when the workflow must time out, else ``None``. + + ``submitted_at`` is the caller's reconciliation clock and governs the whole + pass, so an unusable value is raised rather than quietly disabling every + timeout. A running workflow whose own ``started_at`` cannot be parsed has no + derivable deadline; reporting "not timed out" would pin it in RUNNING + forever, so it is reclaimed under a distinct reason instead. + """ + + current = parse_timestamp(submitted_at).timestamp() try: deadline = parse_timestamp(normalized.started_at).timestamp() + timeout_seconds - current = parse_timestamp(submitted_at).timestamp() - except Exception: - return False - return current >= deadline + except (TypeError, ValueError): + return UNPARSEABLE_START_REASON + return TIMED_OUT_REASON if current >= deadline else None def _timed_out_workflow_update( @@ -97,8 +109,9 @@ def _timed_out_workflow_update( timeout_seconds: int, orchestration_history: dict[str, list[dict[str, object]]], submitted_at: str, + terminal_reason: str, ) -> tuple[dict[str, object], list[dict[str, object]]]: - timed_out_state = _timed_out_workflow_state(normalized, submitted_at) + timed_out_state = _timed_out_workflow_state(normalized, submitted_at, terminal_reason) history = orchestration_history.setdefault(workflow_address, []) history.append( WorkflowHistoryEvent( @@ -119,6 +132,7 @@ def _timed_out_workflow_update( def _timed_out_workflow_state( normalized: WorkflowExecutionState, submitted_at: str, + terminal_reason: str, ) -> WorkflowExecutionState: return WorkflowExecutionState( state_schema_version=normalized.state_schema_version, @@ -126,7 +140,7 @@ def _timed_out_workflow_state( run_id=normalized.run_id, started_at=normalized.started_at, updated_at=submitted_at, - terminal_reason="workflow timed out", + terminal_reason=terminal_reason, compensation_status=WorkflowCompensationStatus.NOT_REQUIRED, compensation_started_at=None, compensation_updated_at=None, diff --git a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py new file mode 100644 index 000000000..fe475dd6c --- /dev/null +++ b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py @@ -0,0 +1,91 @@ +"""Workflow timeout reconciliation edge cases for the runtime control plane.""" + +from __future__ import annotations + +import pytest +from raes_contracts.planning import RuntimeDomain +from raes_contracts.runtime_state import RuntimeSnapshot, SnapshotEntry +from raes_contracts.workflow import WorkflowExecutionState, WorkflowStatus +from raes_runtime.control_plane_timeouts import ( + TIMED_OUT_REASON, + UNPARSEABLE_START_REASON, + workflow_timeout_update, +) + +_WORKFLOW_ADDRESS = "orchestration.workflow.response" + + +def _workflow_entry(timeout_seconds: int = 1) -> SnapshotEntry: + return SnapshotEntry( + address=_WORKFLOW_ADDRESS, + domain=RuntimeDomain.ORCHESTRATION, + resource_type="workflow", + payload={"execution_contract": {"timeout_seconds": timeout_seconds}}, + ) + + +def _running_result(started_at: str) -> dict[str, object]: + """Build a persisted RUNNING workflow payload with ``started_at`` as recorded. + + The payload is edited after construction because the model rejects values it + considers unusable, while reconciliation reads snapshots back through + ``WorkflowExecutionState.from_payload`` and must cope with whatever the store + actually holds. + """ + + payload = WorkflowExecutionState( + workflow_status=WorkflowStatus.RUNNING, + run_id="run-1", + started_at="2000-01-01T00:00:00Z", + updated_at="2000-01-01T00:00:00Z", + ).to_payload() + payload["started_at"] = started_at + return payload + + +def _reconcile(started_at: str, submitted_at: str) -> tuple[dict[str, object], list[dict[str, object]]] | None: + return workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + _workflow_entry(), + {_WORKFLOW_ADDRESS: _running_result(started_at)}, + {}, + submitted_at, + ) + + +def test_expired_workflow_is_timed_out(): + update = _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:01:00Z") + + assert update is not None + assert update[0]["workflow_status"] == WorkflowStatus.TIMED_OUT.value + assert update[0]["terminal_reason"] == TIMED_OUT_REASON + + +def test_workflow_inside_its_deadline_is_left_running(): + assert _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:00:00Z") is None + + +@pytest.mark.parametrize("started_at", ["None", "not-a-timestamp", "2000-13-45T99:99:99Z"]) +def test_workflow_with_unparseable_started_at_is_reclaimed(started_at: str): + """A running workflow with no derivable deadline must not stay RUNNING forever. + + Swallowing the parse failure and reporting "not timed out" pinned such a + workflow in RUNNING for the lifetime of the control plane, so reconciliation + could never reclaim it. + """ + update = _reconcile(started_at, "2000-01-01T00:01:00Z") + + assert update is not None + assert update[0]["workflow_status"] == WorkflowStatus.TIMED_OUT.value + assert update[0]["terminal_reason"] == UNPARSEABLE_START_REASON + + +def test_unparseable_reconciliation_clock_is_raised_not_swallowed(): + """A bad caller-supplied ``now`` governs every workflow, so it must surface. + + Reported as ``ValueError``; the HTTP adapter maps that to 409 rather than + silently disabling timeouts for the whole pass. + """ + with pytest.raises(ValueError): + _reconcile("2000-01-01T00:00:00Z", "not-a-timestamp") From 20a7e0e2c45d0f59501dbcea8e2c4191b89fce91 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 17:17:36 -0700 Subject: [PATCH 03/13] fix(runtime): release participant reservations when a concurrent batch fails `_execute_concurrent_batch` reserved capacity (bumping `attempted_actions` and `in_flight`, clearing `quiescent`) before calling the backend batch method, but only `_commit_concurrent_result` clears a participant's `in_flight`. Two paths escaped before any result was committed: - a backend returning the wrong number of results raised `ValueError` - a backend raising propagated straight out of the scheduler Both left every selected participant permanently in-flight and the service non-quiescent, so no later occurrence could be admitted for them, and neither reached the scheduler's diagnostic channel. Both are now reported the way this layer reports other backend conformance failures, via `_set_concurrent_failure`, after releasing the reservations. The abandoned attempt is settled as failed rather than erased: the backend was asked to run it, so it happened and did not succeed, and that keeps the snapshot invariant `attempted_actions == succeeded + failed + in_flight` intact. Co-Authored-By: Claude Opus 5 (1M context) --- .../participant_scheduler_concurrency.py | 73 +++++++- ...rticipant_concurrent_batch_reservations.py | 163 ++++++++++++++++++ 2 files changed, 233 insertions(+), 3 deletions(-) create mode 100644 implementations/python/tests/test_participant_concurrent_batch_reservations.py diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py index 901cac398..de907761e 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py @@ -138,6 +138,44 @@ def _reserve_concurrent_actions( ) +def _release_concurrent_reservations( + run: SchedulerRunState, + contexts: tuple[_DueActionContext, ...], + policy_address: str, +) -> None: + """Undo ``_reserve_concurrent_actions`` when the batch never produced results. + + Reservations are taken before the backend batch call, and only + ``_commit_concurrent_result`` clears a participant's ``in_flight``. Abandoning + the batch without releasing them would leave participants permanently + in-flight and the service non-quiescent, so no later occurrence can be + admitted for them. + + The reserved attempt is settled as failed rather than erased: the backend was + asked to run it, so it did happen and did not succeed. That also preserves the + snapshot invariant ``attempted_actions == succeeded + failed + in_flight``, + which a bare ``in_flight`` reset would break. + """ + + states = dict(run.working.participant_autonomous_execution_states) + for context in contexts: + payload = states.get(context.key) + if payload is None: + continue + state = ParticipantAutonomousExecutionStateModel.model_validate(payload) + states[context.key] = state.model_copy( + update={ + "in_flight": 0, + "failed_actions": state.failed_actions + state.in_flight, + } + ).model_dump(mode="json") + run.working = run.working.with_entries( + dict(run.working.entries), + participant_autonomous_execution_states=states, + ) + _finish_concurrent_service_state(run, policy_address) + + def _finish_concurrent_service_state( run: SchedulerRunState, policy_address: str, @@ -350,9 +388,38 @@ def _execute_concurrent_batch(batch: _ConcurrentBatch) -> None: ) _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") + # The batch method is backend-supplied. A raising or miscounting backend is a + # conformance failure to report, not an exception to leak through the + # scheduler, and either way the reservations taken above must be released. + try: + results = batch_method(requests, base, len(requests)) + result_count = len(results) + except Exception as exc: # noqa: BLE001 - backend trust boundary + _release_concurrent_reservations(batch.run, selected_contexts, batch.policy.address) + _set_concurrent_failure( + batch.run, + Diagnostic( + code="runtime.participant-concurrent-batch-failed", + domain="participant", + address=batch.policy.address, + message=f"Backend concurrent participant batch raised: {exc}", + ), + ) + return + if result_count != len(requests): + _release_concurrent_reservations(batch.run, selected_contexts, batch.policy.address) + _set_concurrent_failure( + batch.run, + Diagnostic( + code="runtime.participant-concurrent-result-count-invalid", + domain="participant", + address=batch.policy.address, + message=( + f"Backend returned {result_count} concurrent participant results for {len(requests)} requests." + ), + ), + ) + return 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: 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 000000000..be1ae885d --- /dev/null +++ b/implementations/python/tests/test_participant_concurrent_batch_reservations.py @@ -0,0 +1,163 @@ +"""Reservation release when a backend concurrent participant batch misbehaves.""" + +from __future__ import annotations + +from dataclasses import dataclass +from types import SimpleNamespace + +import pytest +import raes_runtime.participant_scheduler_operations as scheduler_operations +from raes_contracts.contracts import ParticipantAutonomousExecutionStateModel +from raes_contracts.contracts.participant_execution import ParticipantExecutionServiceStateModel +from raes_contracts.runtime_state import RuntimeSnapshot +from raes_runtime.participant_scheduler_concurrency import _execute_concurrent_batch +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 + + +def _execution_state(participant_address: str) -> 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=0, + succeeded_actions=0, + failed_actions=0, + ) + + +def _service_state() -> 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=True, + resources_released=False, + policy_digest=digest, + binding_digest=digest, + time_declaration_digest=digest, + capacity=2, + reserved=0, + in_flight=0, + 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) -> SimpleNamespace: + policy = SimpleNamespace(address=_POLICY_ADDRESS, max_in_flight=2) + states = [_execution_state(address) 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().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), + ) + + +def _assert_reservations_released(run: SchedulerRunState) -> None: + service = ParticipantExecutionServiceStateModel.model_validate( + run.working.participant_execution_services[_POLICY_ADDRESS] + ) + assert (service.in_flight, service.reserved, service.quiescent) == (0, 0, True) + for address in _PARTICIPANTS: + state = ParticipantAutonomousExecutionStateModel.model_validate( + run.working.participant_autonomous_execution_states[_state_key(address)] + ) + # The reserved attempt is settled as failed, preserving + # attempted_actions == succeeded + failed + in_flight. + assert state.in_flight == 0 + assert state.attempted_actions == state.succeeded_actions + state.failed_actions + state.in_flight + + +def test_miscounted_backend_batch_is_reported_and_releases_reservations(): + """A wrong result count previously raised, leaking every reservation. + + Reservations are taken before the backend call and only cleared per + committed result, so escaping here left participants permanently in-flight + and the service non-quiescent, blocking all later occurrences. + """ + batch = _batch(SimpleNamespace(admit_actions_concurrently=lambda requests, snapshot, workers: ())) + + _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_reservations_released(batch.run) + + +def test_raising_backend_batch_is_reported_and_releases_reservations(): + """A raising backend is a conformance failure, not an exception to leak.""" + + def _explode(requests, snapshot, workers): + raise RuntimeError("backend exploded") + + batch = _batch(SimpleNamespace(admit_actions_concurrently=_explode)) + + _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 + _assert_reservations_released(batch.run) From 6bf453e46bfbe91eb49c3f0bd5af6f15a4b45638 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 17:23:39 -0700 Subject: [PATCH 04/13] fix(processor): detect dependency cycles without recursing per resource `dependency_cycles` ran Tarjan recursively, so DFS depth tracked the longest dependency path. A scenario with roughly a thousand or more chained resources raised `RecursionError` out of `_ordering_cycle_diagnostics`, aborting planning instead of reporting ordering cycles; the iterative `topological_dependency_order` beside it handled the same graph. Measured: the recursive walk dies between 900 and 2000 chained resources, while a 5000-node chain and a 3000-node cycle now both resolve. The walk is driven from an explicit frame stack. Emitted cycles are unchanged: a differential property test compares it against the recursive semantics it replaces over graphs containing self-loops and multi-node cycles. Also drops `_ordering_graph`, which was unreferenced anywhere in the repository. Co-Authored-By: Claude Opus 5 (1M context) --- .../raes_processor/planner/ordering.py | 6 - .../raes_processor/semantics/planner.py | 42 +++++-- .../python/tests/test_semantics_planner.py | 109 ++++++++++++++++++ 3 files changed, 139 insertions(+), 18 deletions(-) diff --git a/implementations/python/packages/raes_processor/planner/ordering.py b/implementations/python/packages/raes_processor/planner/ordering.py index 52e4e1cfb..7ef6cf23a 100644 --- a/implementations/python/packages/raes_processor/planner/ordering.py +++ b/implementations/python/packages/raes_processor/planner/ordering.py @@ -2,8 +2,6 @@ from ..models import Diagnostic, PlannedResource, RuntimeDomain, SnapshotEntry from ..semantics.planner import ( - DependencyKind, - dependency_graph_for_resources, resource_delete_order, resource_dependency_cycles, resource_topological_order, @@ -12,10 +10,6 @@ from ..semantics.realization_snapshot_sanitization import realization_payloads_match -def _ordering_graph(resources: dict[str, PlannedResource]) -> dict[str, tuple[str, ...]]: - return dependency_graph_for_resources(resources, kind=DependencyKind.ORDERING) - - def _ordering_cycles(resources: dict[str, PlannedResource]) -> list[tuple[str, ...]]: return resource_dependency_cycles(resources) diff --git a/implementations/python/packages/raes_processor/semantics/planner.py b/implementations/python/packages/raes_processor/semantics/planner.py index b246a16d4..26ff9f6e1 100644 --- a/implementations/python/packages/raes_processor/semantics/planner.py +++ b/implementations/python/packages/raes_processor/semantics/planner.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections import deque -from collections.abc import Callable, Iterable, Mapping +from collections.abc import Callable, Iterable, Iterator, Mapping from dataclasses import dataclass from enum import Enum from typing import Protocol, TypeVar @@ -123,7 +123,7 @@ def dependency_cycles( on_stack: set[str] = set() cycles: list[tuple[str, ...]] = [] - def strongconnect(node: str) -> None: + def visit(node: str) -> None: nonlocal index indices[node] = index lowlinks[node] = index @@ -131,16 +131,7 @@ def strongconnect(node: str) -> None: stack.append(node) on_stack.add(node) - for dependency in graph[node]: - if dependency not in indices: - strongconnect(dependency) - lowlinks[node] = min(lowlinks[node], lowlinks[dependency]) - elif dependency in on_stack: - lowlinks[node] = min(lowlinks[node], indices[dependency]) - - if lowlinks[node] != indices[node]: - return - + def close_component(node: str) -> None: component: list[str] = [] while stack: member = stack.pop() @@ -153,6 +144,33 @@ def strongconnect(node: str) -> None: if len(component) > 1 or component[0] in graph[component[0]]: cycles.append(tuple(component)) + # Tarjan is driven from an explicit stack rather than recursion: DFS depth + # reaches the longest dependency path, so a recursive walk raises + # RecursionError on scenarios with more than roughly a thousand chained + # resources, aborting planning instead of reporting cycles. + def strongconnect(root: str) -> None: + visit(root) + frames: list[tuple[str, Iterator[str]]] = [(root, iter(graph[root]))] + while frames: + node, dependencies = frames[-1] + descended = False + for dependency in dependencies: + if dependency not in indices: + visit(dependency) + frames.append((dependency, iter(graph[dependency]))) + descended = True + break + if dependency in on_stack: + lowlinks[node] = min(lowlinks[node], indices[dependency]) + if descended: + continue + frames.pop() + if frames: + parent = frames[-1][0] + lowlinks[parent] = min(lowlinks[parent], lowlinks[node]) + if lowlinks[node] == indices[node]: + close_component(node) + for node in sorted(graph, key=canonical_resource_identity): if node not in indices: strongconnect(node) diff --git a/implementations/python/tests/test_semantics_planner.py b/implementations/python/tests/test_semantics_planner.py index 4f74ba0dd..6055fe34c 100644 --- a/implementations/python/tests/test_semantics_planner.py +++ b/implementations/python/tests/test_semantics_planner.py @@ -2,16 +2,21 @@ from __future__ import annotations +import itertools from types import SimpleNamespace from hypothesis import given from hypothesis import strategies as st from raes_processor.semantics.planner import ( DependencyKind, + canonical_resource_identity, + dependency_cycles, dependency_edges, + dependency_graph, refresh_impacted_nodes, resource_delete_order, resource_topological_order, + topological_dependency_order, ) @@ -81,6 +86,74 @@ def _dag_resources_with_change_sets(draw): return resources, subset_a, subset_b +def _dependency_graphs() -> st.SearchStrategy[dict[str, tuple[str, ...]]]: + """Small graphs that freely admit self-loops and multi-node cycles.""" + + def _build(size: int, choices: list[list[int]]) -> dict[str, tuple[str, ...]]: + nodes = [f"nodes.host-{index}" for index in range(size)] + return { + node: tuple(nodes[target % size] for target in targets) + for node, targets in zip(nodes, choices, strict=True) + } + + return st.integers(min_value=1, max_value=8).flatmap( + lambda size: st.lists( + st.lists(st.integers(min_value=0, max_value=7), max_size=3), + min_size=size, + max_size=size, + ).map(lambda choices: _build(size, choices)) + ) + + +def _reference_dependency_cycles( + dependencies_by_node: dict[str, tuple[str, ...]], +) -> list[tuple[str, ...]]: + """Recursive Tarjan reference for differential comparison. + + Kept deliberately naive: it mirrors the textbook recursion the production + walk replaced, so any behavioural drift in the explicit-stack version shows + up as a mismatch rather than as a silently different plan order. + """ + + graph = dependency_graph(dependencies_by_node) + if not graph: + return [] + counter = itertools.count() + indices: dict[str, int] = {} + lowlinks: dict[str, int] = {} + stack: list[str] = [] + on_stack: set[str] = set() + cycles: list[tuple[str, ...]] = [] + + def strongconnect(node: str) -> None: + indices[node] = lowlinks[node] = next(counter) + stack.append(node) + on_stack.add(node) + for dependency in graph[node]: + if dependency not in indices: + strongconnect(dependency) + lowlinks[node] = min(lowlinks[node], lowlinks[dependency]) + elif dependency in on_stack: + lowlinks[node] = min(lowlinks[node], indices[dependency]) + if lowlinks[node] != indices[node]: + return + component: list[str] = [] + while stack: + member = stack.pop() + on_stack.remove(member) + component.append(member) + if member == node: + break + component = sorted(component) + if len(component) > 1 or component[0] in graph[component[0]]: + cycles.append(tuple(component)) + + for node in sorted(graph, key=canonical_resource_identity): + if node not in indices: + strongconnect(node) + return sorted(cycles, key=lambda cycle: tuple(canonical_resource_identity(node) for node in cycle)) + + class TestPlannerSemantics: def test_dependency_edges_preserve_kinds(self): resources = { @@ -134,3 +207,39 @@ def test_refresh_propagation_is_monotonic(self, payload): impacted_b = subset_b | set(refresh_impacted_nodes(resources, subset_b)) assert impacted_a <= impacted_b + + +class TestDependencyCycleScale: + """Cycle detection must survive dependency chains longer than the recursion limit.""" + + _DEEP = 5000 + + def _chain(self, size: int) -> dict[str, tuple[str, ...]]: + graph: dict[str, tuple[str, ...]] = { + f"nodes.host-{index:05d}": (f"nodes.host-{index + 1:05d}",) for index in range(size) + } + graph[f"nodes.host-{size:05d}"] = () + return graph + + def test_deep_acyclic_chain_reports_no_cycles(self): + assert dependency_cycles(self._chain(self._DEEP)) == [] + + def test_deep_cycle_is_still_detected(self): + size = 3000 + graph = {f"nodes.host-{index:05d}": (f"nodes.host-{(index + 1) % size:05d}",) for index in range(size)} + + cycles = dependency_cycles(graph) + + assert len(cycles) == 1 + assert len(cycles[0]) == size + + def test_deep_chain_topological_order_is_complete(self): + graph = self._chain(self._DEEP) + + assert len(topological_dependency_order(graph)) == len(graph) + + @given(_dependency_graphs()) + def test_cycle_detection_matches_a_reference_walk(self, graph): + """Guards the explicit-stack walk against the recursive semantics it replaced.""" + + assert dependency_cycles(graph) == _reference_dependency_cycles(graph) From ad0586242581491ae2f075ff35e071d5cbc0007a Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 17:39:06 -0700 Subject: [PATCH 05/13] fix(runtime): keep backend exception text out of participant diagnostics The concurrent-batch failure diagnostic interpolated `str(exc)` from a backend-supplied call. Backend messages can carry host paths, credentials, or participant data, and this diagnostic travels in a portable `ApplyResult`. Only the exception type now crosses the boundary, matching `_backend_call_failed` in `backend_calls.py`. Co-Authored-By: Claude Opus 5 (1M context) --- .../raes_runtime/participant_scheduler_concurrency.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py index de907761e..5dd8ef2d2 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py @@ -402,7 +402,11 @@ def _execute_concurrent_batch(batch: _ConcurrentBatch) -> None: code="runtime.participant-concurrent-batch-failed", domain="participant", address=batch.policy.address, - message=f"Backend concurrent participant batch raised: {exc}", + # Only the exception type crosses the boundary, matching + # `_backend_call_failed`: backend messages can carry host paths, + # credentials, or participant data that must not enter a + # portable diagnostic. + message=(f"Backend concurrent participant batch did not complete ({type(exc).__name__})."), ), ) return From bd8c59c5b23842566c57df3e0039445f29dc7a5c Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 17:49:13 -0700 Subject: [PATCH 06/13] fix(runtime): measure request chunks before buffering and revert stale reservations Two follow-ups from review of the preceding fixes: - The streaming body guard appended each chunk before comparing the running total, so a single ASGI chunk larger than the limit was copied in full before the 413. The prospective length is now checked first. - Releasing a failed concurrent batch settled the reserved attempt as a failed action, but without per-action results there is no basis for the rest of a failed-action transition (`next_tick`, `next_action_index`, lifecycle). Recording the failure while leaving those untouched would let the same occurrence be serviced again at the same tick, so the reservation is reverted instead, restoring the pre-batch state exactly. Co-Authored-By: Claude Opus 5 (1M context) --- .../raes_runtime/control_plane_api_guards.py | 6 ++-- .../participant_scheduler_concurrency.py | 13 +++++--- ...rticipant_concurrent_batch_reservations.py | 21 +++++++++--- .../tests/test_runtime_control_plane_api.py | 32 +++++++++++++++++++ 4 files changed, 60 insertions(+), 12 deletions(-) diff --git a/implementations/python/packages/raes_runtime/control_plane_api_guards.py b/implementations/python/packages/raes_runtime/control_plane_api_guards.py index ea8207d04..7ad5bdb74 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api_guards.py +++ b/implementations/python/packages/raes_runtime/control_plane_api_guards.py @@ -63,9 +63,11 @@ async def _body_size_guard_response( # and could exhaust memory before any size check runs. body = bytearray() async for chunk in request.stream(): - body.extend(chunk) - if len(body) > max_request_bytes: + # Measure before copying: a single oversized chunk would otherwise be + # appended in full before the limit is consulted. + if len(body) + len(chunk) > max_request_bytes: return _request_too_large_response(control_plane, request) + body.extend(chunk) accepted_body = bytes(body) # Streaming consumes the receive channel, so seed Starlette's body cache the # way ``Request.body()`` would. Route handlers and FastAPI's own body parsing diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py index 5dd8ef2d2..d33fa9e8a 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py @@ -151,10 +151,13 @@ def _release_concurrent_reservations( in-flight and the service non-quiescent, so no later occurrence can be admitted for them. - The reserved attempt is settled as failed rather than erased: the backend was - asked to run it, so it did happen and did not succeed. That also preserves the - snapshot invariant ``attempted_actions == succeeded + failed + in_flight``, - which a bare ``in_flight`` reset would break. + The reservation is reverted rather than settled as a failed action. Without + per-action results there is no basis for the rest of a failed-action + transition (``next_tick``, ``next_action_index``, lifecycle), and recording a + failure while leaving those untouched would let the same occurrence be + serviced again at the same tick. Reverting restores the pre-batch state + exactly, so the returned failure snapshot reports the diagnostic without + inventing a half-applied occurrence. """ states = dict(run.working.participant_autonomous_execution_states) @@ -166,7 +169,7 @@ def _release_concurrent_reservations( states[context.key] = state.model_copy( update={ "in_flight": 0, - "failed_actions": state.failed_actions + state.in_flight, + "attempted_actions": state.attempted_actions - state.in_flight, } ).model_dump(mode="json") run.working = run.working.with_entries( diff --git a/implementations/python/tests/test_participant_concurrent_batch_reservations.py b/implementations/python/tests/test_participant_concurrent_batch_reservations.py index be1ae885d..035313528 100644 --- a/implementations/python/tests/test_participant_concurrent_batch_reservations.py +++ b/implementations/python/tests/test_participant_concurrent_batch_reservations.py @@ -115,7 +115,7 @@ def _stub_request_binding(monkeypatch: pytest.MonkeyPatch) -> None: ) -def _assert_reservations_released(run: SchedulerRunState) -> None: +def _assert_reservations_released(run: SchedulerRunState, before: RuntimeSnapshot) -> None: service = ParticipantExecutionServiceStateModel.model_validate( run.working.participant_execution_services[_POLICY_ADDRESS] ) @@ -124,10 +124,13 @@ def _assert_reservations_released(run: SchedulerRunState) -> None: state = ParticipantAutonomousExecutionStateModel.model_validate( run.working.participant_autonomous_execution_states[_state_key(address)] ) - # The reserved attempt is settled as failed, preserving - # attempted_actions == succeeded + failed + in_flight. assert state.in_flight == 0 assert state.attempted_actions == state.succeeded_actions + state.failed_actions + state.in_flight + # Reverting to the pre-batch state is the point: a partly-applied occurrence + # (a recorded failure whose next_tick/next_action_index never advanced) could + # be serviced again at the same tick. + assert run.working.participant_autonomous_execution_states == before.participant_autonomous_execution_states + assert run.working.participant_execution_services == before.participant_execution_services def test_miscounted_backend_batch_is_reported_and_releases_reservations(): @@ -138,13 +141,14 @@ def test_miscounted_backend_batch_is_reported_and_releases_reservations(): and the service non-quiescent, blocking all later occurrences. """ 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_reservations_released(batch.run) + _assert_reservations_released(batch.run, before) def test_raising_backend_batch_is_reported_and_releases_reservations(): @@ -154,10 +158,17 @@ 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 - _assert_reservations_released(batch.run) + # Only the exception type may cross the backend 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" in message + _assert_reservations_released(batch.run, before) diff --git a/implementations/python/tests/test_runtime_control_plane_api.py b/implementations/python/tests/test_runtime_control_plane_api.py index 8aa190ea9..6b92970f3 100644 --- a/implementations/python/tests/test_runtime_control_plane_api.py +++ b/implementations/python/tests/test_runtime_control_plane_api.py @@ -746,6 +746,38 @@ async def receive() -> dict[str, object]: assert delivered <= 3 +def test_request_size_guard_rejects_a_single_oversized_chunk_without_buffering_it(): + """One chunk larger than the cap must be refused before it is copied.""" + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + oversized = b"x" * 4096 + delivered = 0 + + async def receive() -> dict[str, object]: + nonlocal delivered + delivered += 1 + return {"type": "http.request", "body": oversized, "more_body": False} + + request = Request( + { + "type": "http", + "method": "POST", + "path": "/operations/provisioning", + "headers": [(b"content-type", b"application/json")], + "query_string": b"", + }, + receive=receive, + ) + + response = asyncio.run( + request_size_guard_response(control_plane, request, max_request_bytes=64), + ) + + assert response is not None + assert response.status_code == 413 + assert not hasattr(request.state, "raw_body") + + def test_local_control_plane_store_saves_snapshot_with_atomic_replace( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, From 05d05f7b455cfae3dd357bfe06a4d58a20997e35 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 18:01:12 -0700 Subject: [PATCH 07/13] fix(runtime): compare elapsed time against the workflow timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `timeout_seconds` carries no declared upper bound, so adding it to the start instant overflowed for a very large value — as a float timestamp and as a `timedelta`. The previous blanket `except Exception` hid that as "not timed out"; with the exception handling narrowed, it would instead abort the whole reconciliation pass and surface as a 500. Elapsed time is now compared against the timeout, which Python evaluates exactly for an arbitrarily large integer. Co-Authored-By: Claude Opus 5 (1M context) --- .../raes_runtime/control_plane_timeouts.py | 10 +++++-- ...runtime_workflow_timeout_reconciliation.py | 28 +++++++++++++++++++ 2 files changed, 35 insertions(+), 3 deletions(-) diff --git a/implementations/python/packages/raes_runtime/control_plane_timeouts.py b/implementations/python/packages/raes_runtime/control_plane_timeouts.py index ec1aaa537..24114cc81 100644 --- a/implementations/python/packages/raes_runtime/control_plane_timeouts.py +++ b/implementations/python/packages/raes_runtime/control_plane_timeouts.py @@ -94,12 +94,16 @@ def _workflow_timeout_reason( forever, so it is reclaimed under a distinct reason instead. """ - current = parse_timestamp(submitted_at).timestamp() + current = parse_timestamp(submitted_at) try: - deadline = parse_timestamp(normalized.started_at).timestamp() + timeout_seconds + started = parse_timestamp(normalized.started_at) except (TypeError, ValueError): return UNPARSEABLE_START_REASON - return TIMED_OUT_REASON if current >= deadline else None + # Elapsed time is compared against the timeout rather than added to the start + # instant: `timeout_seconds` has no declared upper bound, and folding a very + # large one into a float timestamp or a timedelta overflows. + elapsed_seconds = (current - started).total_seconds() + return TIMED_OUT_REASON if elapsed_seconds >= timeout_seconds else None def _timed_out_workflow_update( diff --git a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py index fe475dd6c..f5dcbf1a5 100644 --- a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py +++ b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py @@ -15,6 +15,15 @@ _WORKFLOW_ADDRESS = "orchestration.workflow.response" +def _workflow_entry_with_timeout(timeout_seconds: int) -> SnapshotEntry: + return SnapshotEntry( + address=_WORKFLOW_ADDRESS, + domain=RuntimeDomain.ORCHESTRATION, + resource_type="workflow", + payload={"execution_contract": {"timeout_seconds": timeout_seconds}}, + ) + + def _workflow_entry(timeout_seconds: int = 1) -> SnapshotEntry: return SnapshotEntry( address=_WORKFLOW_ADDRESS, @@ -81,6 +90,25 @@ def test_workflow_with_unparseable_started_at_is_reclaimed(started_at: str): assert update[0]["terminal_reason"] == UNPARSEABLE_START_REASON +def test_enormous_timeout_reports_not_timed_out_instead_of_overflowing(): + """`timeout_seconds` has no declared upper bound, so it must not overflow. + + Folding a very large timeout into a float timestamp or a timedelta raises + `OverflowError`, which would abort the whole reconciliation pass and surface + as a 500 from the HTTP adapter. + """ + update = workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + _workflow_entry_with_timeout(10**400), + {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")}, + {}, + "2030-01-01T00:00:00Z", + ) + + assert update is None + + def test_unparseable_reconciliation_clock_is_raised_not_swallowed(): """A bad caller-supplied ``now`` governs every workflow, so it must surface. From 28f220cd26c17a41204591d24071e28b84f4bdc9 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 18:08:47 -0700 Subject: [PATCH 08/13] fix(runtime): withdraw only this batch's reservation on rollback The concurrent-batch rollback zeroed the aggregate `in_flight` and subtracted all of it from `attempted_actions`. `_due_contexts` does not require `in_flight == 0`, so a participant can be due while an earlier action is still outstanding; a failed batch then erased that earlier work from the counters, turning `(attempted_actions, in_flight) == (1, 1)` into `(0, 0)`. `_reserve_concurrent_actions` adds exactly one per context, so exactly one is now withdrawn. Co-Authored-By: Claude Opus 5 (1M context) --- .../participant_scheduler_concurrency.py | 7 +++- ...rticipant_concurrent_batch_reservations.py | 39 +++++++++++++++++-- 2 files changed, 40 insertions(+), 6 deletions(-) diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py index d33fa9e8a..9c8ec77bf 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py @@ -166,10 +166,13 @@ def _release_concurrent_reservations( if payload is None: continue state = ParticipantAutonomousExecutionStateModel.model_validate(payload) + # Exactly the one reservation this batch added is withdrawn. Clearing the + # aggregate instead would erase in-flight work a previous batch is still + # accounting for, since `_due_contexts` does not require `in_flight == 0`. states[context.key] = state.model_copy( update={ - "in_flight": 0, - "attempted_actions": state.attempted_actions - state.in_flight, + "in_flight": state.in_flight - 1, + "attempted_actions": state.attempted_actions - 1, } ).model_dump(mode="json") run.working = run.working.with_entries( diff --git a/implementations/python/tests/test_participant_concurrent_batch_reservations.py b/implementations/python/tests/test_participant_concurrent_batch_reservations.py index 035313528..4931a7ae5 100644 --- a/implementations/python/tests/test_participant_concurrent_batch_reservations.py +++ b/implementations/python/tests/test_participant_concurrent_batch_reservations.py @@ -30,7 +30,11 @@ class _StubContext: key: str -def _execution_state(participant_address: str) -> ParticipantAutonomousExecutionStateModel: +def _execution_state( + participant_address: str, + *, + in_flight: int = 0, +) -> ParticipantAutonomousExecutionStateModel: return ParticipantAutonomousExecutionStateModel( policy_address=_POLICY_ADDRESS, policy_digest="sha256:" + "0" * 64, @@ -42,9 +46,11 @@ def _execution_state(participant_address: str) -> ParticipantAutonomousExecution lifecycle_state="running", next_tick=0, next_action_index=0, - attempted_actions=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, ) @@ -78,9 +84,9 @@ def _state_key(participant_address: str) -> str: return f"{_POLICY_ADDRESS}.state.{participant_address}" -def _batch(participant_runtime: object) -> SimpleNamespace: +def _batch(participant_runtime: object, *, in_flight: int = 0) -> SimpleNamespace: policy = SimpleNamespace(address=_POLICY_ADDRESS, max_in_flight=2) - states = [_execution_state(address) for address in _PARTICIPANTS] + 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") @@ -172,3 +178,28 @@ def _explode(requests, snapshot, workers): assert "backend exploded" not in message assert "RuntimeError" in message _assert_reservations_released(batch.run, before) + + +def test_rollback_withdraws_only_this_batch_reservation(): + """Pre-existing in-flight work must survive a failed batch rollback. + + `_due_contexts` does not require `in_flight == 0`, so a participant can be + due while an earlier action is still outstanding. Clearing the aggregate + would erase that earlier work from the counters. + """ + batch = _batch( + SimpleNamespace(admit_actions_concurrently=lambda requests, snapshot, workers: ()), + in_flight=1, + ) + before = batch.run.working + + _execute_concurrent_batch(batch) + + assert batch.run.failure is not None + for address in _PARTICIPANTS: + state = ParticipantAutonomousExecutionStateModel.model_validate( + batch.run.working.participant_autonomous_execution_states[_state_key(address)] + ) + assert (state.in_flight, state.attempted_actions) == (1, 1) + assert state.attempted_actions == state.succeeded_actions + state.failed_actions + state.in_flight + assert batch.run.working.participant_autonomous_execution_states == before.participant_autonomous_execution_states From 9ccdf56795c076956a2b8a4807d41827ac1b2ce8 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 18:17:43 -0700 Subject: [PATCH 09/13] fix(runtime): reinstate the pre-batch snapshot when a concurrent batch fails Undoing the reservation field by field had to stay consistent with in-flight work from an earlier batch, which `_due_contexts` does not exclude. Two ways it did not: the participant rollback withdrew only this batch's reservation while `_finish_concurrent_service_state` forced service `in_flight` to zero and `quiescent` to true, so service readback claimed no in-flight work while an earlier participant action was still live. The pre-batch snapshot is captured before reserving and reinstated wholesale on the failure paths, which is exact for both the participant counters and the service counters and needs no arithmetic. Co-Authored-By: Claude Opus 5 (1M context) --- .../participant_scheduler_concurrency.py | 52 ++++++------------- ...rticipant_concurrent_batch_reservations.py | 39 +++++++------- 2 files changed, 34 insertions(+), 57 deletions(-) diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py index 9c8ec77bf..fd5e92cd3 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py @@ -138,48 +138,25 @@ def _reserve_concurrent_actions( ) -def _release_concurrent_reservations( +def _restore_pre_batch_snapshot( run: SchedulerRunState, - contexts: tuple[_DueActionContext, ...], - policy_address: str, + pre_batch: RuntimeSnapshot, ) -> None: - """Undo ``_reserve_concurrent_actions`` when the batch never produced results. + """Undo ``_reserve_concurrent_actions`` when the batch produced no results. Reservations are taken before the backend batch call, and only ``_commit_concurrent_result`` clears a participant's ``in_flight``. Abandoning - the batch without releasing them would leave participants permanently - in-flight and the service non-quiescent, so no later occurrence can be - admitted for them. - - The reservation is reverted rather than settled as a failed action. Without - per-action results there is no basis for the rest of a failed-action - transition (``next_tick``, ``next_action_index``, lifecycle), and recording a - failure while leaving those untouched would let the same occurrence be - serviced again at the same tick. Reverting restores the pre-batch state - exactly, so the returned failure snapshot reports the diagnostic without - inventing a half-applied occurrence. + the batch without undoing them would leave participants in-flight and the + service non-quiescent with nothing to complete them. + + The pre-batch snapshot is reinstated wholesale rather than adjusted field by + field. Without per-action results there is no basis for a failed-action + transition (``next_tick``, ``next_action_index``, lifecycle), and arithmetic + on the counters has to agree with pre-existing in-flight work that + ``_due_contexts`` does not exclude. Reinstating is exact for both. """ - states = dict(run.working.participant_autonomous_execution_states) - for context in contexts: - payload = states.get(context.key) - if payload is None: - continue - state = ParticipantAutonomousExecutionStateModel.model_validate(payload) - # Exactly the one reservation this batch added is withdrawn. Clearing the - # aggregate instead would erase in-flight work a previous batch is still - # accounting for, since `_due_contexts` does not require `in_flight == 0`. - states[context.key] = state.model_copy( - update={ - "in_flight": state.in_flight - 1, - "attempted_actions": state.attempted_actions - 1, - } - ).model_dump(mode="json") - run.working = run.working.with_entries( - dict(run.working.entries), - participant_autonomous_execution_states=states, - ) - _finish_concurrent_service_state(run, policy_address) + run.working = pre_batch def _finish_concurrent_service_state( @@ -392,6 +369,7 @@ def _execute_concurrent_batch(batch: _ConcurrentBatch) -> None: _bound_action_request(context, batch.run.working, state) for context, state in zip(selected_contexts, selected_states, strict=True) ) + pre_batch = batch.run.working _reserve_concurrent_actions(batch.run, selected_contexts) base = batch.run.working # The batch method is backend-supplied. A raising or miscounting backend is a @@ -401,7 +379,7 @@ def _execute_concurrent_batch(batch: _ConcurrentBatch) -> None: results = batch_method(requests, base, len(requests)) result_count = len(results) except Exception as exc: # noqa: BLE001 - backend trust boundary - _release_concurrent_reservations(batch.run, selected_contexts, batch.policy.address) + _restore_pre_batch_snapshot(batch.run, pre_batch) _set_concurrent_failure( batch.run, Diagnostic( @@ -417,7 +395,7 @@ def _execute_concurrent_batch(batch: _ConcurrentBatch) -> None: ) return if result_count != len(requests): - _release_concurrent_reservations(batch.run, selected_contexts, batch.policy.address) + _restore_pre_batch_snapshot(batch.run, pre_batch) _set_concurrent_failure( batch.run, Diagnostic( diff --git a/implementations/python/tests/test_participant_concurrent_batch_reservations.py b/implementations/python/tests/test_participant_concurrent_batch_reservations.py index 4931a7ae5..aa0c4471a 100644 --- a/implementations/python/tests/test_participant_concurrent_batch_reservations.py +++ b/implementations/python/tests/test_participant_concurrent_batch_reservations.py @@ -54,7 +54,7 @@ def _execution_state( ) -def _service_state() -> ParticipantExecutionServiceStateModel: +def _service_state(*, in_flight: int = 0) -> ParticipantExecutionServiceStateModel: digest = "sha256:" + "0" * 64 return ParticipantExecutionServiceStateModel( execution_scope_ref="participant.execution-scope.green", @@ -67,14 +67,14 @@ def _service_state() -> ParticipantExecutionServiceStateModel: readiness="ready", accepting_new_work=True, draining=False, - quiescent=True, + quiescent=in_flight == 0, resources_released=False, policy_digest=digest, binding_digest=digest, time_declaration_digest=digest, capacity=2, reserved=0, - in_flight=0, + in_flight=in_flight, last_transition_ref=f"operation:{_POLICY_ADDRESS}:start:generation-1", evidence_refs=("evidence.green-login.native-action",), ) @@ -92,7 +92,7 @@ def _batch(participant_runtime: object, *, in_flight: int = 0) -> SimpleNamespac _state_key(address): state.model_dump(mode="json") for address, state in zip(_PARTICIPANTS, states, strict=True) }, - participant_execution_services={_POLICY_ADDRESS: _service_state().model_dump(mode="json")}, + participant_execution_services={_POLICY_ADDRESS: _service_state(in_flight=in_flight).model_dump(mode="json")}, ) run = SchedulerRunState(working=snapshot, diagnostics=[], changed=[]) contexts = [ @@ -122,21 +122,15 @@ def _stub_request_binding(monkeypatch: pytest.MonkeyPatch) -> None: def _assert_reservations_released(run: SchedulerRunState, before: RuntimeSnapshot) -> None: - service = ParticipantExecutionServiceStateModel.model_validate( - run.working.participant_execution_services[_POLICY_ADDRESS] - ) - assert (service.in_flight, service.reserved, service.quiescent) == (0, 0, True) - for address in _PARTICIPANTS: - state = ParticipantAutonomousExecutionStateModel.model_validate( - run.working.participant_autonomous_execution_states[_state_key(address)] - ) - assert state.in_flight == 0 - assert state.attempted_actions == state.succeeded_actions + state.failed_actions + state.in_flight - # Reverting to the pre-batch state is the point: a partly-applied occurrence - # (a recorded failure whose next_tick/next_action_index never advanced) could - # be serviced again at the same tick. - assert run.working.participant_autonomous_execution_states == before.participant_autonomous_execution_states - assert run.working.participant_execution_services == before.participant_execution_services + """The failure snapshot must equal the pre-batch snapshot exactly. + + A partly-applied occurrence (a recorded failure whose next_tick and + next_action_index never advanced) could be serviced again at the same tick, + and adjusting counters by hand has to stay consistent with in-flight work + from an earlier batch. + """ + + assert run.working == before def test_miscounted_backend_batch_is_reported_and_releases_reservations(): @@ -202,4 +196,9 @@ def test_rollback_withdraws_only_this_batch_reservation(): ) assert (state.in_flight, state.attempted_actions) == (1, 1) assert state.attempted_actions == state.succeeded_actions + state.failed_actions + state.in_flight - assert batch.run.working.participant_autonomous_execution_states == before.participant_autonomous_execution_states + # Service readback must not claim zero in-flight while earlier work is live. + service = ParticipantExecutionServiceStateModel.model_validate( + batch.run.working.participant_execution_services[_POLICY_ADDRESS] + ) + assert (service.in_flight, service.quiescent) == (1, False) + assert batch.run.working == before From 0550f1cc4b1b60115301a7a19b23c222354535f5 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 19:05:42 -0700 Subject: [PATCH 10/13] test(processor): widen the dependency-cycle differential strategy The differential test reused a strategy capped at 8 nodes and 3 edges, which under-samples the case the explicit frame stack actually has to get right: a frame resumed while its remaining iterator still holds an on-stack dependency. The strategy now reaches 18 nodes and 6 edges per node at 400 examples. Verified separately over 20000 random graphs up to 22 nodes with no divergence from the recursive walk. Co-Authored-By: Claude Opus 5 (1M context) --- .../python/tests/test_semantics_planner.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/implementations/python/tests/test_semantics_planner.py b/implementations/python/tests/test_semantics_planner.py index 6055fe34c..1da53068f 100644 --- a/implementations/python/tests/test_semantics_planner.py +++ b/implementations/python/tests/test_semantics_planner.py @@ -5,7 +5,7 @@ import itertools from types import SimpleNamespace -from hypothesis import given +from hypothesis import given, settings from hypothesis import strategies as st from raes_processor.semantics.planner import ( DependencyKind, @@ -96,9 +96,13 @@ def _build(size: int, choices: list[list[int]]) -> dict[str, tuple[str, ...]]: for node, targets in zip(nodes, choices, strict=True) } - return st.integers(min_value=1, max_value=8).flatmap( + # Deliberately larger and denser than the DAG strategies above: the explicit + # frame stack has to resume a partially consumed iterator, so the graphs that + # matter are the ones where a node still has unvisited dependencies left when + # a descent happens and some of those are already on the stack. + return st.integers(min_value=1, max_value=18).flatmap( lambda size: st.lists( - st.lists(st.integers(min_value=0, max_value=7), max_size=3), + st.lists(st.integers(min_value=0, max_value=17), max_size=6), min_size=size, max_size=size, ).map(lambda choices: _build(size, choices)) @@ -238,6 +242,7 @@ def test_deep_chain_topological_order_is_complete(self): assert len(topological_dependency_order(graph)) == len(graph) + @settings(max_examples=400) @given(_dependency_graphs()) def test_cycle_detection_matches_a_reference_walk(self, graph): """Guards the explicit-stack walk against the recursive semantics it replaced.""" From 52950318a86e3ca7f678f0a3bb1365f14ff0b5d7 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 20:03:30 -0700 Subject: [PATCH 11/13] refactor(runtime): fold the concurrent-batch abandon paths together SonarCloud reported one new issue in each of the two files this branch touches most. The two batch-failure paths differed only in their diagnostic code and message, so they now share `_abandon_concurrent_batch`, which undoes the reservations and records the failure. That removes the duplication and shortens `_execute_concurrent_batch`. The remaining two constructs are deliberate and are marked as such. The broad `except` is a backend trust boundary, where any failure has to become a diagnostic rather than escape the scheduler, exactly as `backend_calls` already does. Seeding `request._body` is Starlette's own hook for middleware that consumes the body -- `_CachedRequest.wrapped_receive` replays it to the inner app -- and has no public equivalent. Co-Authored-By: Claude Opus 5 (1M context) --- .../raes_runtime/control_plane_api_guards.py | 7 ++- .../participant_scheduler_concurrency.py | 61 +++++++++++-------- 2 files changed, 41 insertions(+), 27 deletions(-) diff --git a/implementations/python/packages/raes_runtime/control_plane_api_guards.py b/implementations/python/packages/raes_runtime/control_plane_api_guards.py index 7ad5bdb74..788a03cf8 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api_guards.py +++ b/implementations/python/packages/raes_runtime/control_plane_api_guards.py @@ -71,8 +71,11 @@ async def _body_size_guard_response( accepted_body = bytes(body) # Streaming consumes the receive channel, so seed Starlette's body cache the # way ``Request.body()`` would. Route handlers and FastAPI's own body parsing - # then still see the payload instead of an exhausted stream. - request._body = accepted_body # noqa: SLF001 + # then still see the payload instead of an exhausted stream: Starlette's own + # ``_CachedRequest.wrapped_receive`` replays ``_body`` to the inner app, which + # is the framework's hook for middleware that consumes the body, and there is + # no public equivalent. + request._body = accepted_body # noqa: SLF001 # NOSONAR - documented Starlette body-cache hook request.state.raw_body = accepted_body return None diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py index fd5e92cd3..f50f03035 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py @@ -159,6 +159,27 @@ def _restore_pre_batch_snapshot( run.working = pre_batch +def _abandon_concurrent_batch( + batch: _ConcurrentBatch, + pre_batch: RuntimeSnapshot, + *, + code: str, + message: str, +) -> None: + """Report a backend batch that produced no usable results and undo its reservations.""" + + _restore_pre_batch_snapshot(batch.run, pre_batch) + _set_concurrent_failure( + batch.run, + Diagnostic( + code=code, + domain="participant", + address=batch.policy.address, + message=message, + ), + ) + + def _finish_concurrent_service_state( run: SchedulerRunState, policy_address: str, @@ -378,34 +399,24 @@ def _execute_concurrent_batch(batch: _ConcurrentBatch) -> None: try: results = batch_method(requests, base, len(requests)) result_count = len(results) - except Exception as exc: # noqa: BLE001 - backend trust boundary - _restore_pre_batch_snapshot(batch.run, pre_batch) - _set_concurrent_failure( - batch.run, - Diagnostic( - code="runtime.participant-concurrent-batch-failed", - domain="participant", - address=batch.policy.address, - # Only the exception type crosses the boundary, matching - # `_backend_call_failed`: backend messages can carry host paths, - # credentials, or participant data that must not enter a - # portable diagnostic. - message=(f"Backend concurrent participant batch did not complete ({type(exc).__name__})."), - ), + except Exception as exc: # NOSONAR - backend trust boundary; any failure becomes a diagnostic + # Only the exception type crosses the boundary, matching + # `_backend_call_failed`: backend messages can carry host paths, + # credentials, or participant data that must not enter a portable + # diagnostic. + _abandon_concurrent_batch( + batch, + pre_batch, + code="runtime.participant-concurrent-batch-failed", + message=f"Backend concurrent participant batch did not complete ({type(exc).__name__}).", ) return if result_count != len(requests): - _restore_pre_batch_snapshot(batch.run, pre_batch) - _set_concurrent_failure( - batch.run, - Diagnostic( - code="runtime.participant-concurrent-result-count-invalid", - domain="participant", - address=batch.policy.address, - message=( - f"Backend returned {result_count} concurrent participant results for {len(requests)} requests." - ), - ), + _abandon_concurrent_batch( + batch, + pre_batch, + code="runtime.participant-concurrent-result-count-invalid", + message=f"Backend returned {result_count} concurrent participant results for {len(requests)} requests.", ) return for context, state, request, result in zip(selected_contexts, selected_states, requests, results, strict=True): From fea9046d1027784c6baed5259e1bc5290dfc3a2f Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 20:18:33 -0700 Subject: [PATCH 12/13] chore(runtime): let Sonar parse the body-cache suppression The line carried `# noqa: SLF001` before its `# NOSONAR` marker. SLF is not in this project's ruff select list, so that directive suppressed a rule that was never enabled while occupying the first comment position, which is where Sonar looks for NOSONAR. Co-Authored-By: Claude Opus 5 (1M context) --- .../python/packages/raes_runtime/control_plane_api_guards.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/implementations/python/packages/raes_runtime/control_plane_api_guards.py b/implementations/python/packages/raes_runtime/control_plane_api_guards.py index 788a03cf8..fe21eb711 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api_guards.py +++ b/implementations/python/packages/raes_runtime/control_plane_api_guards.py @@ -75,7 +75,7 @@ async def _body_size_guard_response( # ``_CachedRequest.wrapped_receive`` replays ``_body`` to the inner app, which # is the framework's hook for middleware that consumes the body, and there is # no public equivalent. - request._body = accepted_body # noqa: SLF001 # NOSONAR - documented Starlette body-cache hook + request._body = accepted_body # NOSONAR - documented Starlette body-cache hook, no public equivalent request.state.raw_body = accepted_body return None From a79d30261bc5ba7bfd8cb7c3809b091878abba80 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov Date: Tue, 11 Aug 2026 19:35:15 -0700 Subject: [PATCH 13/13] test(mcp): drive MCP tool calls through asyncio.run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `asyncio.get_event_loop()` outside a running loop has warned since 3.12 and raises `RuntimeError: There is no current event loop` on 3.14, so the whole module fails there — 20+ tests — while only emitting a DeprecationWarning on the interpreters CI currently pins. `asyncio.run` is the supported spelling for driving a coroutine from sync test code. Verified by running the module under `-W error::DeprecationWarning`, which reproduces the 3.14 failure: the previous spelling errors out, this one passes all 85 tests. Co-Authored-By: Claude Opus 5 (1M context) --- implementations/python/tests/test_mcp_server.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/implementations/python/tests/test_mcp_server.py b/implementations/python/tests/test_mcp_server.py index 57c736697..89f409bad 100644 --- a/implementations/python/tests/test_mcp_server.py +++ b/implementations/python/tests/test_mcp_server.py @@ -31,7 +31,7 @@ def _text(result) -> str: def _call(server, tool: str, args: dict | None = None) -> str: """Synchronously call a tool and return its text.""" - return asyncio.get_event_loop().run_until_complete(_async_call(server, tool, args or {})) + return asyncio.run(_async_call(server, tool, args or {})) async def _async_call(server, tool: str, args: dict) -> str: @@ -897,7 +897,7 @@ def test_server_has_all_tools(self): # Using the real registration surface (rather than a hand-copied # literal) means a drift between what the server exposes and what # raes_tool_surface advertises cannot pass silently. - registered = asyncio.get_event_loop().run_until_complete(server.list_tools()) + registered = asyncio.run(server.list_tools()) registered_names = {tool.name for tool in registered} assert registered_names, "server registered no tools"