Fix k8s namespace (#958)

* docs: update contributors list

* docs: update contributors list (zh)

* fix: fix k8s namespace handling in execution dispatcher and recovery service

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: 修改测试用例中的用户名为user7

* fix: 修正测试用例中的executor_name以匹配实际值

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: yunpeng7 <yunpeng7@staff.weibo.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: 4 people

backend/app/services/execution/__init__.py

@@ -14,6 +14,8 @@
 - ExecutorRecoveryService: Recovers executor Pods after deletion
 """
 
+from shared.models import ExecutionRequest
+
 from .dispatcher import ExecutionDispatcher, execution_dispatcher
 from .emitters import (
     BaseResultEmitter,
@@ -35,24 +37,13 @@
 from .schedule_helper import schedule_dispatch
 
 
-def get_sandbox_manager():
-    """Get the SandboxManager instance for creating sandboxes.
-
-    Returns:
-        SandboxManager instance from executor_manager service
-    """
-    import httpx
-
-    from app.core.config import settings
-
-    # For recovery, we need to call executor_manager's sandbox API
-    # This is a thin wrapper that will be replaced with direct call
-    # when executor_manager is properly integrated
-    return _SandboxManagerClient()
+def get_executor_runtime_client():
+    """Get the executor runtime client for runtime preparation APIs."""
+    return _ExecutorRuntimeClient()
 
 
-class _SandboxManagerClient:
-    """Thin client for calling executor_manager sandbox API."""
+class _ExecutorRuntimeClient:
+    """Thin client for calling executor_manager runtime APIs."""
 
     async def create_sandbox(
         self,
@@ -109,6 +100,35 @@ def __init__(self, data):
         except Exception as e:
             return None, str(e)
 
+    async def prepare_executor(self, request: ExecutionRequest):
+        """Prepare a normal executor runtime without dispatching the task."""
+        import httpx
+
+        from app.core.config import settings
+
+        base_url = settings.EXECUTOR_MANAGER_URL.rstrip("/")
+        url = f"{base_url}/executor-manager/executors/prepare"
+
+        try:
+            async with httpx.AsyncClient(timeout=180.0) as client:
+                response = await client.post(
+                    url,
+                    json=request.to_dict(),
+                    headers={"Content-Type": "application/json"},
+                )
+                response.raise_for_status()
+                data = response.json()
+
+                class SimpleSandbox:
+                    def __init__(self, data):
+                        self.container_name = data.get("executor_name")
+                        self.executor_namespace = data.get("executor_namespace")
+                        self.metadata = data
+
+                return SimpleSandbox(data), None
+        except Exception as e:
+            return None, str(e)
+
 
 __all__ = [
     # Router
@@ -125,7 +145,7 @@ def __init__(self, data):
     # Recovery Service
     "ExecutorRecoveryService",
     "recovery_service",
-    "get_sandbox_manager",
+    "get_executor_runtime_client",
     # Emitters - Protocol
     "ResultEmitter",
     "StreamableEmitter",

backend/app/services/execution/dispatcher.py

@@ -446,13 +446,6 @@ async def _recover_executor_if_needed(self, request: ExecutionRequest) -> None:
                     f"Task {request.task_id} not found for executor recovery"
                 )
 
-            user_id = request.user.get("id") if request.user else request.user_id
-            user_name = (
-                request.user.get("name")
-                if request.user and request.user.get("name")
-                else request.user_name
-            )
-
             logger.info(
                 "[ExecutionDispatcher] Recovering deleted executor: "
                 "task_id=%s, subtask_id=%s, executor=%s",
@@ -465,8 +458,7 @@ async def _recover_executor_if_needed(self, request: ExecutionRequest) -> None:
                 db=db,
                 subtask=subtask,
                 task=task,
-                user_id=user_id,
-                user_name=user_name,
+                request=request,
             )
             if not recovered:
                 raise RuntimeError(

backend/app/services/execution/recovery_service.py

@@ -16,16 +16,15 @@
 with git clone enabled.
 """
 
-import asyncio
 import logging
 from typing import Optional
 
 from sqlalchemy.orm import Session
 
 from app.models.subtask import Subtask
 from app.models.task import TaskResource
-from app.schemas.kind import Task
 from app.services.workspace_archive import archive_service
+from shared.models import ExecutionRequest
 
 logger = logging.getLogger(__name__)
 
@@ -74,17 +73,15 @@ async def recover(
         db: Session,
         subtask: Subtask,
         task: TaskResource,
-        user_id: int,
-        user_name: str,
+        request: ExecutionRequest,
     ) -> bool:
         """Recover executor Pod for a task.
 
         Args:
             db: Database session
             subtask: Subtask with executor_deleted_at=True
             task: Parent task
-            user_id: User ID
-            user_name: Username
+            request: Execution request carrying the normal executor config
 
         Returns:
             True if recovery successful, False otherwise
@@ -122,8 +119,7 @@ async def recover(
                     db=db,
                     subtask=subtask,
                     task=task,
-                    user_id=user_id,
-                    user_name=user_name,
+                    request=request,
                 )
             else:
                 logger.info(
@@ -134,8 +130,7 @@ async def recover(
                     db=db,
                     subtask=subtask,
                     task=task,
-                    user_id=user_id,
-                    user_name=user_name,
+                    request=request,
                 )
 
         except RuntimeError:
@@ -152,8 +147,7 @@ async def _recover_with_archive(
         db: Session,
         subtask: Subtask,
         task: TaskResource,
-        user_id: int,
-        user_name: str,
+        request: ExecutionRequest,
     ) -> bool:
         """Recover Pod using workspace archive.
 
@@ -165,33 +159,29 @@ async def _recover_with_archive(
             db: Database session
             subtask: Subtask
             task: Task
-            user_id: User ID
-            user_name: Username
+            request: Execution request carrying the normal executor config
 
         Returns:
             True if successful, False otherwise
         """
         task_id = task.id
 
         try:
-            # Create Pod with skip_git_clone=true
-            sandbox, error = await self._create_sandbox(
-                task=task,
-                subtask=subtask,
-                user_id=user_id,
-                user_name=user_name,
-                skip_git_clone=True,
-            )
+            request.skip_git_clone = True
+            request.executor_name = None
+            request.executor_namespace = None
 
-            if error or not sandbox:
+            executor, error = await self._prepare_executor(request, True)
+
+            if error or not executor:
                 logger.error(
-                    f"[RecoveryService] Failed to create sandbox for task {task_id}: {error}"
+                    f"[RecoveryService] Failed to prepare executor for task {task_id}: {error}"
                 )
                 return False
 
-            executor_name = sandbox.container_name
+            executor_name = executor.container_name
             executor_namespace = self._resolve_executor_namespace(
-                sandbox=sandbox,
+                sandbox=executor,
                 previous_namespace=subtask.executor_namespace,
             )
 
@@ -213,6 +203,8 @@ async def _recover_with_archive(
             subtask.executor_name = executor_name
             subtask.executor_namespace = executor_namespace
             subtask.executor_deleted_at = False
+            request.executor_name = executor_name
+            request.executor_namespace = executor_namespace
             db.add(subtask)
             db.commit()
 
@@ -235,8 +227,7 @@ async def _recover_without_archive(
         db: Session,
         subtask: Subtask,
         task: TaskResource,
-        user_id: int,
-        user_name: str,
+        request: ExecutionRequest,
     ) -> bool:
         """Recover Pod with normal git clone (no archive).
 
@@ -247,40 +238,38 @@ async def _recover_without_archive(
             db: Database session
             subtask: Subtask
             task: Task
-            user_id: User ID
-            user_name: Username
+            request: Execution request carrying the normal executor config
 
         Returns:
             True if successful, False otherwise
         """
         task_id = task.id
 
         try:
-            # Create Pod normally
-            sandbox, error = await self._create_sandbox(
-                task=task,
-                subtask=subtask,
-                user_id=user_id,
-                user_name=user_name,
-                skip_git_clone=False,
-            )
+            request.skip_git_clone = False
+            request.executor_name = None
+            request.executor_namespace = None
+
+            executor, error = await self._prepare_executor(request, False)
 
-            if error or not sandbox:
+            if error or not executor:
                 logger.error(
-                    f"[RecoveryService] Failed to create sandbox for task {task_id}: {error}"
+                    f"[RecoveryService] Failed to prepare executor for task {task_id}: {error}"
                 )
                 return False
 
-            executor_name = sandbox.container_name
+            executor_name = executor.container_name
             executor_namespace = self._resolve_executor_namespace(
-                sandbox=sandbox,
+                sandbox=executor,
                 previous_namespace=subtask.executor_namespace,
             )
 
             # Update subtask with new executor info and reset deleted flag
             subtask.executor_name = executor_name
             subtask.executor_namespace = executor_namespace
             subtask.executor_deleted_at = False
+            request.executor_name = executor_name
+            request.executor_namespace = executor_namespace
             db.add(subtask)
             db.commit()
 
@@ -298,62 +287,32 @@ async def _recover_without_archive(
             )
             return False
 
-    async def _create_sandbox(
+    async def _prepare_executor(
         self,
-        task: TaskResource,
-        subtask: Subtask,
-        user_id: int,
-        user_name: str,
+        request: ExecutionRequest,
         skip_git_clone: bool,
     ):
-        """Create a new sandbox Pod for the task.
+        """Prepare a normal executor runtime without initial dispatch.
 
         Args:
-            task: Task resource
-            subtask: Subtask
-            user_id: User ID
-            user_name: Username
+            request: Execution request carrying the normal executor config
             skip_git_clone: Whether to skip git clone
 
         Returns:
-            Tuple of (Sandbox, error_message)
+            Tuple of (executor runtime, error_message)
         """
-        from app.services.execution import get_sandbox_manager
-
-        task_crd = Task.model_validate(task.json)
-
-        # Get shell type from subtask or task
-        shell_type = "ClaudeCode"  # Default
+        from app.services.execution import get_executor_runtime_client
 
-        # Build bot_config from task
-        bot_config = {}
-
-        # Build metadata
-        metadata = {
-            "task_id": task.id,
-            "subtask_id": subtask.id,
-            "skip_git_clone": skip_git_clone,
-        }
-        if subtask.executor_namespace:
-            metadata["executor_namespace"] = subtask.executor_namespace
-
-        # Get workspace ref if available
-        workspace_ref = (
-            task_crd.spec.workspaceRef.name if task_crd.spec.workspaceRef else None
-        )
+        runtime_client = get_executor_runtime_client()
+        request.skip_git_clone = skip_git_clone
+        request.executor_name = None
+        request.executor_namespace = None
 
-        # Create sandbox
-        sandbox_manager = get_sandbox_manager()
-        sandbox, error = await sandbox_manager.create_sandbox(
-            shell_type=shell_type,
-            user_id=user_id,
-            user_name=user_name,
-            workspace_ref=workspace_ref,
-            bot_config=bot_config,
-            metadata=metadata,
+        executor, error = await runtime_client.prepare_executor(
+            request=request,
         )
 
-        return sandbox, error
+        return executor, error
 
 
 # Global service instance