Feat/executor cleanup cursor

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

@@ -11,11 +11,13 @@
 
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
 from fastapi.responses import StreamingResponse
+from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
 from app.api.dependencies import get_db, with_task_telemetry
 from app.core import security
 from app.core.config import settings
+from app.db.session import get_async_db
 from app.models.user import User
 from app.schemas.remote_workspace import (
     RemoteWorkspaceStatusResponse,
@@ -844,18 +846,18 @@ def cancel_preserve_executor(
 
 
 @router.post("/{task_id}/cleanup-executor", response_model=dict)
-def cleanup_task_executor(
+async def cleanup_task_executor(
     task_id: int = Depends(with_task_telemetry),
     current_user: User = Depends(security.get_current_user),
-    db: Session = Depends(get_db),
+    db: AsyncSession = Depends(get_async_db),
 ):
     """
     Clean up the executor for a finished task.
 
     This endpoint reuses the same cleanup rules as the scheduled executor cleanup
     job and only affects the specified task.
     """
-    return job_service.cleanup_task_executor(
+    return await job_service.cleanup_task_executor(
         db=db,
         task_id=task_id,
         user_id=current_user.id,

backend/app/core/cache.py

@@ -2,7 +2,6 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-import asyncio
 import logging
 from typing import Any, Dict, List, Optional
 
@@ -134,14 +133,20 @@ def get_user_repositories_sync(
         return self.get_sync(cache_key)
 
     async def set(
-        self, key: str, value: Any, expire: int = settings.REPO_CACHE_EXPIRED_TIME
+        self,
+        key: str,
+        value: Any,
+        expire: int | None = settings.REPO_CACHE_EXPIRED_TIME,
     ) -> bool:
-        """Set value to cache with expiration (seconds)"""
+        """Set value to cache with optional expiration (seconds)"""
         try:
             client = await self._get_client()
             try:
                 payload = orjson.dumps(value)
-                ok = await client.set(key, payload, ex=expire)
+                if expire is None:
+                    ok = await client.set(key, payload)
+                else:
+                    ok = await client.set(key, payload, ex=expire)
                 return bool(ok)
             finally:
                 await client.aclose()

backend/app/core/config.py

@@ -169,6 +169,10 @@ class Settings(BaseSettings):
     # and updated_at exceeds this threshold, the executor task will be deleted automatically.
     CHAT_TASK_EXECUTOR_DELETE_AFTER_HOURS: int = 2
     CODE_TASK_EXECUTOR_DELETE_AFTER_HOURS: int = 24
+    STALE_NON_TERMINAL_TASK_EXECUTOR_DELETE_AFTER_HOURS: int = 24
+    TASK_EXECUTOR_CLEANUP_PRIMARY_SCAN_BATCH_SIZE: int = 2000
+    TASK_EXECUTOR_CLEANUP_LOOKBACK_HOURS: int = 96
+    TASK_EXECUTOR_CLEANUP_LOOKBACK_SCAN_LIMIT: int = 500
     # Cleanup scanning interval seconds
     TASK_EXECUTOR_CLEANUP_INTERVAL_SECONDS: int = 600
 

backend/app/core/distributed_lock.py

@@ -29,12 +29,14 @@
             distributed_lock.release("my_task")
 """
 
+import asyncio
 import logging
-from contextlib import contextmanager
+from contextlib import asynccontextmanager, contextmanager
 from threading import Event, Thread
-from typing import Generator, Optional
+from typing import AsyncGenerator, Generator, Optional
 
 import redis
+from redis.asyncio import Redis as AsyncRedis
 
 from app.core.config import settings
 
@@ -56,6 +58,7 @@ class DistributedLock:
     def __init__(self):
         """Initialize the distributed lock with lazy Redis client loading."""
         self._redis_client: Optional[redis.Redis] = None
+        self._async_redis_client: Optional[AsyncRedis] = None
 
     @property
     def redis_client(self) -> Optional[redis.Redis]:
@@ -252,6 +255,125 @@ def _watchdog() -> None:
             if acquired:
                 self.release(lock_name)
 
+    @property
+    def async_redis_client(self) -> Optional[AsyncRedis]:
+        """Lazy-load async Redis client from settings."""
+        if self._async_redis_client is None:
+            try:
+                redis_url = getattr(settings, "CELERY_BROKER_URL", None) or getattr(
+                    settings, "REDIS_URL", None
+                )
+                if redis_url:
+                    self._async_redis_client = AsyncRedis.from_url(
+                        redis_url,
+                        decode_responses=True,
+                        socket_timeout=5,
+                        socket_connect_timeout=5,
+                    )
+            except Exception as e:
+                logger.warning(
+                    f"[DistributedLock] Failed to create async Redis client: {e}"
+                )
+                self._async_redis_client = None
+        return self._async_redis_client
+
+    async def acquire_async(self, lock_name: str, expire_seconds: int = 60) -> bool:
+        """Acquire a distributed lock asynchronously."""
+        if self.async_redis_client is None:
+            logger.debug(
+                f"[DistributedLock] Async Redis not available, allowing {lock_name}"
+            )
+            return True
+
+        lock_key = f"{LOCK_KEY_PREFIX}{lock_name}"
+        try:
+            result = await self.async_redis_client.set(
+                lock_key, "1", nx=True, ex=expire_seconds
+            )
+            if result:
+                logger.debug(f"[DistributedLock] Acquired async lock: {lock_name}")
+            else:
+                logger.debug(
+                    f"[DistributedLock] Async lock {lock_name} already held by another instance"
+                )
+            return result is True
+        except Exception as e:
+            logger.error(
+                f"[DistributedLock] Failed to acquire async lock {lock_name}: {e}"
+            )
+            return True
+
+    async def release_async(self, lock_name: str) -> bool:
+        """Release a distributed lock asynchronously."""
+        if self.async_redis_client is None:
+            return True
+
+        lock_key = f"{LOCK_KEY_PREFIX}{lock_name}"
+        try:
+            await self.async_redis_client.delete(lock_key)
+            logger.debug(f"[DistributedLock] Released async lock: {lock_name}")
+            return True
+        except Exception as e:
+            logger.error(
+                f"[DistributedLock] Failed to release async lock {lock_name}: {e}"
+            )
+            return False
+
+    async def extend_async(self, lock_name: str, expire_seconds: int = 60) -> bool:
+        """Extend the expiration time of a held lock asynchronously."""
+        if self.async_redis_client is None:
+            return True
+
+        lock_key = f"{LOCK_KEY_PREFIX}{lock_name}"
+        try:
+            result = await self.async_redis_client.expire(lock_key, expire_seconds)
+            return result is True
+        except Exception as e:
+            logger.error(
+                f"[DistributedLock] Failed to extend async lock {lock_name}: {e}"
+            )
+            return False
+
+    @asynccontextmanager
+    async def acquire_watchdog_context_async(
+        self,
+        lock_name: str,
+        expire_seconds: int = 60,
+        extend_interval_seconds: int = 30,
+    ) -> AsyncGenerator[bool, None]:
+        """Acquire a lock and keep extending it with an async watchdog task."""
+        acquired = await self.acquire_async(lock_name, expire_seconds)
+        watchdog_task: Optional[asyncio.Task] = None
+
+        if (
+            acquired
+            and self.async_redis_client is not None
+            and extend_interval_seconds > 0
+            and extend_interval_seconds < expire_seconds
+        ):
+
+            async def _watchdog() -> None:
+                try:
+                    while True:
+                        await asyncio.sleep(extend_interval_seconds)
+                        await self.extend_async(lock_name, expire_seconds)
+                except asyncio.CancelledError:
+                    pass
+
+            watchdog_task = asyncio.create_task(_watchdog())
+
+        try:
+            yield acquired
+        finally:
+            if watchdog_task is not None:
+                watchdog_task.cancel()
+                try:
+                    await watchdog_task
+                except asyncio.CancelledError:
+                    pass
+            if acquired:
+                await self.release_async(lock_name)
+
 
 # Singleton instance
 distributed_lock = DistributedLock()

backend/app/db/session.py

@@ -2,10 +2,11 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from contextlib import contextmanager
-from typing import Generator
+from contextlib import asynccontextmanager, contextmanager
+from typing import AsyncGenerator, Generator, Optional
 
 from sqlalchemy import create_engine
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -54,6 +55,78 @@ def _create_engine():
 # Sync session factory
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
+# Async database support (lazily initialized)
+_async_engine: Optional[AsyncEngine] = None
+_AsyncSessionLocal: Optional[sessionmaker] = None
+
+
+def _get_async_database_url() -> str:
+    """Convert sync DATABASE_URL to async driver URL."""
+    url = SQLALCHEMY_DATABASE_URL
+    if "pymysql" in url:
+        return url.replace("pymysql", "asyncmy")
+    elif url.startswith("mysql://"):
+        return url.replace("mysql://", "mysql+asyncmy://")
+    elif "asyncmy" in url:
+        return url
+    elif url.startswith("sqlite"):
+        return url.replace("sqlite://", "sqlite+aiosqlite://")
+    return url.replace("mysql+", "mysql+asyncmy+", 1)
+
+
+def _create_async_engine() -> AsyncEngine:
+    """Create async database engine."""
+    async_url = _get_async_database_url()
+    if async_url.startswith("sqlite"):
+        return create_async_engine(async_url, pool_pre_ping=True)
+    return create_async_engine(
+        async_url,
+        pool_pre_ping=True,
+        pool_size=10,
+        max_overflow=20,
+        pool_timeout=30,
+        pool_recycle=3600,
+    )
+
+
+def get_async_engine() -> AsyncEngine:
+    """Get the async database engine, initializing if needed."""
+    global _async_engine
+    if _async_engine is None:
+        _async_engine = _create_async_engine()
+    return _async_engine
+
+
+def get_async_session_factory() -> sessionmaker:
+    """Get the async session factory, initializing if needed."""
+    global _AsyncSessionLocal
+    if _AsyncSessionLocal is None:
+        _AsyncSessionLocal = sessionmaker(
+            bind=get_async_engine(),
+            class_=AsyncSession,
+            autocommit=False,
+            autoflush=False,
+            expire_on_commit=False,
+        )
+    return _AsyncSessionLocal
+
+
+# Convenience alias for direct use
+def AsyncSessionLocal() -> AsyncSession:
+    """Create a new async session."""
+    factory = get_async_session_factory()
+    return factory()
+
+
+async def get_async_db() -> AsyncGenerator[AsyncSession, None]:
+    """Async database session dependency for FastAPI."""
+    session = AsyncSessionLocal()
+    try:
+        yield session
+    finally:
+        await session.close()
+
+
 # Import Base from shared package for consistency
 # All models should use the same Base for proper relationship resolution
 from app.db.base import Base