fix(runtime): close control-plane, timeout, and scheduler failure-handling gaps

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)
 

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,24 +123,15 @@ 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
         index += 1
         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)

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

implementations/python/packages/raes_runtime/control_plane_api_guards.py

@@ -56,10 +56,27 @@ 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():
+        # 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
+    # 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  # NOSONAR - documented Starlette body-cache hook, no public equivalent
+    request.state.raw_body = accepted_body
     return None
 
 

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:

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,18 +29,18 @@ 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,
             normalized,
             timeout_seconds,
             orchestration_history,
             submitted_at,
+            terminal_reason,
         )
     return update
 
@@ -77,17 +80,30 @@ 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)
     try:
-        deadline = parse_timestamp(normalized.started_at).timestamp() + timeout_seconds
-        current = parse_timestamp(submitted_at).timestamp()
-    except Exception:
-        return False
-    return current >= deadline
+        started = parse_timestamp(normalized.started_at)
+    except (TypeError, ValueError):
+        return UNPARSEABLE_START_REASON
+    # 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(
@@ -97,8 +113,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,14 +136,15 @@ 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,
         workflow_status=WorkflowStatus.TIMED_OUT,
         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,

implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py

@@ -138,6 +138,48 @@ def _reserve_concurrent_actions(
     )
 
 
+def _restore_pre_batch_snapshot(
+    run: SchedulerRunState,
+    pre_batch: RuntimeSnapshot,
+) -> None:
+    """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 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.
+    """
+
+    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,
@@ -348,11 +390,35 @@ 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
-    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:  # 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):
+        _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):
         _commit_concurrent_result(context, state, request, result, base, batch.run)
         if batch.run.failure is not None: