diff --git a/app/api/search.py b/app/api/search.py
index 50add2b..da861b7 100644
--- a/app/api/search.py
+++ b/app/api/search.py
@@ -15,7 +15,7 @@ from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from core.auth import get_current_user
+from core.auth import get_current_user, get_egress_class
 from core.database import get_session
 from core.utils import setup_logger
 from models.user import User
@@ -139,6 +139,7 @@ def _build_search_debug(pr: PipelineResult) -> SearchDebug:
 async def search(
     q: str,
     user: Annotated[User, Depends(get_current_user)],
+    egress_class: Annotated[str, Depends(get_egress_class)],
     session: Annotated[AsyncSession, Depends(get_session)],
     background_tasks: BackgroundTasks,
     mode: str = Query("hybrid", pattern="^(fts|trgm|vector|hybrid)$"),
@@ -225,6 +226,7 @@ async def search(
             year_to=year_to,
             domain_buckets=[b.strip() for b in domain_bucket.split(",") if b.strip()] if domain_bucket else None,
             exclude_buckets=[b.strip() for b in exclude_bucket.split(",") if b.strip()] if exclude_bucket else None,
+            cloud_egress=(egress_class == "cloud"),
         )
         pr = await run_search(
             session,
diff --git a/app/core/auth.py b/app/core/auth.py
index d35257f..6a16ebd 100644
--- a/app/core/auth.py
+++ b/app/core/auth.py
@@ -31,11 +31,11 @@ def hash_password(password: str) -> str:
     return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
 
 
-def create_access_token(subject: str, expires_minutes: int | None = None) -> str:
+def create_access_token(subject: str, expires_minutes: int | None = None, egress: str = "local") -> str:
     minutes = expires_minutes if expires_minutes is not None else ACCESS_TOKEN_EXPIRE_MINUTES
     now = datetime.now(timezone.utc)
     expire = now + timedelta(minutes=minutes)
-    payload = {"sub": subject, "exp": expire, "iat": int(now.timestamp()), "type": "access"}
+    payload = {"sub": subject, "exp": expire, "iat": int(now.timestamp()), "type": "access", "egress": egress}
     return jwt.encode(payload, settings.jwt_secret, algorithm=ALGORITHM)
 
 
@@ -100,6 +100,15 @@ def verify_totp(code: str, secret: str | None = None) -> bool:
     return totp.verify(code)
 
 
+async def get_egress_class(
+    credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
+) -> str:
+    """토큰 egress claim -> 'cloud'|'local' (갭2 cloud-egress allowlist). claim 부재=local
+    (비파괴; 기존 토큰=신뢰/로컬). 쿼리파라미터 아님 -> 호출자가 끌 수 없음(우회 차단)."""
+    payload = decode_token(credentials.credentials)
+    return (payload or {}).get("egress", "local")
+
+
 async def get_current_user(
     credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
     session: Annotated[AsyncSession, Depends(get_session)],
diff --git a/app/services/search/retrieval_service.py b/app/services/search/retrieval_service.py
index 936ed1e..7a22e76 100644
--- a/app/services/search/retrieval_service.py
+++ b/app/services/search/retrieval_service.py
@@ -78,11 +78,13 @@ class AxisFilter:
     year_to: int | None = None
     domain_buckets: list[str] | None = None    # 377: domain_bucket = ANY (도메인 스코프)
     exclude_buckets: list[str] | None = None    # 377: domain_bucket <> ALL (예: News 제외)
+    cloud_egress: bool = False    # 갭2: 클라우드 소비자 cloud-eligibility allowlist 강제(토큰 claim 유래)
 
     def active(self) -> bool:
         return bool(self.material_types or self.jurisdiction
                     or self.year_from is not None or self.year_to is not None
-                    or self.domain_buckets or self.exclude_buckets)
+                    or self.domain_buckets or self.exclude_buckets
+                    or self.cloud_egress)
 
 
 def _axis_sql(alias: str, af: "AxisFilter | None", params: dict) -> str:
@@ -113,6 +115,16 @@ def _axis_sql(alias: str, af: "AxisFilter | None", params: dict) -> str:
     if af.exclude_buckets:
         cl.append(f"{p}domain_bucket <> ALL(:af_xdb)")
         params["af_xdb"] = af.exclude_buckets
+    if af.cloud_egress:
+        # 갭2 클라우드 egress allowlist(default-deny). restricted 는 _license_sql 별도 차단.
+        cl.append(
+            f"({p}data_origin = 'external' OR ("
+            f"{p}data_origin = 'work' "
+            f"AND {p}domain_bucket IN ('Engineering','Safety','Law') "
+            f"AND ({p}source_channel IS NULL OR {p}source_channel::text NOT IN ('voice','chat','memo')) "
+            f"AND {p}category::text IS DISTINCT FROM 'memo' "
+            f"AND ({p}user_note IS NULL OR {p}user_note = '')))"
+        )
     return " AND " + " AND ".join(cl)