diff --git a/api/services/billing_service.py b/api/services/billing_service.py
index 4fc61c6c0f..6ffb1f0cb6 100644
--- a/api/services/billing_service.py
+++ b/api/services/billing_service.py
@@ -31,21 +31,20 @@ class BillingService:
     @classmethod
     def get_info_bulk(cls, tenant_ids: list[str]) -> dict[str, dict]:
         """
-        Temporary bulk billing info fetch. Will be replaced by a real batch API.
+        Bulk billing info fetch via billing API.
 
-        Args:
-            tenant_ids: list of tenant ids
+        Payload: {"tenant_ids": ["t1", "t2", ...]} (max 200 per request)
 
         Returns:
-            Mapping of tenant_id -> billing info dict
+            Mapping of tenant_id -> plan
         """
-        result: dict[str, dict] = {}
-        for tenant_id in tenant_ids:
-            try:
-                result[tenant_id] = cls.get_info(tenant_id)
-            except Exception:
-                logger.exception("Failed to fetch billing info for tenant %s in bulk mode", tenant_id)
-        return result
+
+        try:
+            resp = cls._send_request("POST", "/subscription/plan/batch", json={"tenant_ids": tenant_ids})
+        except Exception:
+            logger.exception("Failed to fetch billing info batch for tenants: %s", tenant_ids)
+
+        return resp.get("data", {})
 
     @classmethod
     def get_tenant_feature_plan_usage_info(cls, tenant_id: str):
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 98543c2928..5972440732 100644
--- a/api/services/clear_free_plan_expired_workflow_run_logs.py
+++ b/api/services/clear_free_plan_expired_workflow_run_logs.py
@@ -8,7 +8,6 @@ import sqlalchemy as sa
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
-from configs import dify_config
 from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from models import WorkflowAppLog, WorkflowNodeExecutionModel, WorkflowRun
@@ -140,9 +139,6 @@ class WorkflowRunCleanup:
         return [WorkflowRunRow(id=row.id, tenant_id=row.tenant_id, created_at=row.created_at) for row in rows]
 
     def _filter_free_tenants(self, tenant_ids: Iterable[str]) -> set[str]:
-        if not dify_config.BILLING_ENABLED:
-            return set(tenant_ids)
-
         tenant_id_list = list(tenant_ids)
         uncached_tenants = [tenant_id for tenant_id in tenant_id_list if tenant_id not in self.billing_cache]
 
@@ -158,8 +154,7 @@ class WorkflowRunCleanup:
                 info = bulk_info.get(tenant_id)
                 if info:
                     try:
-                        raw_plan = info.get("subscription", {}).get("plan")
-                        plan = CloudPlan(raw_plan)
+                        plan = CloudPlan(info)
                     except Exception:
                         logger.exception("Failed to parse billing plan for tenant %s", tenant_id)
                 else: