Expose bounded raw Factory ingress to opt-in receiver v2

.github/workflows/run-tests.yaml

@@ -52,6 +52,7 @@ jobs:
           source .venv/bin/activate
           python -m pytest -p pytest_asyncio.plugin -q --disable-warnings \
             tests/test_factory_admission.py tests/test_factory_gateway.py tests/test_factory_sdk_asgi_hold.py \
+            tests/test_factory_receiver_v2.py \
             tests/test_text_peer_request_guards.py tests/test_text_request_deadline.py \
             tests/test_openai_api.py tests/test_official_openai_client.py tests/test_text_mesh.py \
             -k 'not RealTransportTests'
@@ -94,6 +95,7 @@ jobs:
             tests/test_factory_admission.py \
             tests/test_factory_gateway.py \
             tests/test_factory_sdk_asgi_hold.py \
+            tests/test_factory_receiver_v2.py \
             tests/test_device_portability.py \
             tests/test_gate16_catalog_channel.py \
             tests/test_gate16_catalog_drill.py \

docs/factory_receiver_v2_candidate.md

@@ -0,0 +1,57 @@
+# Factory receiver v2 candidate (offline)
+
+`create_factory_receiver_app_v2(receiver_profile=...)` is an opt-in route to the
+existing ordinary distributed text path. It does not change the v1
+`create_factory_app(factory_admission=...)` callback. The two adapters cannot be
+configured together. A v2 adapter must return `FactoryAdmissionV2`; a bare v1
+admission is denied.
+
+An outer ASGI gate caps every POST to the v2 app before FastAPI parses JSON.
+It retains at most 32,768 body bytes, stops at the first excess frame with 413,
+rejects a declared oversized `Content-Length` before reading any frame, and
+replays the accepted bytes unchanged. The receiver then supplies a
+`FactoryIngressObservationV2` to the host. It contains those de-chunked bytes,
+their SHA-256, the independently computed digest of the Pydantic normalized
+body, and the receiver-observed `POST` method, exact route, and
+allowlisted header values. Query strings, encoded route variants, duplicate
+projected headers, malformed headers, and size overflow are denied before the
+host's transport or Original-admission callback. Authorization, host, forwarded,
+and cookie headers are excluded from the projection. The projected names are
+the ten OpenAI SDK headers in the real Flow capture plus the four routing names
+`openai-organization`, `openai-project`, `http-referer`, and `x-title` if present.
+
+These are **receiver-observed ASGI values**, not an authenticated copy of the
+SDK-final wire. A proxy can change the path, body, headers, scheme, and host;
+sorting the receiver's selected headers cannot prove the sender's ordered
+ten-header projection. The host must authenticate the separate Original issuer
+and sender commitment, verify the actual sender principal and credential
+generation, bind the call and slot, and compare the two projections under a
+declared and verified proxy transformation policy. It must also authenticate
+the deployed receiver identity, build and generation. The v2 profile's opaque
+in-process marker and transport verifier receive only host-populated ASGI scope
+extensions; caller headers and the API bearer cannot create that marker. The
+host must issue it only after checking the transport channel and generation.
+The source checks marker identity and generation before calling the Original
+adapter. The returned admission binds the observed projection digest and
+transport fields and continues to use the existing fresh authority and
+one-attempt dispatch claim. These checks cannot turn an unverified host
+callback into a trust root. The admission's fresh authority callback must also
+recheck Original and transport validity immediately before dispatch.
+
+The real Flow fixture has a 299-byte SDK body with SHA-256
+`b86e125b6a031a226acc03147b5d865402ad149a621626fcdbf7629f5541fd03`.
+PR35's Python-normalized body has 307 bytes and SHA-256
+`ab288a143d34de92156f01d0d4c4f5d05bbdc51eda4a335e16c8816205cd08b7`.
+The separate FACTORY held v3 candidate now commits that same 307-byte Python
+normalization. Its `HELD_BEFORE_POST` expected observation includes the
+authenticated sender and credential fields and the ten SDK-final headers; it
+cannot be filled by this post-ingress ASGI observation. FACTORY has a separate
+synthetic post-POST receipt shape, but a real authenticated receipt and host
+trust root are still required for the join.
+
+ASGI delivers whole frames, so an oversized single frame can exist transiently
+before the gate rejects it. The deployment ingress still needs a global edge
+limit on request/frame size and duration, including non-POST paths. This
+candidate has no Original issuer, authenticated transport,
+deployed receiver receipt, or physical send. The offline test host supplies
+synthetic values and holds the dispatch claim.

