fix(api-gateway): fix RuntimeError(Stream consumed) for multipart uploads

api-gateway/src/main.py

@@ -35,6 +35,7 @@
 from pydantic import BaseModel
 from security import verify_public_membrane
 from starlette.middleware.cors import CORSMiddleware
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 from telemetry import init_telemetry
 
 from config import get_settings
@@ -130,6 +131,49 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
     allow_headers=["*"],
 )
 
+
+class BodyCachingMiddleware:
+    """Pure ASGI middleware that reads and replays the request body.
+
+    Stores raw bytes in scope["body_cache"] so that security dependencies
+    can read the body even after File(...) params have consumed the stream.
+    """
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        chunks: list[bytes] = []
+        more_body = True
+        while more_body:
+            message: Message = await receive()
+            chunks.append(message.get("body", b""))
+            more_body = message.get("more_body", False)
+
+        scope["body_cache"] = b"".join(chunks)
+
+        received = False
+
+        async def replayable_receive() -> Message:
+            nonlocal received
+            if not received:
+                received = True
+                return {
+                    "type": "http.request",
+                    "body": scope["body_cache"],
+                    "more_body": False,
+                }
+            return {"type": "http.disconnect"}
+
+        await self.app(scope, replayable_receive, send)
+
+
+app.add_middleware(BodyCachingMiddleware)
+
 FastAPIInstrumentor.instrument_app(app)
 
 REQUEST_ID_METADATA_KEY = "x-request-id"
@@ -141,11 +185,6 @@ async def request_id_middleware(
 ) -> Response:
     request_id = str(uuid.uuid4())
     bind_request_id(request_id)
-    # Pre-cache the raw body so that request.body() in the security dependency
-    # and request.form() for File(...) parameters both see the same bytes.
-    # Without this, File(...) consuming the ASGI stream first causes
-    # request.body() to raise RuntimeError("Stream consumed").
-    await request.body()
     try:
         return await call_next(request)
     finally:

api-gateway/src/security.py

@@ -25,10 +25,17 @@
 async def _parse_request_body(request: Request) -> bytes:
     """Read body bytes, parse JSON if applicable, and cache in request.state.parsed_body.
 
+    Reads from scope["body_cache"] set by BodyCachingMiddleware to avoid
+    RuntimeError("Stream consumed") when File(...) params consume the stream
+    before this dependency runs.
+
     Returns raw body bytes for callers that need them (e.g. for signature hashing).
     Raises HTTPException 400 on malformed JSON.
     """
-    body_bytes = await request.body()
+    # Use body cached by BodyCachingMiddleware (stored in ASGI scope).
+    # Falls back to b"" (safe: produces 401 not 500) if middleware not active.
+    body_bytes: bytes = request.scope.get("body_cache", b"")
+
     content_type = request.headers.get("content-type", "")
     if body_bytes and "application/json" in content_type:
         try:
@@ -37,6 +44,7 @@ async def _parse_request_body(request: Request) -> bytes:
             raise HTTPException(status_code=400, detail="Invalid JSON body") from None
     else:
         request.state.parsed_body = {}
+
     return body_bytes
 
 

core/src/hive/membrane/main.py

@@ -137,7 +137,8 @@ async def inspect_outbound(self, decision: Intent, context: Context) -> Intent:
                 )
                 return Intent(
                     action=cast(ActionType, ActionType.ACTION_TYPE_REJECT),
-                    reasoning=decision.reasoning + " [MEMBRANE: KYC compliance failure]",
+                    reasoning=decision.reasoning
+                    + " [MEMBRANE: KYC compliance failure]",
                     rwa_vault=rwa_intent,
                 )
             return decision
@@ -155,7 +156,8 @@ async def inspect_outbound(self, decision: Intent, context: Context) -> Intent:
                 )
                 return Intent(
                     action=cast(ActionType, ActionType.ACTION_TYPE_REJECT),
-                    reasoning=decision.reasoning + " [MEMBRANE: high-risk trade blocked]",
+                    reasoning=decision.reasoning
+                    + " [MEMBRANE: high-risk trade blocked]",
                     trade=trade_intent,
                 )
             return decision

