fix: 修复无法使用自定义镜像以及自定义镜像无法校验的问题

backend/app/api/endpoints/adapter/shells.py

@@ -2,6 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+import asyncio
 import logging
 import os
 import re
@@ -26,6 +27,18 @@
 # Redis key prefix and TTL for validation status
 VALIDATION_STATUS_KEY_PREFIX = "shell_validation:"
 VALIDATION_STATUS_TTL = 300  # 5 minutes
+# Delay before cleaning up validation container after successful validation.
+# Default is immediate cleanup (0 seconds). Keeps backward compatibility with
+# the previous variable name.
+VALIDATION_CONTAINER_CLEANUP_DELAY_SECONDS = int(
+    os.getenv(
+        "VALIDATION_SUCCESS_CLEANUP_DELAY_SECONDS",
+        os.getenv("VALIDATION_CONTAINER_CLEANUP_DELAY_SECONDS", "0"),
+    )
+)
+VALIDATION_SUBMIT_TIMEOUT_SECONDS = float(
+    os.getenv("VALIDATION_SUBMIT_TIMEOUT_SECONDS", "90")
+)
 
 
 # Request/Response Models
@@ -745,7 +758,9 @@ async def validate_image(
 
         # Call executor manager's validate-image API with validation_id
         # Use AsyncClient to avoid blocking the event loop
-        async with httpx.AsyncClient(timeout=30.0) as client:
+        async with httpx.AsyncClient(
+            timeout=VALIDATION_SUBMIT_TIMEOUT_SECONDS
+        ) as client:
             response = await client.post(
                 validate_url,
                 json={
@@ -796,18 +811,24 @@ async def validate_image(
         logger.error(f"Timeout submitting validation task for image: {image}")
         await _update_validation_status(
             validation_id,
-            status="completed",
-            stage="Error",
-            progress=100,
-            valid=False,
-            error_message="Request timed out while submitting validation task",
+            status="submitted",
+            stage="Submission timed out, validation may still be running",
+            progress=15,
+            valid=None,
+            error_message=(
+                "Submission request timed out. Validation may still be in progress; "
+                "continue polling status."
+            ),
         )
         return ImageValidationResponse(
-            status="error",
-            message="Request timed out while submitting validation task",
+            status="submitted",
+            message=(
+                "Validation submission timed out, but task may still be running. "
+                "Please continue polling."
+            ),
             validationId=validation_id,
-            valid=False,
-            errors=["Validation request timed out"],
+            valid=None,
+            errors=["Validation submission timed out; continue polling"],
         )
     except httpx.RequestError as e:
         logger.error(f"Error calling executor manager: {e}")
@@ -970,9 +991,10 @@ async def update_validation_status(
                 status_code=500, detail="Failed to update validation status"
             )
 
-        # Cleanup validation container if validation is completed
+        # Cleanup validation container if validation is completed.
+        # Run cleanup asynchronously to avoid blocking callback response.
         if request.executor_name and request.valid is True:
-            await _cleanup_validation_container(request.executor_name)
+            asyncio.create_task(_cleanup_validation_container(request.executor_name))
 
         return {"status": "success", "message": "Validation status updated"}
     except HTTPException:
@@ -997,6 +1019,14 @@ async def _cleanup_validation_container(executor_name: str) -> None:
         logger.warning("No executor_name provided for cleanup")
         return
 
+    if VALIDATION_CONTAINER_CLEANUP_DELAY_SECONDS > 0:
+        logger.info(
+            "Delaying validation container cleanup by %s seconds: %s",
+            VALIDATION_CONTAINER_CLEANUP_DELAY_SECONDS,
+            executor_name,
+        )
+        await asyncio.sleep(VALIDATION_CONTAINER_CLEANUP_DELAY_SECONDS)
+
     executor_manager_url = os.getenv("EXECUTOR_MANAGER_URL", "http://localhost:8001")
     delete_url = f"{executor_manager_url}/executor-manager/executor/delete"
 

backend/app/services/execution/request_builder.py

@@ -60,6 +60,8 @@ def __init__(self, db: Session):
         self.db = db
         # Cache shell_type to avoid repeated database queries
         self._cached_shell_type: str | None = None
+        # Cache shell info (shell_type + base_image) to avoid repeated database queries
+        self._cached_shell_info: dict | None = None
 
     def build(
         self,
@@ -643,6 +645,22 @@ def _resolve_shell_type(self, bot: Kind, user_id: int) -> str:
         This method queries the Shell CRD to get the shell_type.
         It's called once per builder instance and the result is cached.
 
+        Args:
+            bot: Bot Kind object
+            user_id: User ID for shell lookup
+
+        Returns:
+            Shell type string (e.g., "Chat", "ClaudeCode", "Agno")
+        """
+        shell_info = self._resolve_shell_info(bot, user_id)
+        return shell_info["shell_type"]
+
+    def _resolve_shell_info(self, bot: Kind, user_id: int) -> dict:
+        """Resolve shell info from bot's shellRef.
+
+        This method queries the Shell CRD to get shell_type and base_image.
+        It's called once per builder instance and the result is cached.
+
         Search order:
         1. User's private shell (user_id == user_id, namespace == shell_ref.namespace)
         2. Group shell (any user_id, namespace == shell_ref.namespace) - for group scenarios
@@ -653,22 +671,27 @@ def _resolve_shell_type(self, bot: Kind, user_id: int) -> str:
             user_id: User ID for shell lookup
 
         Returns:
-            Shell type string (e.g., "Chat", "ClaudeCode", "Agno")
+            dict: {"shell_type": str, "base_image": Optional[str]}
         """
         # Return cached value if available
-        if self._cached_shell_type is not None:
-            return self._cached_shell_type
+        if self._cached_shell_info is not None:
+            return self._cached_shell_info
 
-        # Default value
+        # Default values
         shell_type = "Chat"
+        base_image = None
 
         bot_crd = Bot.model_validate(bot.json)
 
         if not (bot_crd.spec and bot_crd.spec.shellRef):
-            self._cached_shell_type = shell_type
-            return shell_type
+            self._cached_shell_info = {
+                "shell_type": shell_type,
+                "base_image": base_image,
+            }
+            return self._cached_shell_info
 
         shell_ref = bot_crd.spec.shellRef
+        shell = None
 
         # 1. Query user's private shell first
         shell = (
@@ -710,22 +733,28 @@ def _resolve_shell_type(self, bot: Kind, user_id: int) -> str:
                 .first()
             )
 
-        # Extract shell_type from Shell CRD
+        # Extract shell_type and base_image from Shell CRD
         if shell and shell.json:
             shell_crd = Shell.model_validate(shell.json)
-            if shell_crd.spec and shell_crd.spec.shellType:
-                shell_type = shell_crd.spec.shellType
+            if shell_crd.spec:
+                if shell_crd.spec.shellType:
+                    shell_type = shell_crd.spec.shellType
+                base_image = shell_crd.spec.baseImage
 
         logger.debug(
-            "[TaskRequestBuilder] Resolved shell_type=%s for bot=%s (shell_ref=%s/%s)",
-            shell_type,
+            "[TaskRequestBuilder] Resolved shell_info for bot=%s (shell_ref=%s/%s): shell_type=%s, base_image=%s",
             bot_crd.metadata.name if bot_crd.metadata else "unknown",
             shell_ref.namespace,
             shell_ref.name,
+            shell_type,
+            base_image,
         )
 
-        self._cached_shell_type = shell_type
-        return shell_type
+        self._cached_shell_info = {
+            "shell_type": shell_type,
+            "base_image": base_image,
+        }
+        return self._cached_shell_info
 
     # =========================================================================
     # Skill Resolution (from ChatConfigBuilder)
@@ -1227,8 +1256,10 @@ def _build_bot_config(
             bot_json = bot.json or {}
             bot_spec_json = bot_json.get("spec", {})
 
-            # Get shell_type
-            shell_type = self._resolve_shell_type(bot, team.user_id)
+            # Get shell_type and base_image from Shell CRD
+            shell_info = self._resolve_shell_info(bot, team.user_id)
+            shell_type = shell_info["shell_type"]
+            base_image = shell_info["base_image"]
 
             # Get ghost info for system_prompt and skills
             ghost_system_prompt = ""
@@ -1272,7 +1303,7 @@ def _build_bot_config(
                 "mcp_servers": ghost_mcp_servers,
                 "skills": ghost_skills,
                 "role": member.role or "worker",
-                "base_image": bot_spec_json.get("baseImage"),
+                "base_image": base_image,
             }
             bot_configs.append(bot_config)
 

executor_manager/executors/docker/executor.py

@@ -310,6 +310,15 @@ def _execute_in_existing_container(
             context=f"existing container: {executor_name}",
         )
 
+    def _should_keep_failed_validation_container(self, task: Dict[str, Any]) -> bool:
+        """Whether to keep failed validation containers for debugging."""
+        is_validation_task = get_metadata_field(task, "type") == "validation"
+        if not is_validation_task:
+            return False
+
+        keep_flag = os.getenv("VALIDATION_KEEP_FAILED_CONTAINER", "false").lower()
+        return keep_flag in ("1", "true", "yes", "on")
+
     def _get_container_port(
         self, executor_name: str
     ) -> tuple[Optional[int], Optional[str]]:
@@ -375,13 +384,20 @@ def _create_new_container(
                 self.dispatch_task_to_instance(task, executor_name, ready_info)
             except Exception:
                 # Avoid leaking an idle container when initial dispatch fails.
-                try:
-                    delete_container(executor_name)
-                except Exception as cleanup_error:
-                    logger.warning(
-                        f"Failed to cleanup container {executor_name} after "
-                        f"initial dispatch failure: {cleanup_error}"
+                if self._should_keep_failed_validation_container(task):
+                    logger.info(
+                        "Keeping failed validation container %s for debugging "
+                        "(VALIDATION_KEEP_FAILED_CONTAINER=true)",
+                        executor_name,
                     )
+                else:
+                    try:
+                        delete_container(executor_name)
+                    except Exception as cleanup_error:
+                        logger.warning(
+                            f"Failed to cleanup container {executor_name} after "
+                            f"initial dispatch failure: {cleanup_error}"
+                        )
                 raise
 
     def create_instance(
@@ -588,18 +604,35 @@ def _dispatch_initial_task_to_new_container(
         Raises:
             RuntimeError: If request dispatch fails after retries
         """
-        max_retries = max(
-            int(os.getenv("EXECUTOR_INITIAL_DISPATCH_MAX_RETRIES", "3")),
-            1,
-        )
-        retry_interval = max(
-            float(os.getenv("EXECUTOR_INITIAL_DISPATCH_RETRY_INTERVAL", "1")),
-            0,
-        )
-        request_timeout = max(
-            float(os.getenv("EXECUTOR_INITIAL_DISPATCH_TIMEOUT", "10")),
-            0.1,
-        )
+        is_validation_task = get_metadata_field(task, "type") == "validation"
+        if is_validation_task:
+            # Validation tasks may spend longer in start callback path.
+            # Use dedicated limits to avoid false timeout and duplicate dispatches.
+            max_retries = max(
+                int(os.getenv("VALIDATION_INITIAL_DISPATCH_MAX_RETRIES", "1")),
+                1,
+            )
+            retry_interval = max(
+                float(os.getenv("VALIDATION_INITIAL_DISPATCH_RETRY_INTERVAL", "1")),
+                0,
+            )
+            request_timeout = max(
+                float(os.getenv("VALIDATION_INITIAL_DISPATCH_TIMEOUT", "60")),
+                0.1,
+            )
+        else:
+            max_retries = max(
+                int(os.getenv("EXECUTOR_INITIAL_DISPATCH_MAX_RETRIES", "3")),
+                1,
+            )
+            retry_interval = max(
+                float(os.getenv("EXECUTOR_INITIAL_DISPATCH_RETRY_INTERVAL", "1")),
+                0,
+            )
+            request_timeout = max(
+                float(os.getenv("EXECUTOR_INITIAL_DISPATCH_TIMEOUT", "10")),
+                0.1,
+            )
         last_error = "unknown error"
 
         for attempt in range(1, max_retries + 1):
@@ -727,14 +760,21 @@ def _check_container_health(
                     )
 
                     # Clean up the failed container
-                    try:
-                        self.subprocess.run(
-                            ["docker", "rm", "-f", executor_name],
-                            capture_output=True,
-                            timeout=10,
+                    if self._should_keep_failed_validation_container(task):
+                        logger.info(
+                            "Keeping exited validation container %s for debugging "
+                            "(VALIDATION_KEEP_FAILED_CONTAINER=true)",
+                            executor_name,
                         )
-                    except Exception:
-                        pass
+                    else:
+                        try:
+                            self.subprocess.run(
+                                ["docker", "rm", "-f", executor_name],
+                                capture_output=True,
+                                timeout=10,
+                            )
+                        except Exception:
+                            pass
 
                     # Raise exception to mark task as failed
                     raise RuntimeError(f"Container exited immediately: {error_msg}")
@@ -1005,9 +1045,16 @@ def _add_heartbeat_env_vars(self, cmd: List[str], task: Dict[str, Any]) -> None:
             cmd: Docker command list to extend
             task: Task dictionary containing task info and sandbox_metadata
         """
-        # Skip validation tasks - they are short-lived and don't need heartbeat
         task_type = get_metadata_field(task, "type", "online")
-        if task_type == "validation":
+        is_validation = task_type == "validation"
+
+        # For validation tasks, only set EXECUTOR_MANAGER_HEARTBEAT_BASE_URL for callback
+        # Skip other heartbeat settings as validation tasks are short-lived
+        if is_validation:
+            callback_url = build_callback_url(task)
+            if callback_url and "/callback" in callback_url:
+                base_url = callback_url.replace("/callback", "")
+                cmd.extend(["-e", f"EXECUTOR_MANAGER_HEARTBEAT_BASE_URL={base_url}"])
             return
 
         is_sandbox = task_type == "sandbox"