Feat/workspace archive recovery

backend/app/core/celery_app.py

@@ -43,6 +43,7 @@
     include=[
         "app.tasks.subscription_tasks",
         "app.tasks.knowledge_tasks",
+        "app.tasks.archive_tasks",
     ],
 )
 
@@ -73,6 +74,12 @@
             "task": "app.tasks.subscription_tasks.check_due_subscriptions",
             "schedule": float(settings.FLOW_SCHEDULER_INTERVAL_SECONDS),
         },
+        # Cleanup expired workspace archives daily at 3:00 AM UTC
+        "cleanup-expired-archives": {
+            "task": "app.tasks.archive_tasks.cleanup_expired_archives",
+            "schedule": 86400.0,  # 24 hours in seconds
+            "options": {"queue": settings.CELERY_TASK_DEFAULT_QUEUE},
+        },
     },
     # Beat scheduler class - Use default PersistentScheduler (file-based)
     # Note: Only run ONE Celery Beat instance in production

backend/app/schemas/kind.py

@@ -535,6 +535,32 @@ class TaskApp(BaseModel):
     previewUrl: str
 
 
+class ArchiveInfo(BaseModel):
+    """Workspace archive information for Pod recovery after deletion.
+
+    Stores metadata about archived workspace files that can be restored
+    when a user sends a message after the executor Pod has been deleted.
+    """
+
+    storageKey: Optional[str] = Field(
+        None,
+        description="MinIO storage key (e.g., 'workspace-archives/12345/archive.tar.gz')",
+    )
+    archivedAt: Optional[datetime] = Field(
+        None, description="Timestamp when the workspace was archived"
+    )
+    expiresAt: Optional[datetime] = Field(
+        None, description="Timestamp when the archive expires (30 days after archiving)"
+    )
+    sizeBytes: Optional[int] = Field(None, description="Archive file size in bytes")
+    sessionFileIncluded: bool = Field(
+        False, description="Whether .claude_session_id file is included in the archive"
+    )
+    gitIncluded: bool = Field(
+        False, description="Whether .git directory is included in the archive"
+    )
+
+
 class TaskStatus(Status):
     """Task status"""
 
@@ -548,6 +574,11 @@ class TaskStatus(Status):
     completedAt: Optional[datetime] = None
     subTasks: Optional[List[Dict[str, Any]]] = None
     app: Optional[TaskApp] = None  # App preview information
+    archive: Optional[ArchiveInfo] = Field(
+        None,
+        description="Workspace archive information for Pod recovery after deletion. "
+        "Set when executor Pod is deleted after task completion.",
+    )
 
 
 class Task(BaseModel):

backend/app/services/adapters/executor_job.py

@@ -5,7 +5,7 @@
 import asyncio
 import logging
 from datetime import datetime, timedelta, timezone
-from typing import List, Set
+from typing import Dict, List, Set, Tuple
 
 from sqlalchemy import and_
 from sqlalchemy.orm import Session
@@ -131,18 +131,58 @@ def cleanup_stale_executors(self, db: Session) -> None:
                 )
                 return
 
-            # Deduplicate by (namespace, name)
-            unique_executor_keys: Set[tuple[str, str]] = set()
-            for s in valid_candidates:
-                if s.executor_name:
-                    unique_executor_keys.add((s.executor_namespace, s.executor_name))
+            # Deduplicate by (namespace, name) and collect task info for archiving
+            # Map: (namespace, name) -> (subtask, task) for archive operation
+            executor_task_map: Dict[Tuple[str, str], Tuple[Subtask, TaskResource]] = {}
+            for subtask in valid_candidates:
+                if subtask.executor_name:
+                    key = (subtask.executor_namespace, subtask.executor_name)
+                    if key not in executor_task_map:
+                        # Get task for this subtask
+                        task = (
+                            db.query(TaskResource)
+                            .filter(
+                                TaskResource.id == subtask.task_id,
+                                TaskResource.kind == "Task",
+                                TaskResource.is_active == TaskResource.STATE_ACTIVE,
+                            )
+                            .first()
+                        )
+                        if task:
+                            executor_task_map[key] = (subtask, task)
 
-            if not unique_executor_keys:
+            if not executor_task_map:
                 return
 
             # Use sync version to avoid event loop issues
-            for ns, name in unique_executor_keys:
+            for (ns, name), (subtask, task) in executor_task_map.items():
                 try:
+                    # Archive workspace before deletion (for code tasks)
+                    task_crd = Task.model_validate(task.json)
+                    task_type = (
+                        task_crd.metadata.labels
+                        and task_crd.metadata.labels.get("taskType")
+                        or "chat"
+                    )
+
+                    if task_type == "code":
+                        # Try to archive workspace before deletion
+                        try:
+                            self._archive_workspace_sync(
+                                db=db,
+                                subtask=subtask,
+                                task=task,
+                                executor_name=name,
+                                executor_namespace=ns,
+                            )
+                        except Exception as archive_error:
+                            # Log but continue with deletion
+                            # Recovery will fall back to git clone if archive failed
+                            logger.warning(
+                                f"[executor_job] Failed to archive workspace for task {task.id} "
+                                f"ns={ns} name={name}: {archive_error}"
+                            )
+
                     logger.info(
                         f"[executor_job] Scheduled deleting executor task ns={ns} name={name}"
                     )