src/drift/api/server.py

@@ -31,6 +31,15 @@
 from transformers import StoppingCriteria, StoppingCriteriaList, TextIteratorStreamer
 
 from drift.factory_admission import FactoryAdmission, RequestAdmissionDenied, synchronous_result
+from drift.factory_receiver import (
+    FactoryAdmissionV2,
+    FactoryBoundedBodyMiddleware,
+    FactoryIngressHeadersInvalid,
+    FactoryIngressTooLarge,
+    FactoryReceiverProfileV2,
+    HostVerifiedTransport,
+    observe_factory_ingress_v2,
+)
 from drift.node.model_manager import (
     AmbiguousModelError,
     AutoModelUnavailableError,
@@ -186,6 +195,17 @@ def create_factory_app(*, factory_admission: Callable[[dict], FactoryAdmission],
     return create_app(factory_admission=factory_admission, **kwargs)
 
 
+def create_factory_receiver_app_v2(*, receiver_profile: FactoryReceiverProfileV2, **kwargs) -> FastAPI:
+    """Opt in to observed ASGI ingress plus host-verified Original admission.
+
+    The profile does not itself verify transport, Original provenance, or the
+    sender's SDK-final wire. Those are obligations of the embedding host.
+    """
+    if type(receiver_profile) is not FactoryReceiverProfileV2:
+        raise ValueError("Factory receiver v2 requires a host-owned profile")
+    return create_app(factory_receiver_v2=receiver_profile, **kwargs)
+
+
 def create_app(
     model=None,
     tokenizer=None,
@@ -200,6 +220,7 @@ def create_app(
     route_outcome_observer: Optional[Callable[..., None]] = None,
     request_timeout: float = 900.0,
     factory_admission: Optional[Callable[[dict], FactoryAdmission]] = None,
+    factory_receiver_v2: Optional[FactoryReceiverProfileV2] = None,
 ) -> FastAPI:
     """Create the OpenAI-compatible application.
 
@@ -223,14 +244,20 @@ def create_app(
         raise ValueError("pass either model_manager or the single-model arguments, not both")
     if sum((bool(api_keys), api_key_verifier is not None, api_key_identifier is not None)) > 1:
         raise ValueError("pass only one API key authentication method")
-    if factory_admission is not None and (
-        not callable(factory_admission)
+    if factory_admission is not None and factory_receiver_v2 is not None:
+        raise ValueError("choose one Factory receiver profile")
+    if factory_receiver_v2 is not None and type(factory_receiver_v2) is not FactoryReceiverProfileV2:
+        raise ValueError("Factory receiver v2 requires a host-owned profile")
+    if (factory_admission is not None or factory_receiver_v2 is not None) and (
+        (factory_admission is not None and not callable(factory_admission))
         or not (api_keys or api_key_verifier is not None or api_key_identifier is not None)
     ):
         raise ValueError("Factory admission requires a trusted adapter and API authentication")
 
     app = FastAPI(title="DRIFT-LLM OpenAI-compatible API")
-    if factory_admission is not None:
+    if factory_receiver_v2 is not None:
+        app.add_middleware(FactoryBoundedBodyMiddleware)
+    if factory_admission is not None or factory_receiver_v2 is not None:
         # A slash redirect would precede original admission and allow a body resend.
         app.router.redirect_slashes = False
     semaphore = asyncio.Semaphore(max_concurrent)
@@ -256,8 +283,9 @@ def check_auth(request: Request) -> str | None:
             raise HTTPException(status_code=401, detail="Invalid API key")
         return identity
 
-    def request_context(body, request):
+    async def request_context(body, request, route):
         admission = None
+        dispatch_guard = None
         original_body = body.model_dump(exclude_none=True)
         if factory_admission is not None:
             try:
@@ -267,6 +295,40 @@ def request_context(body, request):
                 if type(admission) is not FactoryAdmission:
                     raise RequestAdmissionDenied()
                 admission.require_current(original_body)
+                dispatch_guard = lambda: admission.require_current(original_body)
+            except Exception:
+                raise HTTPException(status_code=403, detail="Original request admission denied") from None
+        elif factory_receiver_v2 is not None:
+            try:
+                # The v2 outer ASGI gate has already capped the body before
+                # framework parsing. Retain this check at the adapter boundary.
+                raw_body = await request.body()
+                observation = observe_factory_ingress_v2(request.scope, raw_body, original_body, route)
+            except FactoryIngressTooLarge:
+                raise HTTPException(status_code=413, detail="Factory ingress body too large") from None
+            except FactoryIngressHeadersInvalid:
+                raise HTTPException(status_code=431, detail="Factory ingress headers invalid") from None
+            except Exception:
+                raise HTTPException(status_code=403, detail="Original request admission denied") from None
+            try:
+                # Scope extensions are supplied by trusted host middleware. The
+                # HTTP bearer and projected request headers do not mint identity.
+                extensions = request.scope.get("extensions", {})
+                if type(extensions) is not dict:
+                    raise RequestAdmissionDenied()
+                transport = synchronous_result(factory_receiver_v2.transport_verifier(extensions))
+                if (
+                    type(transport) is not HostVerifiedTransport
+                    or transport.host_capability is not factory_receiver_v2.host_capability
+                    or transport.generation_sha256 != factory_receiver_v2.generation_sha256
+                ):
+                    raise RequestAdmissionDenied()
+                decision = synchronous_result(factory_receiver_v2.original_admission(observation, transport))
+                if type(decision) is not FactoryAdmissionV2:
+                    raise RequestAdmissionDenied()
+                decision.require_current(observation, original_body, transport)
+                admission = decision.admission
+                dispatch_guard = lambda: decision.require_current(observation, original_body, transport)
             except Exception:
                 raise HTTPException(status_code=403, detail="Original request admission denied") from None
         caller_id = check_auth(request)
@@ -284,7 +346,7 @@ def request_context(body, request):
             caller_id=caller_id,
             request_id=admission.request_id,
             single_attempt=True,
-            dispatch_guard=lambda: admission.require_current(original_body),
+            dispatch_guard=dispatch_guard,
             dispatch_claim=admission.claim_dispatch,
         )
 
@@ -546,7 +608,7 @@ async def chat_completions(body: ChatCompletionRequest, request: Request):
             raise HTTPException(status_code=400, detail="n > 1 is not supported")
         if body.stream_options is not None and not body.stream:
             raise HTTPException(status_code=400, detail="stream_options requires stream=true")
-        context = request_context(body, request)
+        context = await request_context(body, request, "/v1/chat/completions")
         loaded = await load_with_budget(body.model, context)
         if context.single_attempt and (
             loaded.descriptor.manifest_digest != body.model
@@ -632,7 +694,7 @@ async def completions(body: CompletionRequest, request: Request):
             raise HTTPException(status_code=400, detail="n > 1 is not supported")
         if body.stream_options is not None and not body.stream:
             raise HTTPException(status_code=400, detail="stream_options requires stream=true")
-        context = request_context(body, request)
+        context = await request_context(body, request, "/v1/completions")
         loaded = await load_with_budget(body.model, context)
         if context.single_attempt and (
             loaded.descriptor.manifest_digest != body.model