diff --git a/apps/api/tests/integration/test_drift_repository.py b/apps/api/tests/integration/test_drift_repository.py
new file mode 100644
index 00000000..1a8f9e95
--- /dev/null
+++ b/apps/api/tests/integration/test_drift_repository.py
@@ -0,0 +1,83 @@
+"""
+Drift API + Repository 整合測試
+=================================
+打真實 API 端點 (https://awoooi.wooo.work)，使用真實 DB
+
+建立時間: 2026-04-09 (台北時區)
+建立者: Claude Sonnet 4.6 (B5 整合測試)
+
+覆蓋:
+- GET /api/v1/drift/reports — 列出漂移報告（DB 持久化驗證）
+- POST /api/v1/drift/internal/scan — 觸發掃描（CronJob endpoint）
+- 驗證 DB 持久化：scan 後 reports 列表有新資料
+"""
+
+import os
+import time
+
+import httpx
+import pytest
+
+API_BASE = os.environ.get("API_BASE_URL", "https://awoooi.wooo.work")
+
+
+@pytest.fixture
+def client():
+    with httpx.Client(base_url=API_BASE, timeout=30.0) as c:
+        yield c
+
+
+class TestDriftReportsEndpoint:
+
+    def test_list_reports_returns_200(self, client):
+        resp = client.get("/api/v1/drift/reports")
+        assert resp.status_code == 200
+
+    def test_list_reports_has_required_fields(self, client):
+        resp = client.get("/api/v1/drift/reports")
+        data = resp.json()
+        assert "items" in data
+        assert "total" in data
+        assert isinstance(data["items"], list)
+        assert isinstance(data["total"], int)
+
+    def test_list_reports_items_structure(self, client):
+        """若有資料，驗證每筆 report 結構"""
+        resp = client.get("/api/v1/drift/reports")
+        data = resp.json()
+        for item in data["items"]:
+            assert "report_id" in item
+            assert "namespace" in item
+            assert "high_count" in item
+            assert "status" in item
+            assert "created_at" in item
+
+
+class TestDriftInternalScan:
+
+    def test_internal_scan_returns_triggered(self, client):
+        resp = client.post("/api/v1/drift/internal/scan")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data.get("status") == "scan_triggered"
+        assert "namespaces" in data
+
+    def test_internal_scan_db_persisted(self, client):
+        """掃描後等待背景任務，確認 DB 有新資料"""
+        # 記錄掃描前的 report count
+        before = client.get("/api/v1/drift/reports").json()
+        before_total = before["total"]
+
+        # 觸發掃描
+        client.post("/api/v1/drift/internal/scan")
+
+        # 等待背景掃描完成（最多 15 秒）
+        for _ in range(5):
+            time.sleep(3)
+            after = client.get("/api/v1/drift/reports").json()
+            if after["total"] > before_total:
+                break
+
+        after = client.get("/api/v1/drift/reports").json()
+        # 掃描後 DB 應有新資料（total >= before）
+        assert after["total"] >= before_total