fix(runtime): harden HTTP API admission and offload

docs/decisions/issue-1093-control-plane-offload-preflight.md

@@ -0,0 +1,67 @@
+# Issue 1093 Control-Plane Offload Preflight
+
+Date: 2026-08-11
+
+Issue: #1093. Requirement: API-404.
+
+## Decision
+
+The HTTP adapter remains asynchronous at the network boundary, but every
+synchronous control-plane, backend, and durable-store call runs in AnyIO's
+bounded worker pool. An application-scoped async lock admits one target-mutating
+call at a time. Mutations waiting for that lock do not occupy worker threads,
+so status, snapshot, authentication, and audit requests can still make
+progress while a backend call is slow.
+
+This includes audit persistence on middleware rejection paths. A malformed or
+oversized request is rejected first, while its audit write is awaited through a
+dedicated one-worker AnyIO capacity limiter that never borrows from the default
+pool used by authentication, reads, and ordinary control-plane work. The
+application retains at most
+`ControlPlaneSecurityConfig.max_pending_rejection_audits` such writes (default
+8, including the active write). Further rejection audits are dropped with a
+bounded warning rather than retaining request tasks or starving authenticated
+traffic. Audit persistence is best effort at this pre-routing boundary: an
+unavailable or saturated audit store must not turn a deterministic `400` or
+`413` into an exception or permit the rejected request to reach a route.
+
+The mutation queue is bounded by
+`ControlPlaneSecurityConfig.max_pending_mutations`, including the active call.
+The default is 32. Admission beyond that bound returns `503` with a
+`Retry-After` header instead of retaining an unbounded number of request tasks.
+The value must be positive and should be sized with the deployment's upstream
+connection, timeout, and retry limits.
+
+One `FastAPI` application represents one runtime target and one event loop.
+Per-application serialization prevents concurrent HTTP mutations from racing
+the target's snapshot or invoking a non-thread-safe backend concurrently. This
+does not create a distributed queue or make direct concurrent library calls
+safe across processes.
+
+Worker cancellation is deliberately non-abandoning under Starlette/AnyIO's
+`run_in_threadpool` contract: cancellation of the awaiting request does not
+kill a Python thread midway through a backend mutation. Idempotency and the
+durable operation record remain the recovery surface. Backends must still
+implement their own bounded I/O timeouts; offload is not a substitute for an
+operation deadline or process isolation.
+
+## Verification
+
+Acceptance requires:
+
+1. a blocked provisioning submission while an authenticated snapshot read
+   completes within a short independent deadline;
+2. one active mutation per target;
+3. stable `503` overload behavior at the configured pending bound;
+4. an oversized-request audit that blocks in SQLite while an independent event
+   loop task continues within a short deadline;
+5. saturation beyond the rejection-audit pending bound while a real
+   authenticated route retains a default-pool worker and completes;
+6. stable `400`/`413` rejection when audit persistence raises or is dropped;
+7. unchanged HTTP auth, idempotency, error, participant, and workflow suites;
+8. lint and repository policy; and
+9. full verification before release.
+
+Issue #1092 separately owns local transactional storage. A future multi-host
+service must use a durable broker/worker design with explicit leases and
+recovery rather than treating this in-process lock as distributed scheduling.

docs/explain/sdl/runtime-architecture.md

@@ -484,6 +484,29 @@ header identity must pass an explicit `ControlPlaneSecurityConfig`, set
 `trust_proxy_identity_headers=True`, and only trust those headers behind an
 authenticated proxy that strips caller-supplied identity headers.
 
+Bearer and verified-proxy authentication require the same exact target binding.
+An identity with no target, and an explicitly supplied bearer that is unknown,
+revoked, or scoped to another target, is rejected; a rejected bearer never
+falls back to proxy headers. Request admission is also bounded before FastAPI
+parses or dispatches a body: a public ASGI wrapper checks one digits-only
+`Content-Length`, then consumes at most the configured byte limit and replays
+one coalesced body message. This boundary does not rely on framework-private
+request caches and returns a stable `413` before a route can run.
+
+The HTTP adapter offloads synchronous backend and store calls to AnyIO's
+bounded worker pool. An application-scoped async lock serializes target
+mutations without occupying a worker while requests wait; independent reads
+remain responsive while a backend is slow. The pending mutation count is
+bounded by `ControlPlaneSecurityConfig.max_pending_mutations` and overload
+returns `503` plus `Retry-After`. This in-process execution boundary neither
+replaces backend I/O timeouts nor claims durable or distributed job queuing.
+Request-size rejection also offloads audit persistence; an audit-store failure
+cannot admit an invalid body or replace the stable `400`/`413` response. These
+pre-routing audits use a separate one-worker limiter and a bounded pending
+count, so an unauthenticated rejection flood cannot consume the default AnyIO
+workers required by authenticated reads; excess audit records are dropped with
+an operational warning.
+
 ## Current Scope
 
 The current runtime scope includes:

docs/requirements/API-404/requirement.md

@@ -22,7 +22,16 @@ Requirement inventory phase. Status audit deferred until the full canonical grap
 ## Traceability
 
 - IMPLEMENTS → GITHUB_ISSUE `8` (API-404: Secure, Durable, And Idempotent Control-Plane Semantics)
