diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py
index d1e301ae1..8153bef90 100755
--- a/apps/api/app/api/routes/generation_tasks.py
+++ b/apps/api/app/api/routes/generation_tasks.py
@@ -4,6 +4,7 @@ from typing import Any
 
 from app.api.routes._helpers import check_project_access
 from app.auth import AuthenticatedUser, get_current_user
+from packages.middleware.points_gate import points_gate
 from app.core.storage import OSSStorageService, get_storage_service
 from app.core.task_enqueue import (
     GLOBAL_PENDING_LIMIT,
@@ -211,6 +212,7 @@ def _resolve_project_and_library(
 
 
 @router.post("/tasks", response_model=BatchGenerationTaskResponse)
+@points_gate("ai_video", quantity_field="count")
 def create_generation_task(
     request: CreateGenerationTaskRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
diff --git a/tests/unit/test_generation_tasks.py b/tests/unit/test_generation_tasks.py
index ad33cc4c3..3475f9f60 100755
--- a/tests/unit/test_generation_tasks.py
+++ b/tests/unit/test_generation_tasks.py
@@ -17,6 +17,16 @@ from packages.application.generation_tasks import (
 )
 from packages.domain import GenerationTask
 
+import packages.middleware.points_gate as _pg_module
+
+
+@pytest.fixture(autouse=True)
+def _disable_points_gate(monkeypatch):
+    """默认关闭积分闸门，避免影响既有用例。"""
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: False)
+    yield
+
+
 
 @pytest.fixture
 def mock_repo():
diff --git a/tests/unit/test_generation_tasks_points.py b/tests/unit/test_generation_tasks_points.py
new file mode 100644
index 000000000..001095be4
--- /dev/null
+++ b/tests/unit/test_generation_tasks_points.py
@@ -0,0 +1,55 @@
+"""视频生成 积分扣点单元测试 (#1895 P2 step 2.4)"""
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+import packages.middleware.points_gate as _pg_module
+
+
+@pytest.fixture(autouse=True)
+def _enable(monkeypatch):
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
+    yield
+
+
+class TestGenerationTasksPoints:
+    def test_ai_video_base_cost(self):
+        from packages.domain.points_rules import calculate_points_cost
+        assert calculate_points_cost("ai_video", is_member=False) == 4
+        assert calculate_points_cost("ai_video", is_member=True, member_type="monthly") == 2
+
+    def test_ai_video_quantity_scales(self):
+        from packages.domain.points_rules import calculate_points_cost
+        c1 = calculate_points_cost("ai_video", is_member=False, quantity=1)
+        c3 = calculate_points_cost("ai_video", is_member=False, quantity=3)
+        assert c3 > c1
+
+    def test_insufficient_raises_402(self):
+        from fastapi import HTTPException
+        from app.api.routes.generation_tasks import create_generation_task
+        from app.schemas.generation_task import CreateGenerationTaskRequest
+
+        db = MagicMock()
+        cu = MagicMock()
+        cu.user.id = "u1"
+        cu.user.is_member = False
+        cu.user.member_type = None
+        req = CreateGenerationTaskRequest(template_id="t1", asset_ids=["a1"], count=1)
+        with patch("packages.domain.points_service.PointsService") as MS:
+            svc = MagicMock()
+            svc.check_daily_free_clip.return_value = False
+            svc.deduct_points.return_value = {"success": False, "balance": 0}
+            MS.return_value = svc
+            with pytest.raises(HTTPException) as ei:
+                create_generation_task(
+                    request=req, authenticated_user=cu, db=db,
+                    generation_task_repository=MagicMock(), project_repository=MagicMock(),
+                    asset_library_repository=MagicMock(), asset_repository=MagicMock(),
+                )
+            assert ei.value.status_code == 402
+
+    def test_decorator_attached(self):
+        from app.api.routes.generation_tasks import create_generation_task
+        assert hasattr(create_generation_task, "__wrapped__"), "missing @points_gate"