@@ -166,5 +206,59 @@ def cleanup_stale_executors(self, db: Session) -> None:
         except Exception as e:
             logger.error(f"[executor_job] cleanup_stale_executors error: {e}")
 
+    def _archive_workspace_sync(
+        self,
+        db: Session,
+        subtask: Subtask,
+        task: TaskResource,
+        executor_name: str,
+        executor_namespace: str,
+    ) -> None:
+        """Archive workspace files before Pod deletion (synchronous wrapper).
+
+        This method wraps the async archive_service.archive_workspace() for use
+        in the synchronous cleanup_stale_executors job.
+
+        Args:
+            db: Database session
+            subtask: Subtask with executor info
+            task: Task resource
+            executor_name: Executor name
+            executor_namespace: Executor namespace
+        """
+        from app.services.workspace_archive import archive_service
+
+        logger.info(
+            f"[executor_job] Archiving workspace for task {task.id}, "
+            f"executor={executor_namespace}/{executor_name}"
+        )
+
+        # Run async archive in new event loop
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        try:
+            archive_info = loop.run_until_complete(
+                archive_service.archive_workspace(
+                    db=db,
+                    subtask=subtask,
+                    task=task,
+                    executor_name=executor_name,
+                    executor_namespace=executor_namespace,
+                )
+            )
+
+            if archive_info:
+                logger.info(
+                    f"[executor_job] Workspace archived for task {task.id}, "
+                    f"size={archive_info.sizeBytes} bytes"
+                )
+            else:
+                logger.info(
+                    f"[executor_job] Workspace archiving skipped or failed for task {task.id}"
+                )
+        finally:
+            loop.run_until_complete(loop.shutdown_asyncgens())
+            loop.close()
+
 
 job_service = JobService(Kind)

backend/app/services/execution/__init__.py

@@ -11,6 +11,7 @@
 - ResultEmitter: Unified interface for emitting execution events
 - TaskRequestBuilder: Builds ExecutionRequest from database models
 - schedule_dispatch: Helper for dispatching tasks from sync context
+- ExecutorRecoveryService: Recovers executor Pods after deletion
 """
 
 from .dispatcher import ExecutionDispatcher, execution_dispatcher
@@ -28,10 +29,85 @@
     StreamableEmitter,
     WebSocketResultEmitter,
 )
+from .recovery_service import ExecutorRecoveryService, recovery_service
 from .request_builder import TaskRequestBuilder
 from .router import CommunicationMode, ExecutionRouter, ExecutionTarget
 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()
+
+
+class _SandboxManagerClient:
+    """Thin client for calling executor_manager sandbox API."""
+
+    async def create_sandbox(
+        self,
+        shell_type: str,
+        user_id: int,
+        user_name: str,
+        timeout: int = None,
+        workspace_ref: str = None,
+        bot_config: dict = None,
+        metadata: dict = None,
+    ):
+        """Create a sandbox via executor_manager API."""
+        import httpx
+
+        from app.core.config import settings
+
+        base_url = settings.EXECUTOR_MANAGER_URL.rstrip("/")
+        url = f"{base_url}/executor-manager/sandbox"
+
+        payload = {
+            "shell_type": shell_type,
+            "user_id": user_id,
+            "user_name": user_name,
+            "metadata": metadata or {},
+        }
+
+        if timeout:
+            payload["timeout"] = timeout
+        if workspace_ref:
+            payload["workspace_ref"] = workspace_ref
+        if bot_config:
+            payload["bot_config"] = bot_config
+
+        try:
+            async with httpx.AsyncClient(timeout=180.0) as client:
+                response = await client.post(
+                    url,
+                    json=payload,
+                    headers={"Content-Type": "application/json"},
+                )
+                response.raise_for_status()
+                data = response.json()
+
+                # Create a simple sandbox object
+                class SimpleSandbox:
+                    def __init__(self, data):
+                        self.sandbox_id = data.get("sandbox_id")
+                        self.container_name = data.get("container_name")
+                        self.base_url = data.get("base_url")
+
+                return SimpleSandbox(data), None
+        except Exception as e:
+            return None, str(e)
+
+
 __all__ = [
     # Router
     "ExecutionRouter",
@@ -44,6 +120,10 @@
     "TaskRequestBuilder",
     # Schedule Helper
     "schedule_dispatch",
+    # Recovery Service
+    "ExecutorRecoveryService",
+    "recovery_service",
+    "get_sandbox_manager",
     # Emitters - Protocol
     "ResultEmitter",
     "StreamableEmitter",