Add HTTP disconnection detection to Agentic RL.

dashscope/finetune/agentic_rl.py

@@ -106,6 +106,8 @@ async def register_functions(
             )
             logger.info("Function components registered")
         except Exception as e:
+            if hasattr(e, 'error_code'):
+                raise
             raise RegistrationError(
                 "Function registration failed",
                 error_code=3002,
@@ -249,6 +251,8 @@ def submit_job(
                 **kwargs,
             )
         except Exception as e:
+            if hasattr(e, 'error_code'):
+                raise
             raise RuntimeErrorWithCode(
                 "Job submission failed",
                 error_code=3005,
@@ -312,6 +316,8 @@ async def run(
                 **kwargs,
             )
         except Exception as e:
+            if hasattr(e, 'error_code'):
+                raise
             raise RuntimeErrorWithCode(
                 "RL tuning workflow failed",
                 error_code=3006,

dashscope/finetune/reinforcement/common/errors.py

@@ -23,8 +23,16 @@ def root_cause(self) -> Exception:
             root = root.__cause__
         return root
 
+    def _format_cause(self) -> str:
+        """Format root cause information if available."""
+        if self.__cause__ is not None and self is not self.root_cause:
+            root = self.root_cause
+            cause_msg = str(root).split("\n")[0][:100]
+            return f" (caused by: {type(root).__name__}: {cause_msg})"
+        return ""
+
     def __str__(self):
-        return f"[{self.error_code}] {self.message} (at {self.timestamp})"
+        return f"[{self.error_code}] {self.message} (at {self.timestamp}){self._format_cause()}"
 
 
 class IOErrorWithCode(AgenticRLError):
@@ -58,8 +66,16 @@ def root_cause(self) -> Exception:
             root = root.__cause__
         return root
 
+    def _format_cause(self) -> str:
+        """Format root cause information if available."""
+        if self.__cause__ is not None and self is not self.root_cause:
+            root = self.root_cause
+            cause_msg = str(root).split("\n")[0][:100]
+            return f" (caused by: {type(root).__name__}: {cause_msg})"
+        return ""
+
     def __str__(self):
-        return f"[{self.error_code}] {self.message}"
+        return f"[{self.error_code}] {self.message}{self._format_cause()}"
 
 
 class ValueErrorWithCode(ValueError):
@@ -78,8 +94,16 @@ def root_cause(self) -> Exception:
             root = root.__cause__
         return root
 
+    def _format_cause(self) -> str:
+        """Format root cause information if available."""
+        if self.__cause__ is not None and self is not self.root_cause:
+            root = self.root_cause
+            cause_msg = str(root).split("\n")[0][:100]
+            return f" (caused by: {type(root).__name__}: {cause_msg})"
+        return ""
+
     def __str__(self):
-        return f"[{self.error_code}] {self.message}"
+        return f"[{self.error_code}] {self.message}{self._format_cause()}"
 
 
 class InputError(AgenticRLError):

dashscope/finetune/reinforcement/common/model.py

@@ -348,6 +348,8 @@ async def get_oss(self, oss_id: Optional[str] = None) -> str:
             )
             return self.oss_signed_url
 
+        except OSSConnectionError:
+            raise
         except Exception as e:
             raise OSSConnectionError(
                 "Failed to obtain OSS URL",
@@ -1243,6 +1245,8 @@ async def register_functions(
                         )
 
         except Exception as e:
+            if hasattr(e, 'error_code'):
+                raise
             raise RegistrationError(
                 "Function component registration failed",
                 error_code=2055,

dashscope/finetune/reinforcement/common/utils.py

@@ -1004,3 +1004,46 @@ def serialize_for_output(data: Any) -> Any:
 
     # Return basic types directly
     return data
+
+
+def get_fc_request_id(request) -> str:
+    """Extract Function Compute request ID from request headers.
+
+    Retrieves the 'x-fc-request-id' header from a FastAPI/Starlette Request
+    object. This is useful for correlating logs with specific FC invocations.
+
+    Args:
+        request: FastAPI/Starlette Request object (or any object with a
+            `.headers` mapping).
+
+    Returns:
+        The FC request ID string, or "unknown" if the header is not present.
+    """
+    if request is None:
+        return "unknown"
+    headers = getattr(request, "headers", None)
+    if headers is not None and hasattr(headers, "get"):
+        return headers.get("x-fc-request-id", "unknown")
+    return "unknown"
+
+
+def get_business_summary(processor_input) -> str:
+    """Extract summary business information from processor input.
+
+    Returns the string representation of request_metadata if present,
+    otherwise returns an empty string.
+
+    Args:
+        processor_input: The processor input object (BaseDataModel subclass).
+
+    Returns:
+        String representation of request_metadata, or empty string.
+    """
+    if processor_input is None:
+        return ""
+
+    request_metadata = getattr(processor_input, "request_metadata", None)
+    if request_metadata is None:
+        return ""
+
+    return str(request_metadata)

dashscope/finetune/reinforcement/component/server/server.py

@@ -37,6 +37,7 @@
   GET  /health Health check
 """
 
+import asyncio
 import logging
 import os
 import time
@@ -47,6 +48,10 @@
 from fastapi.responses import JSONResponse
 
 from dashscope.finetune.reinforcement.common.log import logger
+from dashscope.finetune.reinforcement.common.utils import (
+    get_fc_request_id,
+    get_business_summary,
+)
 from dashscope.finetune.reinforcement.common.model_types import (
     FunctionType as FuncType,
 )
@@ -72,7 +77,7 @@
     "0",
     "no",
 )
-_THREAD_POOL_WORKERS = int(os.getenv("THREAD_POOL_WORKERS", "4"))
+_THREAD_POOL_WORKERS = int(os.getenv("THREAD_POOL_WORKERS", "32"))
 _THREAD_POOL_QUEUE = int(os.getenv("THREAD_POOL_QUEUE", "100"))
 
 if not _ENABLE_LOGGING:
@@ -386,6 +391,10 @@ async def handle_endpoint(request: Request) -> JSONResponse:
     request body, executes business logic using configured processor,
     and returns serialized result.
 
+    Monitors client connection state via ASGI disconnect messages. If the
+    client disconnects before processing completes, the processing task is
+    cancelled to avoid wasting resources.
+
     Request body format: JSON, fields determined by FuncType:
     - reward: See RewardInput
     - rollout: See RolloutInput
@@ -398,8 +407,26 @@ async def handle_endpoint(request: Request) -> JSONResponse:
     """
     start_time = time.time()
     success = False
+    cancelled = False
     processor_input = None
 
+    disconnected = asyncio.Event()
+
+    async def _listen_for_disconnect():
+        """Background listener for ASGI disconnect messages."""
+        try:
+            while True:
+                message = await request.receive()
+                if message.get("type") == "http.disconnect":
+                    disconnected.set()
+                    break
+        except Exception:
+            # If receive() raises (e.g. connection already closed),
+            # treat as disconnected.
+            disconnected.set()
+
+    disconnect_listener = None
+
     # Extract trace context from request headers
     _otel_ctx_token, _upstream_tokens = await _extract_trace_context(request)
 
@@ -413,8 +440,38 @@ async def handle_endpoint(request: Request) -> JSONResponse:
         # Check thread pool queue capacity
         await _check_queue_capacity()
 
-        # Execute processor
-        result = await func_manager.processes(processor_input)
+        # Execute processor as a task so we can cancel it on disconnect
+        process_task = asyncio.create_task(
+            func_manager.processes(processor_input),
+        )
+
+        # Wait for either the processing to finish or client disconnect
+        done, _pending = await asyncio.wait(
+            [process_task, disconnect_listener],
+            return_when=asyncio.FIRST_COMPLETED,
+        )
+
+        if disconnected.is_set():
+            # Client disconnected — cancel the processing task
+            process_task.cancel()
+            try:
+                await process_task
+            except asyncio.CancelledError:
+                pass
+            cancelled = True
+            fc_request_id = get_fc_request_id(request)
+            logger.warning(
+                "[Server] Client disconnected during processing, "
+                "task cancelled. x-fc-request-id: %s",
+                fc_request_id,
+            )
+            return JSONResponse(
+                status_code=499,
+                content={"message": "Client disconnected, request cancelled."},
+            )
+
+        # Processing completed normally
+        result = process_task.result()
         success = True
 
         # Serialize result
@@ -425,21 +482,41 @@ async def handle_endpoint(request: Request) -> JSONResponse:
     except HTTPException:
         # Re-raise HTTP exceptions as-is
         raise
+    except asyncio.CancelledError:
+        cancelled = True
+        logger.warning("[Server] Request processing was cancelled.")
+        return JSONResponse(
+            status_code=499,
+            content={"message": "Request cancelled."},
+        )
     except Exception as ex:
         logger.error(f"[Server] Unexpected error: {ex}", exc_info=True)
         return JSONResponse(
             status_code=500,
             content={"message": str(ex)},
         )
     finally:
+        # Cancel the disconnect listener if still running
+        if disconnect_listener is not None:
+            disconnect_listener.cancel()
+            try:
+                await disconnect_listener
+            except asyncio.CancelledError:
+                pass
+
         # Clean up trace context
         await _cleanup_trace_context(_otel_ctx_token, _upstream_tokens)
 
         # Log request metrics
         elapsed = round(time.time() - start_time, 4)
+        fc_req_id = get_fc_request_id(request)
+        biz_summary = get_business_summary(processor_input)
+        biz_part = f" | {biz_summary}" if biz_summary else ""
         logger.info(
             f"[Server] /api/v1 | func_type={func_type.value} | "
-            f"success={success} | elapsed={elapsed}s",
+            f"fc_request_id={fc_req_id} | "
+            f"success={success} | cancelled={cancelled} | "
+            f"elapsed={elapsed}s{biz_part}",
         )
 
         # Best-effort force flush based on platform/internal env config