core/src/hive/metabolism/errors.py

@@ -2,10 +2,14 @@
 Metabolic Errors: Custom exceptions for the Hive's ATCG loop.
 """
 
+
 class MetabolicError(Exception):
     """Base class for all metabolic errors."""
+
     pass
 
+
 class MetabolicSecurityError(MetabolicError):
     """Raised when a security guardrail or membrane enforcement fails."""
+
     pass

core/src/hive/proteins/transaction/skill.py

@@ -227,7 +227,9 @@ async def _execute_rwa_collateral(self, params: dict[str, Any]) -> Observation:
         # 1. Access context metadata for security enforcement
         context = params.get("_context")
         if not context:
-            raise MetabolicSecurityError("Security context missing: HiveContext required")
+            raise MetabolicSecurityError(
+                "Security context missing: HiveContext required"
+            )
 
         # Extract metadata from Context (google.protobuf.Struct)
         metadata = (
@@ -237,7 +239,9 @@ async def _execute_rwa_collateral(self, params: dict[str, Any]) -> Observation:
         aml_risk = metadata.get("aml_risk")
 
         # 2. Strict C2C9 Enforcement logic
-        required_kyc = self.settings.required_kyc_status if self.settings else "APPROVED"
+        required_kyc = (
+            self.settings.required_kyc_status if self.settings else "APPROVED"
+        )
         required_aml = self.settings.required_aml_risk if self.settings else "LOW"
 
         if kyc_status != required_kyc or aml_risk != required_aml:

core/src/hive/proteins/transaction/solana_engine.py

@@ -40,7 +40,9 @@ def __init__(self, private_key_base58: str, rpc_url: str, usdc_mint: str):
     def _derive_ata(self, owner: Pubkey, mint: Pubkey) -> Pubkey:
         return get_associated_token_address(owner, mint)
 
-    async def execute_rwa_collateral(self, wallet_address: str, amount_usdc: float) -> str:
+    async def execute_rwa_collateral(
+        self, wallet_address: str, amount_usdc: float
+    ) -> str:
         """
         Execute an SPL Token transfer (USDC) from the Hive Treasury to the user's wallet.
         Simulates the release of a stablecoin loan.
@@ -49,14 +51,18 @@ async def execute_rwa_collateral(self, wallet_address: str, amount_usdc: float)
         user_ata = self._derive_ata(user_pubkey, self.usdc_mint_pubkey)
 
         # 1. Verification of Token Decimals (Robustness check)
-        mint_info = (await self.async_rpc_client.get_token_supply(self.usdc_mint_pubkey)).value
+        mint_info = (
+            await self.async_rpc_client.get_token_supply(self.usdc_mint_pubkey)
+        ).value
         if mint_info.decimals != USDC_DECIMALS:
             logger.error(
                 "usdc_decimal_mismatch",
                 expected=USDC_DECIMALS,
                 actual=mint_info.decimals,
             )
-            raise ValueError(f"USDC decimal mismatch: expected {USDC_DECIMALS}, got {mint_info.decimals}")
+            raise ValueError(
+                f"USDC decimal mismatch: expected {USDC_DECIMALS}, got {mint_info.decimals}"
+            )
 
         amount_raw = int(amount_usdc * (10**USDC_DECIMALS))
 
@@ -81,7 +87,9 @@ async def execute_rwa_collateral(self, wallet_address: str, amount_usdc: float)
         )
 
         # Build message and transaction using solders
-        recent_blockhash = (await self.async_rpc_client.get_latest_blockhash()).value.blockhash
+        recent_blockhash = (
+            await self.async_rpc_client.get_latest_blockhash()
+        ).value.blockhash
         msg = Message([transfer_ix], self.keypair.pubkey())
         tx = Transaction([self.keypair], msg, recent_blockhash)
 

core/src/hive/transformer/main.py

