diff --git a/api/.env.example b/api/.env.example
index c2bfa9be6a..c537ecdc27 100644
--- a/api/.env.example
+++ b/api/.env.example
@@ -582,11 +582,6 @@ WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE=100
 # Maximum number of scheduled workflows to dispatch per tick (0 for unlimited)
 WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK=0
 
-# Sandbox expired records clean configuration
-SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD=21
-SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE=1000
-SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS=30
-
 # Position configuration
 POSITION_TOOL_PINS=
 POSITION_TOOL_INCLUDES=
diff --git a/api/configs/feature/__init__.py b/api/configs/feature/__init__.py
index 1391d2915e..b854293367 100644
--- a/api/configs/feature/__init__.py
+++ b/api/configs/feature/__init__.py
@@ -1152,21 +1152,6 @@ class CeleryScheduleTasksConfig(BaseSettings):
     )
 
 
-class SandboxExpiredRecordsCleanConfig(BaseSettings):
-    SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD: NonNegativeInt = Field(
-        description="Graceful period in days for sandbox records clean after subscription expiration",
-        default=21,
-    )
-    SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE: PositiveInt = Field(
-        description="Maximum number of records to process in each batch",
-        default=1000,
-    )
-    SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS: PositiveInt = Field(
-        description="Retention days for sandbox expired workflow_run records and message records",
-        default=30,
-    )
-
-
 class PositionConfig(BaseSettings):
     POSITION_PROVIDER_PINS: str = Field(
         description="Comma-separated list of pinned model providers",
diff --git a/api/services/billing_service.py b/api/services/billing_service.py
index 2666956d46..3d7cb6cc8d 100644
--- a/api/services/billing_service.py
+++ b/api/services/billing_service.py
@@ -38,47 +38,6 @@ class BillingService:
         billing_info = cls._send_request("GET", "/subscription/info", params=params)
         return billing_info
 
-    @classmethod
-    def get_info_bulk(cls, tenant_ids: Sequence[str]) -> dict[str, TenantPlanInfo]:
-        """
-        Bulk billing info fetch via billing API.
-
-        Payload: {"tenant_ids": ["t1", "t2", ...]} (max 200 per request)
-
-        Returns:
-            Mapping of tenant_id -> TenantPlanInfo(plan + expiration timestamp)
-        """
-        results: dict[str, TenantPlanInfo] = {}
-
-        chunk_size = 200
-        for i in range(0, len(tenant_ids), chunk_size):
-            chunk = tenant_ids[i : i + chunk_size]
-            try:
-                resp = cls._send_request("POST", "/subscription/plan/batch", json={"tenant_ids": chunk})
-                results.update(cls._parse_bulk_response(chunk, resp))
-            except Exception:
-                logger.exception("Failed to fetch billing info batch for tenants: %s", chunk)
-                raise
-
-        return results
-
-    @classmethod
-    def _parse_bulk_response(cls, expected_ids: Sequence[str], response: dict) -> dict[str, TenantPlanInfo]:
-        data = response.get("data")
-        if not isinstance(data, dict):
-            raise ValueError("Billing API response missing 'data' object.")
-
-        parsed: dict[str, TenantPlanInfo] = {}
-        for tenant_id in expected_ids:
-            payload = data.get(tenant_id)
-
-            try:
-                parsed[tenant_id] = TenantPlanInfo.model_validate(payload)
-            except ValidationError as exc:
-                raise ValueError(f"Invalid billing info for tenant {tenant_id}") from exc
-
-        return parsed
-
     @classmethod
     def get_tenant_feature_plan_usage_info(cls, tenant_id: str):
         params = {"tenant_id": tenant_id}
diff --git a/api/services/clear_free_plan_expired_workflow_run_logs.py b/api/services/clear_free_plan_expired_workflow_run_logs.py
index ed829b853d..459cc5f46f 100644
--- a/api/services/clear_free_plan_expired_workflow_run_logs.py
+++ b/api/services/clear_free_plan_expired_workflow_run_logs.py
@@ -173,7 +173,7 @@ class WorkflowRunCleanup:
 
         if uncached_tenants:
             try:
-                bulk_info = BillingService.get_info_bulk(uncached_tenants)
+                bulk_info = BillingService.get_plan_bulk(uncached_tenants)
             except Exception:
                 bulk_info = {}
                 logger.exception("Failed to fetch billing plans in bulk for tenants: %s", uncached_tenants)
diff --git a/docker/.env.example b/docker/.env.example
index fd0def2b1a..fb24944e37 100644
--- a/docker/.env.example
+++ b/docker/.env.example
@@ -1468,11 +1468,6 @@ WORKFLOW_SCHEDULE_POLLER_INTERVAL=1
 WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE=100
 WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK=0
 
-# Sandbox expired records clean configuration
-SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD=21
-SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE=1000
-SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS=30
-
 # Tenant isolated task queue configuration
 TENANT_ISOLATED_TASK_CONCURRENCY=1
 
diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml
index d90db26755..35c3ef1149 100644
--- a/docker/docker-compose.yaml
+++ b/docker/docker-compose.yaml
@@ -657,9 +657,6 @@ x-shared-env: &shared-api-worker-env
   WORKFLOW_SCHEDULE_POLLER_INTERVAL: ${WORKFLOW_SCHEDULE_POLLER_INTERVAL:-1}
   WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE: ${WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE:-100}
   WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK: ${WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK:-0}
-  SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD: ${SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD:-21}
-  SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE: ${SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE:-1000}
-  SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS: ${SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS:-30}
   TENANT_ISOLATED_TASK_CONCURRENCY: ${TENANT_ISOLATED_TASK_CONCURRENCY:-1}
   ANNOTATION_IMPORT_FILE_SIZE_LIMIT: ${ANNOTATION_IMPORT_FILE_SIZE_LIMIT:-2}
   ANNOTATION_IMPORT_MAX_RECORDS: ${ANNOTATION_IMPORT_MAX_RECORDS:-10000}