+- IMPLEMENTS → GITHUB_ISSUE `1090` (Fail-closed bearer-token authentication and target binding)
+- IMPLEMENTS → GITHUB_ISSUE `1091` (Bounded pre-routing HTTP request admission)
+- DOCUMENTS → GITHUB_ISSUE `1093` (In-process HTTP offload and rejection-audit slice)
 - IMPLEMENTS → SPEC `contracts/schemas/control-plane/operation-receipt-v1.json` (Operation receipt JSON Schema — submission acknowledgment contract)
 - IMPLEMENTS → SPEC `contracts/schemas/control-plane/operation-status-v1.json` (Operation status JSON Schema — durable operation state contract)
+- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/control_plane_api/_auth.py` (Fail-closed bearer and verified-proxy authentication)
+- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/control_plane_api/_offload.py` (Bounded worker offload and mutation admission)
+- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/control_plane_api_guards.py` (Bounded fail-closed request-size admission and rejection-audit offload)
+- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/control_plane_security.py` (HTTP admission and pending-work limits)
+- DOCUMENTS → DOCUMENTATION `docs/decisions/issue-1093-control-plane-offload-preflight.md` (In-process ASGI offload and rejection semantics)
 - TESTS → TEST `implementations/python/tests/test_runtime_control_plane.py` (Core control-plane unit tests)
 - TESTS → TEST `implementations/python/tests/test_runtime_control_plane_api.py` (HTTP/JSON control-plane API tests — auth, idempotency, durability, audit)
+- TESTS → TEST `implementations/python/tests/test_issue_1093_request_rejection_offload.py` (Non-blocking, saturation-bounded, fail-closed request rejection audit tests)

implementations/python/packages/raes_runtime/control_plane_api/__init__.py

@@ -29,6 +29,7 @@
 from ..control_plane_api_participant_retrieval import register_participant_retrieval_routes
 from ..control_plane_security import ControlPlaneSecurityConfig
 from ._auth import _ControlPlaneApiAuth
+from ._offload import _ControlPlaneCallExecutor
 from ._operation_routes import _install_request_guards, _register_operation_routes
 from ._participant_routes import (
     _register_participant_control_routes,
@@ -69,6 +70,9 @@ def create_control_plane_app(
         description="Reference HTTP/JSON adapter over the repo-owned runtime control plane.",
     )
     app.state.control_plane_api_auth = _ControlPlaneApiAuth(control_plane, security)
+    app.state.control_plane_call_executor = _ControlPlaneCallExecutor(
+        max_pending_mutations=security.max_pending_mutations
+    )
     _install_request_guards(app, control_plane, security)
     _register_operation_routes(app, control_plane)
     _register_workflow_routes(app, control_plane)

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,7 +72,25 @@ 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")
-        if identity.target_name and identity.target_name != self._control_plane.target_name:
+        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:
+        """Require an identity scoped exactly to this control-plane target."""
+
+        if 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/_offload.py

@@ -0,0 +1,69 @@
+"""Bounded offload for synchronous control-plane work."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from typing import ParamSpec, TypeVar
+
+from fastapi import HTTPException, Request
+from starlette.concurrency import run_in_threadpool
+
+_P = ParamSpec("_P")
+_T = TypeVar("_T")
+
+
+class _ControlPlaneCallExecutor:
+    """Keep blocking calls off the event loop and serialize target mutation.
+
+    FastAPI/AnyIO owns the bounded worker pool. The async lock admits only one
+    target-mutating call at a time without consuming worker threads while other
+    mutations wait. Read and audit calls may still use separate workers, so a
+    slow backend does not prevent status or authentication requests.
+    """
+
+    def __init__(self, *, max_pending_mutations: int) -> None:
+        if max_pending_mutations <= 0:
+            raise ValueError("max_pending_mutations must be positive")
+        self._mutation_lock = asyncio.Lock()
+        self._max_pending_mutations = max_pending_mutations
+        self._pending_mutations = 0
+
+    @staticmethod
+    async def run(
+        call: Callable[_P, _T],
+        /,
+        *args: _P.args,
+        **kwargs: _P.kwargs,
+    ) -> _T:
+        return await run_in_threadpool(call, *args, **kwargs)
+
+    async def mutate(
+        self,
+        call: Callable[_P, _T],
+        /,
+        *args: _P.args,
+        **kwargs: _P.kwargs,
+    ) -> _T:
+        if self._pending_mutations >= self._max_pending_mutations:
+            raise HTTPException(
+                status_code=503,
+                detail="control-plane mutation queue is full",
+                headers={"Retry-After": "1"},
+            )
+        self._pending_mutations += 1
+        try:
+            async with self._mutation_lock:
+                return await self.run(call, *args, **kwargs)
+        finally:
+            self._pending_mutations -= 1
+
+
+def _control_plane_calls(request: Request) -> _ControlPlaneCallExecutor:
+    executor = getattr(request.app.state, "control_plane_call_executor", None)
+    if not isinstance(executor, _ControlPlaneCallExecutor):
+        raise RuntimeError("control-plane call executor is not configured")
+    return executor
+
+
+__all__ = ("_ControlPlaneCallExecutor", "_control_plane_calls")