@@ -186,17 +186,31 @@ def _build_rwa_context(
         """
         vision_report: dict[str, Any] = _as_dict(metadata.get("vision_report"))  # type: ignore[assignment]
         wallet_address = str(metadata.get("wallet_address", ""))
-        kyc_status_str = "true" if metadata.get("kyc_status", "false") == "true" else "false"
+        kyc_status_str = (
+            "true" if metadata.get("kyc_status", "false") == "true" else "false"
+        )
         ltv_ratio_str = str(self._rwa_ltv_ratio())
         six_rates: dict[str, Any] = _as_dict(metadata.get("six_rates"))  # type: ignore[assignment]
         system_vitals: dict[str, Any] = _as_dict(metadata.get("vitals"))  # type: ignore[assignment]
-        return vision_report, wallet_address, kyc_status_str, ltv_ratio_str, six_rates, system_vitals
+        return (
+            vision_report,
+            wallet_address,
+            kyc_status_str,
+            ltv_ratio_str,
+            six_rates,
+            system_vitals,
+        )
 
     async def _think_rwa(self, context: Context, metadata: dict[str, Any]) -> Intent:
         """RWA path: invoke AppraiseAndVerifyRWA via the reasoning protein."""
-        vision_report, wallet_address, kyc_status_str, ltv_ratio_str, six_rates, system_vitals = (
-            self._build_rwa_context(metadata)
-        )
+        (
+            vision_report,
+            wallet_address,
+            kyc_status_str,
+            ltv_ratio_str,
+            six_rates,
+            system_vitals,
+        ) = self._build_rwa_context(metadata)
 
         obs = await self.registry.execute(
             "reasoning",
@@ -253,7 +267,9 @@ async def _think_rwa(self, context: Context, metadata: dict[str, Any]) -> Intent
 
         is_rejected = compliance_status == "REJECTED"
         action = (
-            ActionType.ACTION_TYPE_REJECT if is_rejected else ActionType.ACTION_TYPE_ACCEPT
+            ActionType.ACTION_TYPE_REJECT
+            if is_rejected
+            else ActionType.ACTION_TYPE_ACCEPT
         )
 
         if is_rejected:
@@ -354,9 +370,14 @@ async def _think_trade(self, context: Context, metadata: dict[str, Any]) -> Inte
         )
 
         risk_threshold = self._trade_risk_threshold()
-        is_high_risk = risk_score > risk_threshold or "REJECTED_HIGH_RISK" in trade_intent.reasoning
+        is_high_risk = (
+            risk_score > risk_threshold
+            or "REJECTED_HIGH_RISK" in trade_intent.reasoning
+        )
         action = (
-            ActionType.ACTION_TYPE_REJECT if is_high_risk else ActionType.ACTION_TYPE_ACCEPT
+            ActionType.ACTION_TYPE_REJECT
+            if is_high_risk
+            else ActionType.ACTION_TYPE_ACCEPT
         )
 
         if is_high_risk:

core/src/hive/transformer/signatures.py

@@ -66,7 +66,7 @@ class AppraiseAndVerifyRWA(dspy.Signature):
 
     vision_report: str = dspy.InputField(
         desc="JSON string from Gemma 3 describing the physical asset: type, condition, "
-             "weight/size, estimated value, and any anomalies detected."
+        "weight/size, estimated value, and any anomalies detected."
     )
     wallet_address: str = dspy.InputField(
         desc="Solana wallet address of the counterparty submitting the asset."
@@ -76,11 +76,11 @@ class AppraiseAndVerifyRWA(dspy.Signature):
     )
     six_rates: str = dspy.InputField(
         desc="JSON string of SIX data partner FX and precious metals rates "
-             "(e.g. XAU/USD, XAG/USD, EUR/USD) used for asset valuation."
+        "(e.g. XAU/USD, XAG/USD, EUR/USD) used for asset valuation."
     )
     ltv_ratio: str = dspy.InputField(
         desc="Loan-to-value ratio as a plain float string (e.g. '0.60'). "
-             "collateral_value_usd = appraised_value_usd * ltv_ratio."
+        "collateral_value_usd = appraised_value_usd * ltv_ratio."
     )
     system_vitals: str = dspy.InputField(
         desc="JSON string with system health metrics (CPU, memory, latency)."
@@ -94,7 +94,7 @@ class AppraiseAndVerifyRWA(dspy.Signature):
     )
     violation_code: str = dspy.OutputField(
         desc='Empty string if approved. One of: "KYC_MISSING", "AML_SUSPICIOUS", '
-             '"ASSET_UNVERIFIED" if rejected.'
+        '"ASSET_UNVERIFIED" if rejected.'
     )
     appraised_value_usd: str = dspy.OutputField(
         desc='Appraised asset value in USD as a plain float string. "0.0" if rejected.'
@@ -104,9 +104,9 @@ class AppraiseAndVerifyRWA(dspy.Signature):
     )
     vault_intent_json: str = dspy.OutputField(
         desc="A single valid JSON object with fields: vault_id, asset_identifier, "
-             "asset_domain, appraised_value_usd (float), ltv_ratio (float), "
-             "collateral_value_usd (float), stablecoin_currency, wallet_address, reasoning. "
-             "No markdown fences, no extra text."
+        "asset_domain, appraised_value_usd (float), ltv_ratio (float), "
+        "collateral_value_usd (float), stablecoin_currency, wallet_address, reasoning. "
+        "No markdown fences, no extra text."
     )
 
 
@@ -141,7 +141,7 @@ class GenerateTradeRisk(dspy.Signature):
     )
     risk_threshold: str = dspy.InputField(
         desc="Maximum acceptable risk score as a plain float string (e.g. '0.10'). "
-             "Trades with risk_score above this value must be rejected."
+        "Trades with risk_score above this value must be rejected."
     )
 
     think: str = dspy.OutputField(
@@ -150,9 +150,7 @@ class GenerateTradeRisk(dspy.Signature):
     risk_score: str = dspy.OutputField(
         desc="Float 0.0–1.0 as a plain string (e.g. '0.07'). Lower is safer."
     )
-    risk_category: str = dspy.OutputField(
-        desc="One of: LOW, MEDIUM, HIGH, CRITICAL."
-    )
+    risk_category: str = dspy.OutputField(desc="One of: LOW, MEDIUM, HIGH, CRITICAL.")
 
 
 class GenerateTradeIntent(dspy.Signature):
@@ -209,7 +207,7 @@ class GenerateTradeIntent(dspy.Signature):
     )
     risk_threshold: str = dspy.InputField(
         desc="Maximum acceptable risk score as a plain float string (e.g. '0.10'). "
-             "Trades with risk_score above this value must be rejected."
+        "Trades with risk_score above this value must be rejected."
     )
 
     trade_intent_json: str = dspy.OutputField(

core/tests/test_rwa_transformer.py

@@ -149,7 +149,11 @@ def test_rwa_negotiator_approved(mocker):
     mocker.patch.object(negotiator.appraise, "forward", return_value=mock_pred)
 
     result = negotiator.forward(
-        vision_report={"asset_type": "GOLD", "weight_oz": 2.0, "condition": "excellent"},
+        vision_report={
+            "asset_type": "GOLD",
+            "weight_oz": 2.0,
+            "condition": "excellent",
+        },
         wallet_address="abc12345xyz",
         kyc_status="true",
         six_rates={"XAU_USD": 31000.0},
@@ -262,9 +266,9 @@ async def test_transformer_rwa_takes_priority_over_trade():
     intent = await transformer.think(context)
 
     params_name, _ = betterproto.which_one_of(intent, "params")
-    assert params_name == "rwa_vault", (
-        f"RWA path should take priority over trade path, got '{params_name}'"
-    )
+    assert (
+        params_name == "rwa_vault"
+    ), f"RWA path should take priority over trade path, got '{params_name}'"
     # Verify reasoning protein was called with 'rwa', not 'trade'
     # skill.execute(intent, params) — intent is the first positional arg
     call_args = mock_reasoning.execute.call_args