diff --git a/app/core/config.py b/app/core/config.py
index 1e98d42..3ec72f9 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -186,6 +186,59 @@ class Settings(BaseSettings):
         alias="POLICY_SYNC_LOOKBACK_DAYS",
         description="배치 수집 시 오늘 포함 N일 (API 3일 제한 고려)",
     )
+    policy_sync_interval_days: int = Field(
+        default=3,
+        ge=1,
+        le=30,
+        alias="POLICY_SYNC_INTERVAL_DAYS",
+        description="자동 sync 최소 간격(일)",
+    )
+    policy_sync_schedule_hour_kst: int = Field(
+        default=1,
+        ge=0,
+        le=23,
+        alias="POLICY_SYNC_SCHEDULE_HOUR_KST",
+        description="정책 핀 sync 스케줄 확인 시각 (KST)",
+    )
+    policy_admin_user_name: str = Field(
+        default="admin",
+        validation_alias=AliasChoices("POLICY_ADMIN_USER_NAME", "POLICY_ADMIN_NICKNAME"),
+        description="정책 핀 등록에 사용할 user.user_name",
+    )
+    policy_sync_batch_size: int = Field(
+        default=5,
+        ge=1,
+        le=25,
+        alias="POLICY_SYNC_BATCH_SIZE",
+        description="sync/transform/import 1회 배치 건수",
+    )
+    policy_transform_concurrency: int = Field(
+        default=3,
+        ge=1,
+        le=10,
+        alias="POLICY_TRANSFORM_CONCURRENCY",
+        description="정책 pin_content·카드뉴스 Gemini 가공 동시 호출 수",
+    )
+    policy_prune_pipeline_after_import: bool = Field(
+        default=True,
+        alias="POLICY_PRUNE_PIPELINE_AFTER_IMPORT",
+        description="DB INSERT 성공 후 JSONL·로컬 카드뉴스 캐시 제거",
+    )
+    policy_cardnews_keep_local_files: bool = Field(
+        default=False,
+        alias="POLICY_CARDNEWS_KEEP_LOCAL_FILES",
+        description="True면 S3 업로드 후에도 rag/output/policy_cardnews 유지",
+    )
+    policy_sync_merge_documents: bool = Field(
+        default=False,
+        alias="POLICY_SYNC_MERGE_DOCUMENTS",
+        description="False면 sync 수집 시 policy_documents.jsonl을 이번 구간으로 덮어씀",
+    )
+    policy_cardnews_s3_prefix: str = Field(
+        default="policy-cardnews",
+        alias="POLICY_CARDNEWS_S3_PREFIX",
+        description="정책 카드뉴스 S3 object key prefix",
+    )
     gemini_cardnews_image_model: str = Field(
         default="gemini-2.5-flash-image",
         alias="GEMINI_CARDNEWS_IMAGE_MODEL",
@@ -215,9 +268,9 @@ class Settings(BaseSettings):
     rag_enable_rerank: bool = Field(default=False, alias="RAG_ENABLE_RERANK")
     rag_vector_query_mode: str = Field(default="hybrid", alias="RAG_VECTOR_QUERY_MODE")
     policy_cardnews_font_dir: str = Field(
-        default="app/assets/fonts",
+        default="../assets/fonts",
         alias="POLICY_CARDNEWS_FONT_DIR",
-        description="Pretendard 등 TTF 폴더",
+        description="Pretendard 등 폰트 폴더 (app/policy_cardnews 기준 상대 경로)",
     )
 
     # 한국관광공사 TourAPI (공공데이터포털 활용신청 키)
@@ -277,7 +330,7 @@ class Settings(BaseSettings):
     @classmethod
     def _empty_string_policy_cardnews_font_dir(cls, value: object) -> object:
         if value == "":
-            return "app/assets/fonts"
+            return "../assets/fonts"
         return value
 
     @field_validator("policy_cardnews_mascot_dir", mode="before")
diff --git a/app/core/deps.py b/app/core/deps.py
index de34a24..3bbefe5 100644
--- a/app/core/deps.py
+++ b/app/core/deps.py
@@ -11,6 +11,7 @@
 from app.core.exceptions import raise_business_exception
 from app.login.http_auth import get_current_user_id, get_optional_user_id
 from app.models.User import User
+from app.repositories.CardnewsImageS3Repo import CardnewsImageS3Repo
 from app.repositories.CommunityRepo import CommunityRepo
 from app.repositories.ComplaintPetitionRepo import ComplaintPetitionRepo
 from app.repositories.DepartmentRepo import DepartmentRepo
@@ -29,7 +30,9 @@
 from app.services.UserService import UserService
 from app.services.ComplaintEmailService import ComplaintEmailService
 from app.services.FestivalPinService import FestivalPinService
+from app.services.PolicyEventIngestService import PolicyEventIngestService
 from app.services.PolicyPinService import PolicyPinService
+from app.services.internal.PolicyPinSchedulerService import PolicyPinSchedulerService
 from app.services.ComplaintPetitionService import ComplaintPetitionService
 from app.services.FestivalEventIngestService import FestivalEventIngestService
 from app.services.RagRerankService import RagRerankService
@@ -519,3 +522,47 @@ def get_policy_pin_service() -> PolicyPinService:
     PolicyPinService,
     Depends(get_policy_pin_service),
 ]
+
+
+def get_cardnews_image_s3_repo(session: DbSessionDep) -> CardnewsImageS3Repo:
+    return CardnewsImageS3Repo(session)
+
+
+CardnewsImageS3RepoDep = Annotated[CardnewsImageS3Repo, Depends(get_cardnews_image_s3_repo)]
+
+
+def get_policy_event_ingest_service(
+    pin_repo: PinRepoDep,
+    event_pin_repo: EventPinRepoDep,
+    community_repo: CommunityRepoDep,
+    cardnews_image_s3_repo: CardnewsImageS3RepoDep,
+    user_repo: UserRepoDep,
+) -> PolicyEventIngestService:
+    return PolicyEventIngestService(
+        pin_repo=pin_repo,
+        event_pin_repo=event_pin_repo,
+        community_repo=community_repo,
+        cardnews_image_s3_repo=cardnews_image_s3_repo,
+        user_repo=user_repo,
+    )
+
+
+PolicyEventIngestServiceDep = Annotated[
+    PolicyEventIngestService,
+    Depends(get_policy_event_ingest_service),
+]
+
+
+def get_policy_pin_scheduler(request: Request) -> PolicyPinSchedulerService | None:
+    scheduler = getattr(request.app.state, "policy_pin_scheduler", None)
+    if scheduler is None:
+        return None
+    if not isinstance(scheduler, PolicyPinSchedulerService):
+        raise RuntimeError("policy_pin_scheduler is not initialized correctly.")
+    return scheduler
+
+
+PolicyPinSchedulerDep = Annotated[
+    PolicyPinSchedulerService | None,
+    Depends(get_policy_pin_scheduler),
+]
diff --git a/app/main.py b/app/main.py
index 4e2fcb7..c274e21 100644
--- a/app/main.py
+++ b/app/main.py
@@ -23,6 +23,7 @@
 from app.services.RagRetrievalService import RagRetrievalService
 from app.services.VectorStoreService import VectorStoreService
 from app.services.internal.ComplaintPetitionSchedulerService import ComplaintPetitionSchedulerService
+from app.services.internal.PolicyPinSchedulerService import PolicyPinSchedulerService
 from app.services.internal.ai.ComplaintEmailLLMService import ComplaintEmailLLMService
 from app.services.internal.ai.VLMService import VLMService
 from app.services.internal.ai.gemini_retry import parse_gemini_model_list
@@ -154,7 +155,7 @@ async def lifespan(app: FastAPI):
             exc,
         )
 
-    scheduler = None
+    complaint_scheduler = None
     if (
         gemini_api_key_secret is not None
         and getattr(app.state, "vector_store_service", None) is not None
@@ -193,24 +194,39 @@ async def lifespan(app: FastAPI):
                 complaint_llm_service=complaint_llm_service,
                 rag_retrieval_service=rag_retrieval_service,
             )
-            scheduler = ComplaintPetitionSchedulerService(
+            complaint_scheduler = ComplaintPetitionSchedulerService(
                 complaint_email_service=complaint_email_service,
                 s3_util=app.state.s3_util,
             )
-            scheduler.start()
-            app.state.complaint_scheduler = scheduler
+            complaint_scheduler.start()
+            app.state.complaint_scheduler = complaint_scheduler
         except Exception as exc:
             logger.warning("Complaint scheduler initialization failed: %s", exc)
 
+    policy_pin_scheduler = None
+    if settings.policy_news_service_key is not None and gemini_api_key_secret is not None:
+        try:
+            policy_pin_scheduler = PolicyPinSchedulerService(s3_util=app.state.s3_util)
+            policy_pin_scheduler.start()
+            app.state.policy_pin_scheduler = policy_pin_scheduler
+        except Exception as exc:
+            logger.warning("Policy pin scheduler initialization failed: %s", exc)
+
     try:
         yield
     finally:
-        scheduler = getattr(app.state, "complaint_scheduler", None)
-        if scheduler is not None:
+        complaint_scheduler = getattr(app.state, "complaint_scheduler", None)
+        if complaint_scheduler is not None:
             try:
-                await scheduler.stop()
+                await complaint_scheduler.stop()
             except Exception as exc:
                 logger.warning("Complaint scheduler stop failed: %s", exc)
+        policy_pin_scheduler = getattr(app.state, "policy_pin_scheduler", None)
+        if policy_pin_scheduler is not None:
+            try:
+                await policy_pin_scheduler.stop()
+            except Exception as exc:
+                logger.warning("Policy pin scheduler stop failed: %s", exc)
         try:
             await ComplaintEmailPdfService.stop_playwright_browser()
         except Exception as exc:
diff --git a/app/models/CardnewsImageS3.py b/app/models/CardnewsImageS3.py
new file mode 100644
index 0000000..dea4764
--- /dev/null
+++ b/app/models/CardnewsImageS3.py
@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from sqlalchemy import BigInteger, ForeignKey, Identity, String
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.core.database import Base
+from app.models.Community import Community
+
+
+class CardnewsImageS3(Base):
+    __tablename__ = "cardnews_image_s3"
+
+    cardnews_image_s3_id: Mapped[int] = mapped_column(
+        "cardnews_image_s3_id",
+        BigInteger,
+        Identity(),
+        primary_key=True,
+    )
+    community_id: Mapped[int] = mapped_column(
+        "community_id",
+        BigInteger,
+        ForeignKey("community.community_id"),
+        nullable=False,
+    )
+    community: Mapped[Community] = relationship(
+        "Community",
+        foreign_keys=[community_id],
+        back_populates="cardnews_images",
+        lazy="selectin",
+    )
+    cardnews_image_s3_key: Mapped[str] = mapped_column(
+        "cardnews_image_s3_key",
+        String(500),
+        nullable=False,
+    )
+    cardnews_image_s3_url: Mapped[str] = mapped_column(
+        "cardnews_image_s3_url",
+        String(500),
+        nullable=False,
+    )
diff --git a/app/models/Community.py b/app/models/Community.py
index e77af6e..9dc6d48 100644
--- a/app/models/Community.py
+++ b/app/models/Community.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-from sqlalchemy import BigInteger, ForeignKey, Identity
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy import BigInteger, Float, ForeignKey, Identity, String
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.models.BaseEntity import BaseEntity
 
@@ -21,3 +21,15 @@ class Community(BaseEntity):
         ForeignKey("pin.pin_id"),
         nullable=True,
     )
+    community_type: Mapped[str] = mapped_column(
+        "community_type",
+        String(50),
+        nullable=False,
+        server_default="POLICY",
+    )
+    popularity: Mapped[float] = mapped_column("popularity", Float, nullable=False, default=0.0)
+    cardnews_images: Mapped[list["CardnewsImageS3"]] = relationship(
+        "CardnewsImageS3",
+        back_populates="community",
+        lazy="selectin",
+    )
diff --git a/app/models/EventPin.py b/app/models/EventPin.py
index 313e29d..519d918 100644
--- a/app/models/EventPin.py
+++ b/app/models/EventPin.py
@@ -31,12 +31,18 @@ class EventPin(BaseEntity):
         back_populates="event_pin",
         lazy="selectin",
     )
-    festival_api_id: Mapped[int] = mapped_column(
+    festival_api_id: Mapped[int | None] = mapped_column(
         "festival_api_id",
         BigInteger,
         unique=True,
         nullable=True,
     )
+    policy_api_id: Mapped[int | None] = mapped_column(
+        "policy_api_id",
+        BigInteger,
+        unique=True,
+        nullable=True,
+    )
     event_start_time: Mapped[datetime] = mapped_column("event_start_time", DateTime, nullable=False)
     event_end_time: Mapped[datetime] = mapped_column("event_end_time", DateTime, nullable=False)
     discount: Mapped[str | None] = mapped_column("discount", String(255), nullable=True)
diff --git a/app/models/__init__.py b/app/models/__init__.py
index d80d2fd..05dfb18 100644
--- a/app/models/__init__.py
+++ b/app/models/__init__.py
@@ -1,3 +1,5 @@
+from app.models.CardnewsImageS3 import CardnewsImageS3
+from app.models.Community import Community
 from app.models.ComplaintPetition import ComplaintPetition
 from app.models.Department import Department
 from app.models.DbConnectionCheck import DbConnectionCheck
@@ -24,6 +26,8 @@
 from app.models.enum.UserRole import UserRole
 
 __all__ = [
+    "CardnewsImageS3",
+    "Community",
     "DbConnectionCheck",
     "ComplaintPetition",
     "Department",
diff --git a/app/policy_cardnews/template/dispatch.py b/app/policy_cardnews/template/dispatch.py
index 0e12333..b9d5e48 100644
--- a/app/policy_cardnews/template/dispatch.py
+++ b/app/policy_cardnews/template/dispatch.py
@@ -56,12 +56,17 @@
 MASCOT_LAYOUTS = {LAYOUT_COVER, LAYOUT_CTA, LAYOUT_THREE_COL, LAYOUT_NUMBERED}
 
 _REPO_ROOT = Path(__file__).resolve().parents[3]
+# JS import처럼 app/policy_cardnews 패키지 기준 상대 경로 해석
+_POLICY_CARDNEWS_DIR = Path(__file__).resolve().parents[1]
 
 
 def _font_dir() -> Path:
     from app.core.config import settings
 
-    return Path(settings.policy_cardnews_font_dir)
+    raw = Path(settings.policy_cardnews_font_dir)
+    if raw.is_absolute():
+        return raw
+    return (_POLICY_CARDNEWS_DIR / raw).resolve()
 
 
 @dataclass(frozen=True)
@@ -244,6 +249,10 @@ def _load_font(size: int, *, bold: bool = False, extra_bold: bool = False) -> Im
         path = base_dir / name
         if path.is_file():
             return ImageFont.truetype(str(path), size=size)
+    logger.warning(
+        "Pretendard 폰트를 찾지 못해 기본 폰트로 렌더합니다 (font_dir=%s)",
+        base_dir,
+    )
     return ImageFont.load_default()
 
 
diff --git a/app/repositories/CardnewsImageS3Repo.py b/app/repositories/CardnewsImageS3Repo.py
new file mode 100644
index 0000000..4eab67d
--- /dev/null
+++ b/app/repositories/CardnewsImageS3Repo.py
@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.CardnewsImageS3 import CardnewsImageS3
+from app.repositories.BaseRepo import BaseRepo
+
+
+class CardnewsImageS3Repo(BaseRepo[CardnewsImageS3]):
+    def __init__(self, session: AsyncSession) -> None:
+        super().__init__(session, CardnewsImageS3)
+
+    async def list_by_community_id(self, community_id: int) -> list[CardnewsImageS3]:
+        result = await self.session.execute(
+            select(CardnewsImageS3)
+            .where(CardnewsImageS3.community_id == community_id)
+            .order_by(CardnewsImageS3.cardnews_image_s3_id.asc()),
+        )
+        return list(result.scalars().all())
+
+    async def delete_by_community_id(self, community_id: int) -> int:
+        result = await self.session.execute(
+            delete(CardnewsImageS3).where(CardnewsImageS3.community_id == community_id),
+        )
+        await self.session.flush()
+        return int(result.rowcount or 0)
diff --git a/app/repositories/CommunityRepo.py b/app/repositories/CommunityRepo.py
index 217de87..20d1897 100644
--- a/app/repositories/CommunityRepo.py
+++ b/app/repositories/CommunityRepo.py
@@ -19,3 +19,9 @@ async def get_community_id_by_pin_id(self, pin_id: int) -> int | None:
         )
         value = result.scalar_one_or_none()
         return int(value) if value is not None else None
+
+    async def get_by_pin_id(self, pin_id: int) -> Community | None:
+        result = await self.session.execute(
+            select(Community).where(Community.pin_id == pin_id).limit(1),
+        )
+        return result.scalar_one_or_none()
diff --git a/app/repositories/EventPinRepo.py b/app/repositories/EventPinRepo.py
index 8f13ddb..4aa165d 100644
--- a/app/repositories/EventPinRepo.py
+++ b/app/repositories/EventPinRepo.py
@@ -39,3 +39,25 @@ async def count_festival_pins(self) -> int:
             ),
         )
         return result.scalar_one()
+
+    async def get_by_policy_api_id(self, policy_api_id: int) -> EventPin | None:
+        result = await self.session.execute(
+            select(EventPin)
+            .where(EventPin.policy_api_id == policy_api_id)
+            .options(*_EVENT_PIN_LOAD_OPTIONS),
+        )
+        return result.scalar_one_or_none()
+
+    async def list_policy_api_ids(self) -> set[int]:
+        result = await self.session.execute(
+            select(EventPin.policy_api_id).where(EventPin.policy_api_id.is_not(None)),
+        )
+        return {int(row) for row in result.scalars().all() if row is not None}
+
+    async def count_policy_pins(self) -> int:
+        result = await self.session.execute(
+            select(func.count(EventPin.event_pin_id)).where(
+                EventPin.policy_api_id.is_not(None),
+            ),
+        )
+        return result.scalar_one()
diff --git a/app/repositories/UserRepo.py b/app/repositories/UserRepo.py
index b1418a0..c46d651 100644
--- a/app/repositories/UserRepo.py
+++ b/app/repositories/UserRepo.py
@@ -21,5 +21,23 @@ async def get_by_email(self, email: str) -> User | None:
         result = await self.session.execute(select(User).where(User.email == email))
         return result.scalar_one_or_none()
 
+    async def get_by_nickname(self, nickname: str) -> User | None:
+        normalized = (nickname or "").strip()
+        if not normalized:
+            return None
+        result = await self.session.execute(
+            select(User).where(User.nickname == normalized).limit(1),
+        )
+        return result.scalar_one_or_none()
+
+    async def get_by_user_name(self, user_name: str) -> User | None:
+        normalized = (user_name or "").strip()
+        if not normalized:
+            return None
+        result = await self.session.execute(
+            select(User).where(User.user_name == normalized).limit(1),
+        )
+        return result.scalar_one_or_none()
+
     async def get_all_users(self) -> list[User]:
         return await self.get_all()
diff --git a/app/routes/PolicyAdminRoute.py b/app/routes/PolicyAdminRoute.py
new file mode 100644
index 0000000..7499404
--- /dev/null
+++ b/app/routes/PolicyAdminRoute.py
@@ -0,0 +1,191 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException, Query
+
+from app.core.codes import ErrorCode, SuccessCode
+from app.core.config import settings
+from app.core.deps import (
+    AdminUserIdDep,
+    PolicyEventIngestServiceDep,
+    PolicyPinSchedulerDep,
+    PolicyPinServiceDep,
+    S3UtilDep,
+)
+from app.core.responses import SuccessEnvelope, success_response
+from app.schemas.PolicyAdminDTO import (
+    PolicyImportBatchResult,
+    PolicyPipelineStatusResult,
+    PolicySyncResult,
+    PolicyTransformBatchResult,
+)
+from app.utils.policy_news_parse import validate_yyyymmdd
+
+router = APIRouter(prefix="/policy-admin", tags=["policy-admin"])
+
+_MAX_BATCH = 25
+
+
+def _runtime_error_response(exc: RuntimeError) -> HTTPException:
+    detail = str(exc)
+    if "GEMINI" in detail or "POLICY" in detail or "S3" in detail:
+        status = ErrorCode.VLM_NOT_CONFIGURED.http_status
+    else:
+        status = ErrorCode.INTERNAL_SERVER_ERROR.http_status
+    return HTTPException(status_code=status, detail=detail)
+
+
+def _clamp_batch(value: int | None) -> int:
+    if value is None:
+        return settings.policy_sync_batch_size
+    return min(max(value, 1), _MAX_BATCH)
+
+
+def _validate_optional_date_range(
+    start_date: str | None,
+    end_date: str | None,
+) -> tuple[str | None, str | None]:
+    if start_date is None and end_date is None:
+        return None, None
+    if start_date is None or end_date is None:
+        raise HTTPException(
+            status_code=ErrorCode.BAD_REQUEST.http_status,
+            detail="start_date와 end_date는 함께 지정해야 합니다.",
+        )
+    parsed_start = validate_yyyymmdd(start_date, label="start_date")
+    parsed_end = validate_yyyymmdd(end_date, label="end_date")
+    if parsed_start > parsed_end:
+        raise HTTPException(
+            status_code=ErrorCode.BAD_REQUEST.http_status,
+            detail="start_date는 end_date보다 이후일 수 없습니다.",
+        )
+    return parsed_start, parsed_end
+
+
+@router.post(
+    "/sync",
+    response_model=SuccessEnvelope[PolicySyncResult],
+    summary="정책 핀 3일 lookback 수집 → 배치 transform → 배치 DB 적재",
+    description=(
+        "오늘 포함 최근 POLICY_SYNC_LOOKBACK_DAYS(기본 3일) 구간을 수집하고, "
+        "신규 NewsItemId만 POLICY_SYNC_BATCH_SIZE 단위로 LLM 가공·DB INSERT합니다."
+    ),
+)
+async def sync_policy_pins(
+    service: PolicyPinServiceDep,
+    ingest_service: PolicyEventIngestServiceDep,
+    s3_util: S3UtilDep,
+    _admin_uid: AdminUserIdDep,
+    start_date: str | None = Query(default=None, min_length=8, max_length=8, description="YYYYMMDD"),
+    end_date: str | None = Query(default=None, min_length=8, max_length=8, description="YYYYMMDD"),
+    transform_limit: int | None = Query(default=None, ge=1, le=100, description="가공 최대 건수"),
+    batch_size: int | None = Query(default=None, ge=1, le=25, description="배치 크기"),
+) -> SuccessEnvelope[PolicySyncResult]:
+    try:
+        validated_start, validated_end = _validate_optional_date_range(start_date, end_date)
+        body = await service.sync_pipeline(
+            ingest_service=ingest_service,
+            s3_util=s3_util,
+            start_date=validated_start,
+            end_date=validated_end,
+            transform_limit=transform_limit,
+            batch_size=_clamp_batch(batch_size),
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=ErrorCode.BAD_REQUEST.http_status, detail=str(exc)) from exc
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=ErrorCode.NOT_FOUND.http_status, detail=str(exc)) from exc
+    except RuntimeError as exc:
+        raise _runtime_error_response(exc) from exc
+
+    return success_response(result=body, success_code=SuccessCode.CREATED)
+
+
+@router.post(
+    "/transform-batch",
+    response_model=SuccessEnvelope[PolicyTransformBatchResult],
+    summary="원문 JSONL 배치 가공 (신규 건만)",
+)
+async def transform_policy_batch(
+    service: PolicyPinServiceDep,
+    ingest_service: PolicyEventIngestServiceDep,
+    s3_util: S3UtilDep,
+    _admin_uid: AdminUserIdDep,
+    batch_size: int | None = Query(default=None, ge=1, le=25, description="이번 배치 가공 건수"),
+) -> SuccessEnvelope[PolicyTransformBatchResult]:
+    try:
+        db_ids = await ingest_service.get_imported_policy_api_ids()
+        body = await service.transform_batch(
+            batch_size=_clamp_batch(batch_size),
+            s3_util=s3_util,
+            db_policy_api_ids=db_ids,
+        )
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=ErrorCode.NOT_FOUND.http_status, detail=str(exc)) from exc
+    except RuntimeError as exc:
+        raise _runtime_error_response(exc) from exc
+
+    return success_response(result=body, success_code=SuccessCode.CREATED)
+
+
+@router.post(
+    "/import-batch",
+    response_model=SuccessEnvelope[PolicyImportBatchResult],
+    summary="handoff JSONL 배치 DB 적재",
+)
+async def import_policy_batch(
+    ingest_service: PolicyEventIngestServiceDep,
+    _admin_uid: AdminUserIdDep,
+    batch_size: int | None = Query(default=None, ge=1, le=25, description="이번 배치 INSERT 건수"),
+) -> SuccessEnvelope[PolicyImportBatchResult]:
+    try:
+        body = await ingest_service.import_handoff_batch(
+            import_all=False,
+            limit=_clamp_batch(batch_size),
+        )
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=ErrorCode.NOT_FOUND.http_status, detail=str(exc)) from exc
+    except RuntimeError as exc:
+        raise _runtime_error_response(exc) from exc
+
+    return success_response(result=body, success_code=SuccessCode.CREATED)
+
+
+@router.get(
+    "/status",
+    response_model=SuccessEnvelope[PolicyPipelineStatusResult],
+    summary="정책 핀 파이프라인 상태",
+)
+async def policy_pipeline_status(
+    ingest_service: PolicyEventIngestServiceDep,
+    _admin_uid: AdminUserIdDep,
+) -> SuccessEnvelope[PolicyPipelineStatusResult]:
+    body = await ingest_service.get_pipeline_status()
+    return success_response(result=body, success_code=SuccessCode.OK)
+
+
+@router.post(
+    "/scheduler/run-once",
+    response_model=SuccessEnvelope[PolicySyncResult],
+    summary="정책 핀 스케줄러 즉시 1회 실행 (테스트용, 간격 무시)",
+)
+async def run_policy_scheduler_once(
+    service: PolicyPinServiceDep,
+    ingest_service: PolicyEventIngestServiceDep,
+    s3_util: S3UtilDep,
+    scheduler: PolicyPinSchedulerDep,
+    _admin_uid: AdminUserIdDep,
+) -> SuccessEnvelope[PolicySyncResult]:
+    try:
+        if scheduler is not None:
+            body = await scheduler.run_once_now(force=True)
+        else:
+            body = await service.sync_pipeline(
+                ingest_service=ingest_service,
+                s3_util=s3_util,
+            )
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=ErrorCode.NOT_FOUND.http_status, detail=str(exc)) from exc
+    except RuntimeError as exc:
+        raise _runtime_error_response(exc) from exc
+
+    return success_response(result=body, success_code=SuccessCode.CREATED)
diff --git a/app/routes/PolicyPinRoute.py b/app/routes/PolicyPinRoute.py
index 3c3877c..bd4095e 100644
--- a/app/routes/PolicyPinRoute.py
+++ b/app/routes/PolicyPinRoute.py
@@ -3,7 +3,7 @@
 from fastapi import APIRouter, HTTPException, Query
 
 from app.core.codes import ErrorCode, SuccessCode
-from app.core.deps import PolicyPinServiceDep
+from app.core.deps import PolicyPinServiceDep, S3UtilDep
 from app.core.responses import SuccessEnvelope, success_response
 from app.schemas.PolicyPinDTO import (
     PolicyPinHandoffResult,
@@ -76,6 +76,7 @@ async def search_policy_news(
 )
 async def transform_policy_documents(
     service: PolicyPinServiceDep,
+    s3_util: S3UtilDep,
     limit: int | None = Query(
         default=None,
         ge=1,
@@ -84,7 +85,7 @@ async def transform_policy_documents(
     ),
 ) -> SuccessEnvelope[PolicyPinTransformResult]:
     try:
-        body = await service.transform_and_save(limit=limit)
+        body = await service.transform_and_save(limit=limit, s3_util=s3_util)
     except FileNotFoundError as exc:
         raise HTTPException(
             status_code=ErrorCode.NOT_FOUND.http_status,
diff --git a/app/routes/__init__.py b/app/routes/__init__.py
index 12bed54..12d94d0 100644
--- a/app/routes/__init__.py
+++ b/app/routes/__init__.py
@@ -6,6 +6,7 @@
 from app.routes.TestRoute import router as test_router
 from app.routes.VectorTestRoute import router as vector_test_router
 from app.routes.FestivalPinRoute import router as festival_pin_router
+from app.routes.PolicyAdminRoute import router as policy_admin_router
 from app.routes.PolicyPinRoute import router as policy_pin_router
 from app.routes.FestivalAdminRoute import router as festival_admin_router
 from app.core.config import settings
@@ -20,6 +21,7 @@
     {"router": vector_test_router, "disabled_envs": {"dev", "prod"}},
     {"router": festival_pin_router, "disabled_envs": {"prod"}},
     {"router": policy_pin_router, "disabled_envs": {"prod"}},
+    {"router": policy_admin_router, "disabled_envs": {"prod"}},
     {"router": festival_admin_router, "disabled_envs": set()},
 )
 
diff --git a/app/schemas/PolicyAdminDTO.py b/app/schemas/PolicyAdminDTO.py
new file mode 100644
index 0000000..535a5d7
--- /dev/null
+++ b/app/schemas/PolicyAdminDTO.py
@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+from app.schemas.PolicyPinDTO import PolicyPinHandoffDTO, PolicyPinSearchResult, PolicyPinTransformResult
+
+
+class PolicyBatchAction(str, Enum):
+    CREATED = "created"
+    SKIPPED = "skipped"
+    ERROR = "error"
+
+
+class PolicyBatchItemResult(BaseModel):
+    policy_api_id: int | None = None
+    pin_title: str | None = None
+    action: PolicyBatchAction
+    message: str | None = None
+
+
+class PolicyImportBatchResult(BaseModel):
+    inserted_count: int
+    skipped_duplicate_count: int
+    pending_import_count: int
+    error_count: int
+    errors: list[dict[str, Any]] = Field(default_factory=list)
+    items: list[PolicyBatchItemResult] = Field(default_factory=list)
+    pin_ids: list[int] = Field(default_factory=list)
+    requested_batch_size: int | None = None
+    hint: str | None = None
+
+
+class PolicyTransformBatchResult(PolicyPinTransformResult):
+    requested_batch_size: int | None = None
+    batches_run: int = 1
+
+
+class PolicyPipelineStatusResult(BaseModel):
+    query_start_date: str | None = None
+    query_end_date: str | None = None
+    last_sync_at: str | None = None
+    documents_count: int
+    handoff_count: int
+    db_policy_count: int
+    pending_transform_count: int
+    pending_import_count: int
+    is_caught_up: bool = Field(
+        default=False,
+        description="미가공·미적재 건이 없어 DB와 파이프라인이 동기화된 상태",
+    )
+    hint: str | None = None
+
+
+class PolicySyncResult(BaseModel):
+    query_start_date: str
+    query_end_date: str
+    search: PolicyPinSearchResult
+    transform: PolicyPinTransformResult
+    import_result: PolicyImportBatchResult
+    hint: str | None = None
diff --git a/app/schemas/PolicyPinDTO.py b/app/schemas/PolicyPinDTO.py
index 8c4b4a5..120d566 100644
--- a/app/schemas/PolicyPinDTO.py
+++ b/app/schemas/PolicyPinDTO.py
@@ -88,6 +88,12 @@ class PolicyPinTransformResult(BaseModel):
         description="output_path JSONL에 저장된 행과 동일 (DB 전달 4필드)",
     )
     hint: str | None = None
+    skipped_duplicate_count: int = 0
+    pending_count: int = 0
+    remaining_pending_count: int = Field(
+        default=0,
+        description="가공 후에도 handoff·DB에 없는 원문 건수",
+    )
 
 
 class PolicyPinHandoffResult(BaseModel):
diff --git a/app/services/PolicyEventIngestService.py b/app/services/PolicyEventIngestService.py
new file mode 100644
index 0000000..6510653
--- /dev/null
+++ b/app/services/PolicyEventIngestService.py
@@ -0,0 +1,362 @@
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, time
+from pathlib import Path
+from typing import Any
+from zoneinfo import ZoneInfo
+
+from anyio import to_thread
+from app.core.config import settings
+from app.models.CardnewsImageS3 import CardnewsImageS3
+from app.models.Community import Community
+from app.models.EventPin import EventPin
+from app.models.Pin import Pin
+from app.models.enum.PinType import PinType
+from app.models.enum.ToneType import ToneType
+from app.repositories.CardnewsImageS3Repo import CardnewsImageS3Repo
+from app.repositories.CommunityRepo import CommunityRepo
+from app.repositories.EventPinRepo import EventPinRepo
+from app.repositories.PinRepo import PinRepo
+from app.repositories.UserRepo import UserRepo
+from app.schemas.PolicyAdminDTO import (
+    PolicyBatchAction,
+    PolicyBatchItemResult,
+    PolicyImportBatchResult,
+    PolicyPipelineStatusResult,
+)
+from app.services.policy_pipeline_cleanup import cleanup_after_policy_import
+from app.services.policy_pin_transform import (
+    POLICY_DOCUMENTS_PATH,
+    POLICY_HANDOFF_PATH,
+    POLICY_SYNC_META_PATH,
+    count_pending_transform,
+    load_jsonl_rows,
+    load_rows_by_content_id,
+    parse_policy_api_id,
+)
+from app.utils.policy_news_parse import approve_date_to_yyyymmdd, parse_policy_datetime
+
+logger = logging.getLogger(__name__)
+
+_KST = ZoneInfo("Asia/Seoul")
+_IMPORT_REPORT_PATH = POLICY_HANDOFF_PATH.with_name("policy_import_batch_report.json")
+_COMMUNITY_TYPE_POLICY = "POLICY"
+
+
+def _parse_event_datetime(value: Any) -> datetime:
+    if value is None:
+        raise ValueError("event datetime 없음")
+    if isinstance(value, datetime):
+        return value.replace(tzinfo=None)
+    text = str(value).strip()
+    if not text:
+        raise ValueError("event datetime 비어 있음")
+    if len(text) == 8 and text.isdigit():
+        year, month, day = int(text[:4]), int(text[4:6]), int(text[6:8])
+        return datetime(year, month, day)
+    parsed = parse_policy_datetime(text)
+    if parsed is not None:
+        return parsed.replace(tzinfo=None)
+    raise ValueError(f"event datetime 파싱 실패: {value!r}")
+
+
+def _event_window_from_row(row: dict[str, Any]) -> tuple[datetime, datetime]:
+    start_raw = row.get("event_start_time") or approve_date_to_yyyymmdd(row.get("approve_date"))
+    end_raw = row.get("event_end_time") or start_raw
+    start_dt = _parse_event_datetime(start_raw)
+    end_dt = _parse_event_datetime(end_raw)
+    if end_dt < start_dt:
+        end_dt = start_dt
+    start_at = datetime.combine(start_dt.date(), time.min)
+    end_at = datetime.combine(end_dt.date(), time.max.replace(microsecond=0))
+    return start_at, end_at
+
+
+class PolicyEventIngestService:
+    def __init__(
+        self,
+        *,
+        pin_repo: PinRepo,
+        event_pin_repo: EventPinRepo,
+        community_repo: CommunityRepo,
+        cardnews_image_s3_repo: CardnewsImageS3Repo,
+        user_repo: UserRepo,
+    ) -> None:
+        self._pin_repo = pin_repo
+        self._event_pin_repo = event_pin_repo
+        self._community_repo = community_repo
+        self._cardnews_image_s3_repo = cardnews_image_s3_repo
+        self._user_repo = user_repo
+
+    async def commit(self) -> None:
+        await self._pin_repo.commit()
+
+    async def rollback(self) -> None:
+        await self._pin_repo.rollback()
+
+    async def get_imported_policy_api_ids(self) -> set[int]:
+        return await self._event_pin_repo.list_policy_api_ids()
+
+    async def resolve_admin_uid(self) -> str:
+        user_name = settings.policy_admin_user_name.strip()
+        user = await self._user_repo.get_by_user_name(user_name)
+        if user is None:
+            raise RuntimeError(
+                f"정책 핀 등록용 사용자를 찾을 수 없습니다 (user_name={user_name!r}).",
+            )
+        return str(user.uid)
+
+    async def import_handoff_batch(
+        self,
+        *,
+        admin_uid: str | None = None,
+        import_all: bool = True,
+        limit: int | None = None,
+    ) -> PolicyImportBatchResult:
+        handoff_rows = await to_thread.run_sync(load_jsonl_rows, POLICY_HANDOFF_PATH)
+        db_ids = await self.get_imported_policy_api_ids()
+        if not handoff_rows:
+            documents = await to_thread.run_sync(load_jsonl_rows, POLICY_DOCUMENTS_PATH)
+            pending_transform = count_pending_transform(documents, {}, db_policy_api_ids=db_ids)
+            if pending_transform > 0:
+                raise FileNotFoundError(
+                    f"핸드오프 JSONL 없음: {POLICY_HANDOFF_PATH}. "
+                    f"transform을 먼저 실행하세요. (미가공 {pending_transform}건)",
+                )
+            effective_batch = None if import_all else limit
+            return PolicyImportBatchResult(
+                inserted_count=0,
+                skipped_duplicate_count=0,
+                pending_import_count=0,
+                error_count=0,
+                requested_batch_size=effective_batch,
+                hint=(
+                    f"DB에 policy 핀 {len(db_ids)}건이 있어 적재 완료 상태입니다. "
+                    "handoff JSONL이 비어 있는 것은 import 후 캐시 정리로 정상입니다."
+                ),
+            )
+
+        uid = admin_uid or await self.resolve_admin_uid()
+        inserted_count = 0
+        skipped_duplicate_count = 0
+        errors: list[dict[str, Any]] = []
+        items: list[PolicyBatchItemResult] = []
+        pin_ids: list[int] = []
+        prune_policy_api_ids: set[int] = set()
+
+        pending_rows: list[dict[str, Any]] = []
+        for row in handoff_rows:
+            policy_api_id = parse_policy_api_id(row)
+            if policy_api_id is None:
+                continue
+            if policy_api_id in db_ids:
+                skipped_duplicate_count += 1
+                prune_policy_api_ids.add(policy_api_id)
+                items.append(
+                    PolicyBatchItemResult(
+                        policy_api_id=policy_api_id,
+                        pin_title=str(row.get("title") or row.get("pin_title") or ""),
+                        action=PolicyBatchAction.SKIPPED,
+                        message="DB에 이미 존재",
+                    ),
+                )
+            else:
+                pending_rows.append(row)
+
+        target_rows = pending_rows if import_all else pending_rows[: (limit or len(pending_rows))]
+        if limit is not None and import_all:
+            target_rows = pending_rows[:limit]
+
+        for row in target_rows:
+            policy_api_id = parse_policy_api_id(row)
+            if policy_api_id is None:
+                errors.append({"row": row, "error": "policy_api_id 없음"})
+                continue
+            if policy_api_id in db_ids:
+                skipped_duplicate_count += 1
+                prune_policy_api_ids.add(policy_api_id)
+                continue
+            try:
+                async with self._pin_repo.session.begin_nested():
+                    pin_id = await self._insert_policy_pin(admin_uid=uid, row=row, policy_api_id=policy_api_id)
+                inserted_count += 1
+                db_ids.add(policy_api_id)
+                prune_policy_api_ids.add(policy_api_id)
+                pin_ids.append(pin_id)
+                items.append(
+                    PolicyBatchItemResult(
+                        policy_api_id=policy_api_id,
+                        pin_title=str(row.get("title") or row.get("pin_title") or ""),
+                        action=PolicyBatchAction.CREATED,
+                    ),
+                )
+            except Exception as exc:
+                logger.exception("policy import failed policy_api_id=%s", policy_api_id)
+                errors.append(
+                    {
+                        "policy_api_id": policy_api_id,
+                        "pin_title": row.get("title") or row.get("pin_title"),
+                        "error": str(exc),
+                    },
+                )
+                items.append(
+                    PolicyBatchItemResult(
+                        policy_api_id=policy_api_id,
+                        pin_title=str(row.get("title") or row.get("pin_title") or ""),
+                        action=PolicyBatchAction.ERROR,
+                        message=str(exc),
+                    ),
+                )
+
+        await self._pin_repo.commit()
+
+        if settings.policy_prune_pipeline_after_import and prune_policy_api_ids:
+            await to_thread.run_sync(cleanup_after_policy_import, prune_policy_api_ids)
+
+        pending_import = 0
+        for row in handoff_rows:
+            policy_api_id = parse_policy_api_id(row)
+            if policy_api_id is not None and policy_api_id not in db_ids:
+                pending_import += 1
+
+        report = {
+            "inserted_count": inserted_count,
+            "skipped_duplicate_count": skipped_duplicate_count,
+            "errors": errors,
+        }
+        _IMPORT_REPORT_PATH.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
+
+        effective_batch = None if import_all else limit
+        return PolicyImportBatchResult(
+            inserted_count=inserted_count,
+            skipped_duplicate_count=skipped_duplicate_count,
+            pending_import_count=pending_import,
+            error_count=len(errors),
+            errors=errors,
+            items=items,
+            pin_ids=pin_ids,
+            requested_batch_size=effective_batch,
+        )
+
+    async def _insert_policy_pin(
+        self,
+        *,
+        admin_uid: str,
+        row: dict[str, Any],
+        policy_api_id: int,
+    ) -> int:
+        title = str(row.get("title") or row.get("pin_title") or "").strip()[:100]
+        content = str(row.get("pin_content") or "").strip()
+        if not title or not content:
+            raise ValueError("title 또는 pin_content가 비어 있음")
+
+        start_at, end_at = _event_window_from_row(row)
+
+        pin = Pin(
+            uid=admin_uid,
+            pin_type=PinType.POLICY,
+            pin_title=title,
+            pin_content=content,
+            tone_type=ToneType.NONE,
+            like_count=0,
+            view_count=0,
+        )
+        await self._pin_repo.save(pin, flush_immediately=True)
+
+        event_pin = EventPin(
+            pin_id=pin.pin_id,
+            festival_api_id=None,
+            policy_api_id=policy_api_id,
+            event_start_time=start_at,
+            event_end_time=end_at,
+            discount=None,
+        )
+        await self._event_pin_repo.save(event_pin, flush_immediately=True)
+
+        community = Community(
+            pin_id=pin.pin_id,
+            community_type=_COMMUNITY_TYPE_POLICY,
+            popularity=0.0,
+        )
+        await self._community_repo.save(community, flush_immediately=True)
+
+        cardnews_images = row.get("cardnews_images") or []
+
+        for image in cardnews_images:
+            if not isinstance(image, dict):
+                continue
+            key = str(image.get("key") or "").strip()
+            url = str(image.get("url") or "").strip()
+            if not key or not url:
+                continue
+            entity = CardnewsImageS3(
+                community_id=community.community_id,
+                cardnews_image_s3_key=key,
+                cardnews_image_s3_url=url,
+            )
+            await self._cardnews_image_s3_repo.save(entity, flush_immediately=True)
+
+        return int(pin.pin_id)
+
+    async def get_pipeline_status(self) -> PolicyPipelineStatusResult:
+        meta = self._load_sync_meta()
+        documents = load_jsonl_rows(POLICY_DOCUMENTS_PATH)
+        handoff_by_id = load_rows_by_content_id(POLICY_HANDOFF_PATH)
+        db_ids = await self.get_imported_policy_api_ids()
+
+        pending_transform = count_pending_transform(documents, handoff_by_id, db_policy_api_ids=db_ids)
+        pending_import = 0
+        for row in handoff_by_id.values():
+            policy_api_id = parse_policy_api_id(row)
+            if policy_api_id is not None and policy_api_id not in db_ids:
+                pending_import += 1
+
+        is_caught_up = pending_transform == 0 and pending_import == 0
+        hint: str | None = None
+        if is_caught_up and len(db_ids) > 0:
+            hint = (
+                f"DB에 policy 핀 {len(db_ids)}건 반영 완료. "
+                "handoff_count=0은 import 후 JSONL 캐시 정리로 정상입니다."
+            )
+        elif pending_transform > 0:
+            hint = f"미가공 {pending_transform}건 — transform-batch 또는 sync를 실행하세요."
+        elif pending_import > 0:
+            hint = f"미적재 {pending_import}건 — import-batch 또는 sync를 실행하세요."
+
+        return PolicyPipelineStatusResult(
+            query_start_date=meta.get("query_start_date"),
+            query_end_date=meta.get("query_end_date"),
+            last_sync_at=meta.get("last_sync_at"),
+            documents_count=len(documents),
+            handoff_count=len(handoff_by_id),
+            db_policy_count=len(db_ids),
+            pending_transform_count=pending_transform,
+            pending_import_count=pending_import,
+            is_caught_up=is_caught_up,
+            hint=hint,
+        )
+
+    @staticmethod
+    def _load_sync_meta() -> dict[str, Any]:
+        if not POLICY_SYNC_META_PATH.is_file():
+            return {}
+        try:
+            return json.loads(POLICY_SYNC_META_PATH.read_text(encoding="utf-8"))
+        except json.JSONDecodeError:
+            return {}
+
+    @staticmethod
+    def write_sync_meta(
+        *,
+        query_start_date: str,
+        query_end_date: str,
+    ) -> None:
+        meta = {
+            "last_sync_at": datetime.now(_KST).isoformat(timespec="seconds"),
+            "query_start_date": query_start_date,
+            "query_end_date": query_end_date,
+        }
+        POLICY_SYNC_META_PATH.parent.mkdir(parents=True, exist_ok=True)
+        POLICY_SYNC_META_PATH.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")
diff --git a/app/services/PolicyPinService.py b/app/services/PolicyPinService.py
index d529148..f1ab01e 100644
--- a/app/services/PolicyPinService.py
+++ b/app/services/PolicyPinService.py
@@ -8,6 +8,13 @@
 
 from app.clients.PolicyNewsClient import PolicyNewsClient
 from app.core.config import settings
+from app.schemas.PolicyAdminDTO import (
+    PolicyBatchAction,
+    PolicyBatchItemResult,
+    PolicyImportBatchResult,
+    PolicySyncResult,
+    PolicyTransformBatchResult,
+)
 from app.schemas.PolicyPinDTO import (
     PolicyPinHandoffDTO,
     PolicyPinHandoffResult,
@@ -15,11 +22,17 @@
     PolicyPinSourceDTO,
     PolicyPinTransformResult,
 )
+from app.services.PolicyEventIngestService import PolicyEventIngestService
+from app.services.policy_pipeline_cleanup import prune_pipeline_imported
 from app.services.policy_pin_transform import (
     POLICY_DOCUMENTS_PATH,
     POLICY_HANDOFF_PATH,
+    load_jsonl_rows,
+    load_rows_by_content_id,
+    merge_documents,
     transform_documents_jsonl,
 )
+from app.utils.S3Util import S3Util
 from rag.scripts.chunk_module import write_jsonl
 from rag.scripts.fetch_policy_news import fetch_policy_documents
 
@@ -27,7 +40,6 @@
 
 
 async def _enrich_cover_image_urls(documents: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    # OpenAPI에 이미지 URL이 없을 때 원문 페이지에서 표지용 URL을 보충
     from app.utils.policy_news_parse import enrich_cover_image_urls as resolve_cover_image_urls
 
     async def enrich_one(doc: dict[str, Any]) -> dict[str, Any]:
@@ -58,6 +70,7 @@ async def search_and_save(
         start_date: str,
         end_date: str,
         limit: int | None = 10,
+        merge: bool = True,
     ) -> PolicyPinSearchResult:
         fetch_limit: int | None
         if limit is None:
@@ -76,10 +89,15 @@ async def search_and_save(
         documents = await _enrich_cover_image_urls(documents)
 
         path = self.documents_path()
-        if documents:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        if merge:
+            existing = load_rows_by_content_id(path)
+            merged = merge_documents(existing, documents)
+            write_jsonl(path, merged)
+            total_in_file = len(merged)
+        else:
             write_jsonl(path, documents)
-        elif not path.is_file():
-            write_jsonl(path, [])
+            total_in_file = len(documents)
 
         pins = [PolicyPinSourceDTO.model_validate(doc) for doc in documents]
         hint: str | None
@@ -87,13 +105,12 @@ async def search_and_save(
             hint = (
                 f"수집 0건 (API 호출 {stats.get('chunks', 0)}회). "
                 "정책뉴스 API는 승인일 기준이며 미래 날짜 구간은 비어 있습니다. "
-                "최근 3일로 다시 시도하세요 (예: start_date=20260522, end_date=20260524). "
-                "기존 policy_documents.jsonl은 유지됩니다."
+                f"JSONL 총 {total_in_file}건 유지."
             )
         else:
             hint = (
-                f"{len(pins)}건을 {path.name}에 저장했습니다. "
-                "다음: POST /policy-pins/transform"
+                f"이번 수집 {len(pins)}건, {path.name} 총 {total_in_file}건. "
+                "다음: POST /policy-pins/transform 또는 /policy-admin/sync"
             )
         return PolicyPinSearchResult(
             query_start_date=start_date,
@@ -110,11 +127,207 @@ async def transform_and_save(
         *,
         limit: int | None = None,
         model: str | None = None,
+        s3_util: S3Util | None = None,
+        db_policy_api_ids: set[int] | None = None,
     ) -> PolicyPinTransformResult:
-        result = await transform_documents_jsonl(limit=limit, model=model)
-        if result.processed_count > 0:
-            return result
-        return result
+        return await transform_documents_jsonl(
+            limit=limit,
+            model=model,
+            s3_util=s3_util,
+            db_policy_api_ids=db_policy_api_ids,
+            merge_handoff=True,
+        )
+
+    async def transform_batch(
+        self,
+        *,
+        batch_size: int | None = None,
+        model: str | None = None,
+        s3_util: S3Util | None = None,
+        db_policy_api_ids: set[int] | None = None,
+    ) -> PolicyTransformBatchResult:
+        size = batch_size or settings.policy_sync_batch_size
+        result = await self.transform_and_save(
+            limit=size,
+            model=model,
+            s3_util=s3_util,
+            db_policy_api_ids=db_policy_api_ids,
+        )
+        return PolicyTransformBatchResult(
+            **result.model_dump(),
+            requested_batch_size=size,
+            batches_run=1,
+        )
+
+    async def sync_pipeline(
+        self,
+        *,
+        ingest_service: PolicyEventIngestService,
+        s3_util: S3Util,
+        start_date: str | None = None,
+        end_date: str | None = None,
+        transform_limit: int | None = None,
+        batch_size: int | None = None,
+    ) -> PolicySyncResult:
+        if start_date is None or end_date is None:
+            start_date, end_date = self.default_date_range()
+
+        effective_batch = batch_size or settings.policy_sync_batch_size
+
+        if settings.policy_prune_pipeline_after_import:
+            db_ids_before = await ingest_service.get_imported_policy_api_ids()
+            prune_pipeline_imported(db_ids_before)
+
+        search = await self.search_and_save(
+            start_date=start_date,
+            end_date=end_date,
+            limit=None,
+            merge=settings.policy_sync_merge_documents,
+        )
+
+        total_processed = 0
+        total_imported = 0
+        total_skipped_import = 0
+        total_skipped_transform = 0
+        transform_errors: list[dict] = []
+        transform_pins: list = []
+        import_errors: list[dict] = []
+        import_items: list[PolicyBatchItemResult] = []
+        import_pin_ids: list[int] = []
+        batches_run = 0
+        last_transform_pending = 0
+        last_import = PolicyImportBatchResult(
+            inserted_count=0,
+            skipped_duplicate_count=0,
+            pending_import_count=0,
+            error_count=0,
+        )
+
+        def _accumulate_import(import_batch: PolicyImportBatchResult) -> None:
+            nonlocal total_imported, total_skipped_import, last_import
+            total_imported += import_batch.inserted_count
+            total_skipped_import += import_batch.skipped_duplicate_count
+            import_errors.extend(import_batch.errors)
+            import_items.extend(import_batch.items)
+            import_pin_ids.extend(import_batch.pin_ids)
+            last_import = import_batch
+
+        db_ids = await ingest_service.get_imported_policy_api_ids()
+
+        def _track_imported_ids(import_batch: PolicyImportBatchResult) -> None:
+            for item in import_batch.items:
+                if item.action == PolicyBatchAction.CREATED and item.policy_api_id is not None:
+                    db_ids.add(item.policy_api_id)
+
+        while True:
+            if transform_limit is not None and total_processed >= transform_limit:
+                break
+
+            batch_limit = effective_batch
+            if transform_limit is not None:
+                batch_limit = min(effective_batch, transform_limit - total_processed)
+
+            transform_batch = await self.transform_and_save(
+                limit=batch_limit,
+                s3_util=s3_util,
+                db_policy_api_ids=db_ids,
+            )
+            batches_run += 1
+            total_processed += transform_batch.processed_count
+            total_skipped_transform += transform_batch.skipped_duplicate_count
+            transform_errors.extend(transform_batch.errors)
+            transform_pins.extend(transform_batch.pins)
+            last_transform_pending = transform_batch.remaining_pending_count
+
+            if transform_batch.processed_count > 0:
+                import_batch = await ingest_service.import_handoff_batch(
+                    import_all=False,
+                    limit=effective_batch,
+                )
+                _accumulate_import(import_batch)
+                _track_imported_ids(import_batch)
+
+            if transform_batch.remaining_pending_count == 0:
+                break
+            if transform_batch.processed_count == 0:
+                break
+
+        while load_jsonl_rows(self.handoff_path()):
+            prev_pending = last_import.pending_import_count
+            import_batch = await ingest_service.import_handoff_batch(
+                import_all=False,
+                limit=effective_batch,
+            )
+            _accumulate_import(import_batch)
+            _track_imported_ids(import_batch)
+            if import_batch.pending_import_count == 0:
+                break
+            if import_batch.inserted_count == 0 and import_batch.pending_import_count >= prev_pending:
+                break
+
+        aggregated_transform = PolicyTransformBatchResult(
+            input_path=str(self.documents_path()),
+            output_path=str(self.handoff_path()),
+            processed_count=total_processed,
+            error_count=len(transform_errors),
+            errors=transform_errors,
+            pins=transform_pins,
+            hint=(
+                f"배치 {batches_run}회, 가공 {total_processed}건 "
+                f"(배치 크기 {effective_batch}, 동시성 {settings.policy_transform_concurrency})"
+            ),
+            skipped_duplicate_count=total_skipped_transform,
+            pending_count=last_transform_pending,
+            requested_batch_size=effective_batch,
+            batches_run=batches_run,
+        )
+        import_result = PolicyImportBatchResult(
+            inserted_count=total_imported,
+            skipped_duplicate_count=total_skipped_import,
+            pending_import_count=last_import.pending_import_count,
+            error_count=len(import_errors),
+            errors=import_errors,
+            items=import_items,
+            pin_ids=import_pin_ids,
+            requested_batch_size=effective_batch,
+        )
+
+        PolicyEventIngestService.write_sync_meta(
+            query_start_date=start_date,
+            query_end_date=end_date,
+        )
+        hint = (
+            f"sync 완료: 수집 {search.count}건, 가공 {total_processed}건({batches_run}배치), "
+            f"DB INSERT {total_imported}건."
+        )
+        if (
+            total_processed == 0
+            and total_imported == 0
+            and last_transform_pending == 0
+            and len(db_ids) > 0
+        ):
+            hint = (
+                f"이미 DB 반영 완료 (policy 핀 {len(db_ids)}건). "
+                "handoff JSONL이 비어 있는 것은 import 후 캐시 정리로 정상입니다."
+            )
+        elif total_processed == 0 and last_transform_pending > 0:
+            hint = (
+                f"수집 {search.count}건 저장됐으나 가공 0건입니다. "
+                f"미가공 {last_transform_pending}건 — GEMINI_API_KEY·transform 오류를 확인하세요."
+            )
+        elif total_processed > 0 and last_transform_pending > 0:
+            hint = (
+                f"sync 부분 완료: 가공 {total_processed}건, DB INSERT {total_imported}건. "
+                f"미가공 {last_transform_pending}건 남음 — /policy-admin/sync 재실행 또는 transform-batch."
+            )
+        return PolicySyncResult(
+            query_start_date=start_date,
+            query_end_date=end_date,
+            search=search,
+            transform=aggregated_transform,
+            import_result=import_result,
+            hint=hint,
+        )
 
     def load_from_jsonl(
         self,
diff --git a/app/services/internal/PolicyPinSchedulerService.py b/app/services/internal/PolicyPinSchedulerService.py
new file mode 100644
index 0000000..9309756
--- /dev/null
+++ b/app/services/internal/PolicyPinSchedulerService.py
@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+
+from app.core.config import settings
+from app.core.database import AsyncSessionLocal
+from app.repositories.CardnewsImageS3Repo import CardnewsImageS3Repo
+from app.repositories.CommunityRepo import CommunityRepo
+from app.repositories.EventPinRepo import EventPinRepo
+from app.repositories.PinRepo import PinRepo
+from app.repositories.UserRepo import UserRepo
+from app.schemas.PolicyAdminDTO import PolicySyncResult
+from app.services.PolicyEventIngestService import PolicyEventIngestService, POLICY_SYNC_META_PATH
+from app.services.PolicyPinService import PolicyPinService
+from app.utils.S3Util import S3Util
+
+logger = logging.getLogger(__name__)
+_KST = ZoneInfo("Asia/Seoul")
+
+
+class PolicyPinSchedulerService:
+    def __init__(self, *, s3_util: S3Util) -> None:
+        self._s3_util = s3_util
+        self._task: asyncio.Task[None] | None = None
+        self._pin_service = PolicyPinService()
+
+    def start(self) -> None:
+        if self._task is not None and not self._task.done():
+            return
+        self._task = asyncio.create_task(self._run_loop(), name="policy_pin_scheduler")
+        logger.info("정책 핀 sync 스케줄러 시작")
+
+    async def stop(self) -> None:
+        if self._task is None:
+            return
+        self._task.cancel()
+        try:
+            await self._task
+        except asyncio.CancelledError:
+            pass
+        finally:
+            self._task = None
+        logger.info("정책 핀 sync 스케줄러 종료")
+
+    async def run_once_now(self, *, force: bool = False) -> PolicySyncResult:
+        if not force and not self._should_run_sync():
+            start, end = self._pin_service.default_date_range()
+            from app.schemas.PolicyAdminDTO import PolicyImportBatchResult
+            from app.schemas.PolicyPinDTO import PolicyPinSearchResult, PolicyPinTransformResult
+
+            return PolicySyncResult(
+                query_start_date=start,
+                query_end_date=end,
+                search=PolicyPinSearchResult(
+                    query_start_date=start,
+                    query_end_date=end,
+                    count=0,
+                    pins=[],
+                    saved_documents_path=str(self._pin_service.documents_path()),
+                    stats={},
+                    hint="POLICY_SYNC_INTERVAL_DAYS 미경과로 sync 생략",
+                ),
+                transform=PolicyPinTransformResult(
+                    input_path=str(self._pin_service.documents_path()),
+                    output_path=str(self._pin_service.handoff_path()),
+                    processed_count=0,
+                    error_count=0,
+                    pins=[],
+                    hint="sync 생략",
+                ),
+                import_result=PolicyImportBatchResult(
+                    inserted_count=0,
+                    skipped_duplicate_count=0,
+                    pending_import_count=0,
+                    error_count=0,
+                ),
+                hint="POLICY_SYNC_INTERVAL_DAYS 미경과로 sync 생략",
+            )
+
+        async with AsyncSessionLocal() as session:
+            ingest = PolicyEventIngestService(
+                pin_repo=PinRepo(session),
+                event_pin_repo=EventPinRepo(session),
+                community_repo=CommunityRepo(session),
+                cardnews_image_s3_repo=CardnewsImageS3Repo(session),
+                user_repo=UserRepo(session),
+            )
+            return await self._pin_service.sync_pipeline(
+                ingest_service=ingest,
+                s3_util=self._s3_util,
+            )
+
+    def _should_run_sync(self) -> bool:
+        if not POLICY_SYNC_META_PATH.is_file():
+            return True
+        try:
+            meta = json.loads(POLICY_SYNC_META_PATH.read_text(encoding="utf-8"))
+            last_sync_at = str(meta.get("last_sync_at") or "").strip()
+            if not last_sync_at:
+                return True
+            last_run = datetime.fromisoformat(last_sync_at)
+            if last_run.tzinfo is None:
+                last_run = last_run.replace(tzinfo=_KST)
+            elapsed = datetime.now(_KST) - last_run.astimezone(_KST)
+            # 스케줄 시각 오차로 하루 밀리는 것 방지
+            return elapsed >= timedelta(days=settings.policy_sync_interval_days) - timedelta(hours=1)
+        except (json.JSONDecodeError, ValueError):
+            return True
+
+    async def _run_loop(self) -> None:
+        while True:
+            wait_seconds = self._seconds_until_next_schedule_kst()
+            logger.info("정책 핀 sync 스케줄러 대기 %.1fs", wait_seconds)
+            await asyncio.sleep(wait_seconds)
+            try:
+                result = await self.run_once_now(force=False)
+                logger.info(
+                    "정책 핀 sync 실행: 가공 %d건, INSERT %d건",
+                    result.transform.processed_count,
+                    result.import_result.inserted_count,
+                )
+            except Exception:
+                logger.exception("정책 핀 sync 스케줄 실행 실패")
+
+    @staticmethod
+    def _seconds_until_next_schedule_kst() -> float:
+        now = datetime.now(_KST)
+        hour = settings.policy_sync_schedule_hour_kst
+        next_run = datetime.combine(now.date(), datetime.min.time(), tzinfo=_KST) + timedelta(hours=hour)
+        if now >= next_run:
+            next_run += timedelta(days=1)
+        delta = next_run - now
+        return max(delta.total_seconds(), 1.0)
diff --git a/app/services/policy_cardnews.py b/app/services/policy_cardnews.py
index 4c94883..be070d7 100644
--- a/app/services/policy_cardnews.py
+++ b/app/services/policy_cardnews.py
@@ -2,7 +2,9 @@
 
 import logging
 from pathlib import Path
-from typing import Any
+from typing import Any, TypedDict
+
+from anyio import to_thread
 
 from app.core.config import settings
 from app.policy_cardnews import (
@@ -16,24 +18,97 @@
     build_policy_cardnews_image_prompt,
     build_policy_cardnews_slide_prompt,
 )
+from app.utils.S3Util import S3Util
 
 logger = logging.getLogger(__name__)
 
 POLICY_CARDNEWS_OUTPUT_DIR = (
     Path(__file__).resolve().parents[2] / "rag" / "output" / "policy_cardnews"
 )
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+class CardnewsS3Image(TypedDict):
+    key: str
+    url: str
+
+
+def cleanup_local_cardnews_dir(content_id: str) -> None:
+    normalized = str(content_id or "").strip()
+    if not normalized:
+        return
+    target = POLICY_CARDNEWS_OUTPUT_DIR / normalized
+    if not target.is_dir():
+        return
+    import shutil
+
+    shutil.rmtree(target, ignore_errors=True)
+    logger.debug("로컬 카드뉴스 디렉터리 삭제: %s", target)
+
+
+def _maybe_cleanup_local_cardnews(content_id: str) -> None:
+    if not settings.policy_cardnews_keep_local_files:
+        cleanup_local_cardnews_dir(content_id)
+
+
+def _slide_object_key(content_id: str, slide_no: int) -> str:
+    prefix = settings.policy_cardnews_s3_prefix.strip("/")
+    return f"{prefix}/{content_id}/slide_{slide_no:02d}.png"
+
+
+async def _upload_slide_bytes(
+    s3_util: S3Util,
+    *,
+    content_id: str,
+    slide_no: int,
+    image_bytes: bytes,
+) -> CardnewsS3Image:
+    key = _slide_object_key(content_id, slide_no)
+    result = await s3_util.upload_bytes(
+        image_bytes,
+        filename=f"slide_{slide_no:02d}.png",
+        content_type="image/png",
+        object_key=key,
+    )
+    return {"key": result["key"], "url": result["url"]}
+
+
+async def _upload_local_handoff_path(
+    s3_util: S3Util,
+    *,
+    content_id: str,
+    handoff_path: str,
+) -> CardnewsS3Image:
+    local_path = Path(handoff_path)
+    if not local_path.is_absolute():
+        local_path = (_REPO_ROOT / handoff_path.strip().lstrip("/")).resolve()
+    if not local_path.is_file():
+        raise FileNotFoundError(f"카드뉴스 로컬 파일 없음: {local_path}")
+    slide_name = local_path.stem
+    slide_no = 1
+    if slide_name.startswith("slide_"):
+        try:
+            slide_no = int(slide_name.split("_", 1)[1])
+        except ValueError:
+            slide_no = 1
+    image_bytes = await to_thread.run_sync(local_path.read_bytes)
+    return await _upload_slide_bytes(
+        s3_util,
+        content_id=content_id,
+        slide_no=slide_no,
+        image_bytes=image_bytes,
+    )
 
 
-async def _generate_with_image_model(
+async def _generate_slide_bytes_with_image_model(
     *,
     image_service: PolicyCardnewsImageService,
     content_id: str,
     pin_title: str,
     minister: str,
     slides: list[dict[str, Any]],
-    output_dir: Path,
-) -> list[str]:
-    saved_paths: list[str] = []
+) -> list[tuple[int, bytes]]:
+    out: list[tuple[int, bytes]] = []
     total = len(slides)
     for index, slide in enumerate(slides, start=1):
         prompt = build_policy_cardnews_image_prompt(
@@ -45,18 +120,11 @@ async def _generate_with_image_model(
         )
         image_bytes = await image_service.generate_slide_image_bytes(prompt=prompt)
         slide_no = int(slide.get("slide") or index)
-        saved_paths.append(
-            save_slide_image_bytes(
-                contentid=content_id,
-                slide_no=slide_no,
-                image_bytes=image_bytes,
-                output_dir=output_dir,
-            )
-        )
-    return saved_paths
+        out.append((slide_no, image_bytes))
+    return out
 
 
-async def _generate_with_pillow(
+async def _generate_local_paths_with_pillow(
     *,
     content_id: str,
     minister: str,
@@ -77,19 +145,19 @@ async def _generate_with_pillow(
     )
 
 
-async def generate_cardnews_image_paths(
+async def generate_cardnews_s3_images(
     text_llm: IssuePinLLMService,
     *,
     row: dict,
     easy_read_content: str,
+    s3_util: S3Util,
     output_dir: Path | None = None,
     image_service: PolicyCardnewsImageService | None = None,
-) -> list[str]:
+) -> list[CardnewsS3Image]:
     content_id = str(row.get("contentid") or "").strip()
     if not content_id:
         return []
 
-    existing = [str(url).strip() for url in (row.get("cardnews_image_urls") or []) if str(url).strip()]
     prompt = build_policy_cardnews_slide_prompt(
         pin_title=str(row.get("pin_title") or ""),
         pin_content=easy_read_content or str(row.get("pin_content") or ""),
@@ -116,17 +184,34 @@ async def generate_cardnews_image_paths(
         first["use_image"] = True
         slides[0] = first
 
+    uploaded: list[CardnewsS3Image] = []
+
     if settings.policy_cardnews_use_image_model:
         img_service = image_service or PolicyCardnewsImageService.from_settings()
         try:
-            generated = await _generate_with_image_model(
+            slide_bytes_list = await _generate_slide_bytes_with_image_model(
                 image_service=img_service,
                 content_id=content_id,
                 pin_title=pin_title,
                 minister=minister,
                 slides=slides,
-                output_dir=target_dir,
             )
+            for slide_no, image_bytes in slide_bytes_list:
+                if settings.policy_cardnews_keep_local_files:
+                    save_slide_image_bytes(
+                        contentid=content_id,
+                        slide_no=slide_no,
+                        image_bytes=image_bytes,
+                        output_dir=target_dir,
+                    )
+                uploaded.append(
+                    await _upload_slide_bytes(
+                        s3_util,
+                        content_id=content_id,
+                        slide_no=slide_no,
+                        image_bytes=image_bytes,
+                    ),
+                )
         except Exception:
             if not settings.policy_cardnews_pillow_fallback:
                 raise
@@ -134,7 +219,7 @@ async def generate_cardnews_image_paths(
                 "카드뉴스 이미지 모델 생성 실패로 Pillow 폴백 contentid=%s",
                 content_id,
             )
-            generated = await _generate_with_pillow(
+            local_paths = await _generate_local_paths_with_pillow(
                 content_id=content_id,
                 minister=minister,
                 slides=slides,
@@ -143,8 +228,16 @@ async def generate_cardnews_image_paths(
                 source_url=source_url,
                 pin_content=easy_read_content or str(row.get("pin_content") or ""),
             )
+            for path in local_paths:
+                uploaded.append(
+                    await _upload_local_handoff_path(
+                        s3_util,
+                        content_id=content_id,
+                        handoff_path=path,
+                    ),
+                )
     else:
-        generated = await _generate_with_pillow(
+        local_paths = await _generate_local_paths_with_pillow(
             content_id=content_id,
             minister=minister,
             slides=slides,
@@ -153,5 +246,36 @@ async def generate_cardnews_image_paths(
             source_url=source_url,
             pin_content=easy_read_content or str(row.get("pin_content") or ""),
         )
+        for path in local_paths:
+            uploaded.append(
+                await _upload_local_handoff_path(
+                    s3_util,
+                    content_id=content_id,
+                    handoff_path=path,
+                ),
+            )
+
+    _maybe_cleanup_local_cardnews(content_id)
+    return uploaded
+
 
-    return existing + generated
+async def generate_cardnews_image_paths(
+    text_llm: IssuePinLLMService,
+    *,
+    row: dict,
+    easy_read_content: str,
+    output_dir: Path | None = None,
+    image_service: PolicyCardnewsImageService | None = None,
+    s3_util: S3Util | None = None,
+) -> list[str]:
+    """하위 호환: S3 URL 문자열 목록만 반환."""
+    s3 = s3_util or S3Util()
+    images = await generate_cardnews_s3_images(
+        text_llm,
+        row=row,
+        easy_read_content=easy_read_content,
+        s3_util=s3,
+        output_dir=output_dir,
+        image_service=image_service,
+    )
+    return [img["url"] for img in images if img.get("url")]
diff --git a/app/services/policy_pin_transform.py b/app/services/policy_pin_transform.py
index b75952e..ef2f4d3 100644
--- a/app/services/policy_pin_transform.py
+++ b/app/services/policy_pin_transform.py
@@ -1,13 +1,16 @@
 from __future__ import annotations
 
+import asyncio
 from pathlib import Path
+from typing import Any
 
 from app.core.config import settings
 from app.schemas.PolicyPinDTO import PolicyPinHandoffDTO, PolicyPinTransformResult
 from app.services.internal.ai.IssuePinLLMService import IssuePinLLMService
 from app.services.internal.ai.gemini_retry import parse_gemini_model_list
-from app.services.policy_cardnews import generate_cardnews_image_paths
+from app.services.policy_cardnews import CardnewsS3Image, generate_cardnews_s3_images
 from app.services.prompts.policy_pin import build_policy_easy_read_prompt
+from app.utils.S3Util import S3Util
 from rag.scripts.chunk_module import iter_jsonl, write_jsonl
 
 POLICY_DOCUMENTS_PATH = (
@@ -16,48 +19,91 @@
 POLICY_HANDOFF_PATH = (
     Path(__file__).resolve().parents[2] / "rag" / "output" / "policy_pins_for_db.jsonl"
 )
+POLICY_SYNC_META_PATH = (
+    Path(__file__).resolve().parents[2] / "rag" / "output" / "policy_sync_meta.json"
+)
+
+
+def parse_policy_api_id(row: dict[str, Any]) -> int | None:
+    raw = row.get("policy_api_id") or row.get("contentid")
+    if raw is None:
+        return None
+    text = str(raw).strip()
+    if not text.isdigit():
+        return None
+    return int(text)
+
+
+def row_content_id(row: dict[str, Any]) -> str:
+    return str(row.get("contentid") or row.get("policy_api_id") or "").strip()
+
+
+def row_raw_content(row: dict[str, Any]) -> str:
+    return (row.get("pin_content_raw") or row.get("pin_content") or row.get("text") or "").strip()
+
+
+def normalize_raw_for_compare(text: str) -> str:
+    return " ".join((text or "").split()).strip()
 
 
-def append_source_link_to_pin_content(body: str, source_url: str) -> str:
-    """가공 본문 끝에 원문 기사 URL을 붙인다 (중복 방지)."""
-    text = (body or "").strip()
-    url = (source_url or "").strip()
-    if not url or not text:
-        return text or url
+def load_jsonl_rows(path: Path) -> list[dict[str, Any]]:
+    if not path.is_file():
+        return []
+    return [row for row in iter_jsonl(path) if isinstance(row, dict)]
 
-    normalized = url.rstrip("/")
-    for line in text.splitlines():
-        line = line.strip()
-        if line == url or line.rstrip("/") == normalized:
-            return text
-        if line.startswith("http") and normalized in line:
-            return text
 
-    return f"{text}\n\n원문 기사: {url}"
+def load_rows_by_content_id(path: Path) -> dict[str, dict[str, Any]]:
+    out: dict[str, dict[str, Any]] = {}
+    for row in load_jsonl_rows(path):
+        content_id = row_content_id(row)
+        if content_id:
+            out[content_id] = row
+    return out
+
+
+def merge_documents(
+    existing: dict[str, dict[str, Any]],
+    new_rows: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    merged = dict(existing)
+    for row in new_rows:
+        content_id = row_content_id(row)
+        if content_id:
+            merged[content_id] = row
+    return list(merged.values())
+
+
+def write_handoff_map(handoff_by_id: dict[str, dict[str, Any]], path: Path | None = None) -> None:
+    dst = path or POLICY_HANDOFF_PATH
+    write_jsonl(dst, list(handoff_by_id.values()))
 
 
 def build_handoff_row(
-    source: dict,
+    source: dict[str, Any],
     *,
     easy_read_content: str,
-    cardnews_image_urls: list[str] | None = None,
-) -> dict:
-    # DB 전달 JSONL 1행: title, pin_content, cardnews_image_urls, source_url 만
-    cardnews_urls = [
-        str(url).strip()
-        for url in (
-            cardnews_image_urls
-            if cardnews_image_urls is not None
-            else (source.get("cardnews_image_urls") or [])
-        )
-        if str(url).strip()
-    ]
+    cardnews_images: list[CardnewsS3Image] | None = None,
+) -> dict[str, Any]:
+    content_id = row_content_id(source)
+    policy_api_id = parse_policy_api_id(source)
+    images = cardnews_images or []
+    cardnews_urls = [img["url"] for img in images if str(img.get("url") or "").strip()]
     source_url = (source.get("source_url") or "").strip()
+    event_start = source.get("event_start_time")
+    event_end = source.get("event_end_time")
+    title = (source.get("pin_title") or source.get("title") or "").strip()
     return {
-        "title": (source.get("pin_title") or source.get("title") or "").strip(),
-        "pin_content": append_source_link_to_pin_content(easy_read_content, source_url),
+        "contentid": content_id,
+        "policy_api_id": policy_api_id,
+        "title": title,
+        "pin_content": (easy_read_content or "").strip(),
+        "pin_content_raw": row_raw_content(source),
         "cardnews_image_urls": cardnews_urls,
+        "cardnews_images": images,
         "source_url": source_url,
+        "event_start_time": event_start,
+        "event_end_time": event_end,
+        "approve_date": (source.get("approve_date") or "").strip(),
     }
 
 
@@ -74,8 +120,13 @@ def build_llm_service(*, model: str | None = None) -> IssuePinLLMService:
     )
 
 
-async def transform_one_row(llm: IssuePinLLMService, row: dict) -> dict:
-    raw_content = (row.get("pin_content") or row.get("text") or "").strip()
+async def transform_one_row(
+    llm: IssuePinLLMService,
+    row: dict[str, Any],
+    *,
+    s3_util: S3Util,
+) -> dict[str, Any]:
+    raw_content = row_raw_content(row)
     if not raw_content:
         raise ValueError("pin_content가 비어 있음")
 
@@ -87,17 +138,53 @@ async def transform_one_row(llm: IssuePinLLMService, row: dict) -> dict:
         approve_date=str(row.get("approve_date") or ""),
     )
     easy_read_text = await llm.generate_pin_text(prompt=prompt)
-    cardnews_urls = await generate_cardnews_image_paths(
-        llm,
+    cardnews_images = await generate_cardnews_s3_images(
+        text_llm=llm,
         row=row,
         easy_read_content=easy_read_text,
+        s3_util=s3_util,
     )
-    raw_html = (row.get("pin_content_raw") or raw_content).strip()
-    source = {**row, "pin_content_raw": raw_html}
+    source = {**row, "pin_content_raw": raw_content}
     return build_handoff_row(
         source,
         easy_read_content=easy_read_text,
-        cardnews_image_urls=cardnews_urls,
+        cardnews_images=cardnews_images,
+    )
+
+
+def list_pending_transform_rows(
+    documents: list[dict[str, Any]],
+    handoff_by_id: dict[str, dict[str, Any]],
+    *,
+    db_policy_api_ids: set[int] | None = None,
+) -> list[dict[str, Any]]:
+    db_ids = db_policy_api_ids or set()
+    pending: list[dict[str, Any]] = []
+    for row in documents:
+        content_id = row_content_id(row)
+        if not content_id:
+            continue
+        policy_id = parse_policy_api_id(row)
+        if policy_id is not None and policy_id in db_ids:
+            continue
+        if content_id in handoff_by_id:
+            continue
+        pending.append(row)
+    return pending
+
+
+def count_pending_transform(
+    documents: list[dict[str, Any]],
+    handoff_by_id: dict[str, dict[str, Any]],
+    *,
+    db_policy_api_ids: set[int] | None = None,
+) -> int:
+    return len(
+        list_pending_transform_rows(
+            documents,
+            handoff_by_id,
+            db_policy_api_ids=db_policy_api_ids,
+        ),
     )
 
 
@@ -107,6 +194,9 @@ async def transform_documents_jsonl(
     output_path: Path | None = None,
     limit: int | None = None,
     model: str | None = None,
+    s3_util: S3Util | None = None,
+    db_policy_api_ids: set[int] | None = None,
+    merge_handoff: bool = True,
 ) -> PolicyPinTransformResult:
     src = input_path or POLICY_DOCUMENTS_PATH
     dst = output_path or POLICY_HANDOFF_PATH
@@ -115,36 +205,68 @@ async def transform_documents_jsonl(
             f"원문 JSONL 없음: {src}. 먼저 GET /policy-pins/search 또는 fetch 스크립트를 실행하세요.",
         )
 
+    documents = load_jsonl_rows(src)
+    handoff_by_id = load_rows_by_content_id(dst) if merge_handoff else {}
+    db_ids = db_policy_api_ids or set()
+    pending = list_pending_transform_rows(
+        documents,
+        handoff_by_id,
+        db_policy_api_ids=db_ids,
+    )
+
     llm = build_llm_service(model=model)
-    results: list[dict] = []
-    errors: list[dict] = []
+    s3 = s3_util or S3Util()
+    processed_rows: list[dict[str, Any]] = []
+    errors: list[dict[str, Any]] = []
+    skipped_duplicate_count = 0
 
-    for row in iter_jsonl(src):
-        if not isinstance(row, dict):
-            continue
-        if limit is not None and len(results) >= limit:
-            break
-
-        content_id = str(row.get("contentid") or "").strip()
-        try:
-            results.append(await transform_one_row(llm, row))
-        except Exception as exc:
-            errors.append(
-                {
+    rows_to_process = pending if limit is None else pending[:limit]
+    concurrency = min(settings.policy_transform_concurrency, len(rows_to_process) or 1)
+    semaphore = asyncio.Semaphore(max(1, concurrency))
+
+    async def _transform_row(row: dict[str, Any]) -> tuple[str, dict[str, Any] | None, dict[str, Any] | None]:
+        content_id = row_content_id(row)
+        async with semaphore:
+            try:
+                handoff_row = await transform_one_row(llm, row, s3_util=s3)
+                return content_id, handoff_row, None
+            except Exception as exc:
+                return content_id, None, {
                     "contentid": content_id,
                     "pin_title": row.get("pin_title"),
                     "error": str(exc),
                 }
-            )
 
-    write_jsonl(dst, results)
-    pins = [PolicyPinHandoffDTO.from_row(item) for item in results]
+    for content_id, handoff_row, err_row in await asyncio.gather(
+        *(_transform_row(row) for row in rows_to_process),
+    ):
+        if handoff_row is not None and content_id:
+            handoff_by_id[content_id] = handoff_row
+            processed_rows.append(handoff_row)
+        elif err_row is not None:
+            errors.append(err_row)
+
+    for row in documents:
+        content_id = row_content_id(row)
+        if not content_id or content_id in handoff_by_id:
+            continue
+        policy_id = parse_policy_api_id(row)
+        if policy_id is not None and policy_id in db_ids:
+            skipped_duplicate_count += 1
+
+    write_handoff_map(handoff_by_id, dst)
+    remaining_pending = count_pending_transform(
+        documents,
+        handoff_by_id,
+        db_policy_api_ids=db_ids,
+    )
+    pins = [PolicyPinHandoffDTO.from_row(item) for item in processed_rows]
 
     hint: str | None
-    if results:
+    if processed_rows or handoff_by_id:
         hint = (
-            f"response.pins({len(pins)}건)와 {dst.name} 내용이 동일합니다. "
-            "기간 필터는 이미 search(policy_documents.jsonl) 단계에서 적용된 범위입니다."
+            f"이번 가공 {len(processed_rows)}건, handoff 총 {len(handoff_by_id)}건. "
+            "기간 필터는 search(policy_documents.jsonl) 단계에서 적용된 범위입니다."
         )
     else:
         hint = "가공 성공 건이 없습니다. GEMINI_API_KEY·원문 pin_content를 확인하세요."
@@ -152,9 +274,12 @@ async def transform_documents_jsonl(
     return PolicyPinTransformResult(
         input_path=str(src),
         output_path=str(dst),
-        processed_count=len(results),
+        processed_count=len(processed_rows),
         error_count=len(errors),
         errors=errors,
         pins=pins,
         hint=hint,
+        skipped_duplicate_count=skipped_duplicate_count,
+        pending_count=len(pending),
+        remaining_pending_count=remaining_pending,
     )
diff --git a/app/services/policy_pipeline_cleanup.py b/app/services/policy_pipeline_cleanup.py
new file mode 100644
index 0000000..38ffb16
--- /dev/null
+++ b/app/services/policy_pipeline_cleanup.py
@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import logging
+import shutil
+from pathlib import Path
+
+from app.services.policy_cardnews import POLICY_CARDNEWS_OUTPUT_DIR
+from app.services.policy_pin_transform import (
+    POLICY_DOCUMENTS_PATH,
+    POLICY_HANDOFF_PATH,
+    load_jsonl_rows,
+    parse_policy_api_id,
+    row_content_id,
+    write_handoff_map,
+)
+from rag.scripts.chunk_module import write_jsonl
+
+logger = logging.getLogger(__name__)
+
+
+def prune_jsonl_by_policy_api_ids(
+    path: Path,
+    policy_api_ids: set[int],
+) -> int:
+    if not policy_api_ids or not path.is_file():
+        return 0
+
+    kept: list[dict] = []
+    removed = 0
+    for row in load_jsonl_rows(path):
+        policy_id = parse_policy_api_id(row)
+        if policy_id is not None and policy_id in policy_api_ids:
+            removed += 1
+            continue
+        kept.append(row)
+
+    if removed > 0:
+        write_jsonl(path, kept)
+        logger.info("JSONL 정리 %s: %d건 제거, %d건 유지", path.name, removed, len(kept))
+    return removed
+
+
+def prune_pipeline_imported(
+    policy_api_ids: set[int],
+    *,
+    documents_path: Path | None = None,
+    handoff_path: Path | None = None,
+) -> dict[str, int]:
+    if not policy_api_ids:
+        return {"documents_removed": 0, "handoff_removed": 0}
+
+    docs = documents_path or POLICY_DOCUMENTS_PATH
+    handoff = handoff_path or POLICY_HANDOFF_PATH
+    return {
+        "documents_removed": prune_jsonl_by_policy_api_ids(docs, policy_api_ids),
+        "handoff_removed": prune_jsonl_by_policy_api_ids(handoff, policy_api_ids),
+    }
+
+
+def prune_handoff_pending_only(
+    *,
+    db_policy_api_ids: set[int],
+    handoff_path: Path | None = None,
+) -> int:
+    """DB에 이미 있는 항목을 handoff에서 제거 (가공 완료·적재 완료 캐시 정리)."""
+    path = handoff_path or POLICY_HANDOFF_PATH
+    if not path.is_file() or not db_policy_api_ids:
+        return 0
+    return prune_jsonl_by_policy_api_ids(path, db_policy_api_ids)
+
+
+def cleanup_local_cardnews_dirs(content_ids: set[str]) -> int:
+    removed = 0
+    for raw_id in content_ids:
+        content_id = str(raw_id or "").strip()
+        if not content_id:
+            continue
+        target = POLICY_CARDNEWS_OUTPUT_DIR / content_id
+        if not target.is_dir():
+            continue
+        shutil.rmtree(target, ignore_errors=True)
+        removed += 1
+        logger.debug("로컬 카드뉴스 삭제: %s", target)
+    return removed
+
+
+def policy_api_ids_to_content_ids(policy_api_ids: set[int]) -> set[str]:
+    return {str(policy_id) for policy_id in policy_api_ids}
+
+
+def cleanup_after_policy_import(policy_api_ids: set[int]) -> dict[str, int]:
+    stats = prune_pipeline_imported(policy_api_ids)
+    stats["cardnews_dirs_removed"] = cleanup_local_cardnews_dirs(
+        policy_api_ids_to_content_ids(policy_api_ids),
+    )
+    return stats
+
+
+def reset_handoff_map(handoff_path: Path | None = None) -> None:
+    write_handoff_map({}, handoff_path)
+
+
+def content_ids_from_rows(rows: list[dict]) -> set[str]:
+    out: set[str] = set()
+    for row in rows:
+        content_id = row_content_id(row)
+        if content_id:
+            out.add(content_id)
+    return out
diff --git a/app/utils/policy_news_parse.py b/app/utils/policy_news_parse.py
index faad181..7ebca9f 100644
--- a/app/utils/policy_news_parse.py
+++ b/app/utils/policy_news_parse.py
@@ -4,6 +4,7 @@
 import logging
 import re
 from datetime import date, datetime, timedelta
+from zoneinfo import ZoneInfo
 from html import unescape
 from typing import Any
 from urllib.parse import urljoin, urlparse
@@ -13,6 +14,7 @@
 
 logger = logging.getLogger(__name__)
 
+_KST = ZoneInfo("Asia/Seoul")
 _POLICY_DATE_FMT = "%m/%d/%Y %H:%M:%S"
 _POLICY_DATE_FMT_SHORT = "%m/%d/%Y"
 _IMG_SRC_RE = re.compile(
@@ -329,6 +331,22 @@ def approve_date_to_yyyymmdd(raw: str | None) -> str | None:
     return parsed.strftime("%Y%m%d")
 
 
+def policy_date_kst(raw: str | None) -> date | None:
+    parsed = parse_policy_datetime(raw)
+    if parsed is None:
+        return None
+    return parsed.date()
+
+
+def is_embargo_active(embargo_raw: str | None, *, as_of: date | None = None) -> bool:
+    """EmbargoDate가 as_of(KST, 기본 오늘)보다 미래이면 아직 엠바고 → True."""
+    embargo_date = policy_date_kst(embargo_raw)
+    if embargo_date is None:
+        return False
+    today = as_of if as_of is not None else datetime.now(_KST).date()
+    return embargo_date > today
+
+
 def local_name(tag: str) -> str:
     if "}" in tag:
         return tag.rsplit("}", 1)[-1]
@@ -434,6 +452,7 @@ def build_policy_document_row(item: dict[str, str]) -> dict[str, Any] | None:
         "source_url": source_url,
         "subtitles": merge_subtitles(item),
         "contents_status": (item.get("ContentsStatus") or "").strip(),
+        "embargo_date": (item.get("EmbargoDate") or "").strip(),
     }
 
 
diff --git a/rag/scripts/fetch_policy_news.py b/rag/scripts/fetch_policy_news.py
index 2e8be1d..19e9e50 100644
--- a/rag/scripts/fetch_policy_news.py
+++ b/rag/scripts/fetch_policy_news.py
@@ -21,6 +21,7 @@
 from app.clients.PolicyNewsClient import PolicyNewsClient
 from app.utils.policy_news_parse import (
     build_policy_document_row,
+    is_embargo_active,
     iter_date_chunks,
     validate_yyyymmdd,
 )
@@ -45,6 +46,7 @@ async def fetch_policy_documents(
         "documents": 0,
         "skipped_duplicate": 0,
         "skipped_invalid": 0,
+        "skipped_embargo": 0,
         "api_errors": 0,
     }
     seen_ids: set[str] = set()
@@ -83,6 +85,10 @@ async def fetch_policy_documents(
                 continue
             seen_ids.add(news_id)
 
+            if is_embargo_active(item.get("EmbargoDate")):
+                stats["skipped_embargo"] += 1
+                continue
+
             row = build_policy_document_row(item)
             if row is None:
                 stats["skipped_invalid"] += 1
diff --git a/tests/test_policy_admin_batch.py b/tests/test_policy_admin_batch.py
new file mode 100644
index 0000000..40e8014
--- /dev/null
+++ b/tests/test_policy_admin_batch.py
@@ -0,0 +1,669 @@
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from datetime import datetime, timedelta
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+from zoneinfo import ZoneInfo
+
+import app.models  # noqa: F401 — SQLAlchemy mapper registry
+from fastapi import HTTPException
+
+from app.routes.PolicyAdminRoute import sync_policy_pins
+from app.schemas.PolicyAdminDTO import (
+    PolicyBatchAction,
+    PolicyBatchItemResult,
+    PolicyImportBatchResult,
+)
+from app.schemas.PolicyPinDTO import PolicyPinSearchResult, PolicyPinTransformResult
+from app.services.PolicyEventIngestService import PolicyEventIngestService
+from app.services.PolicyPinService import PolicyPinService
+from app.services.internal.PolicyPinSchedulerService import PolicyPinSchedulerService
+from app.services.policy_cardnews import _upload_local_handoff_path, cleanup_local_cardnews_dir
+
+_KST = ZoneInfo("Asia/Seoul")
+
+
+def _make_ingest_service() -> PolicyEventIngestService:
+    session = MagicMock()
+    session.begin_nested = MagicMock(return_value=MagicMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()))
+    pin_repo = MagicMock()
+    pin_repo.session = session
+    pin_repo.commit = AsyncMock()
+    pin_repo.rollback = AsyncMock()
+    event_pin_repo = MagicMock()
+    event_pin_repo.list_policy_api_ids = AsyncMock(return_value=set())
+    event_pin_repo.get_by_policy_api_id = AsyncMock(return_value=None)
+    event_pin_repo.save = AsyncMock()
+    community_repo = MagicMock()
+    community_repo.save = AsyncMock()
+    cardnews_repo = MagicMock()
+    cardnews_repo.save = AsyncMock()
+    user_repo = MagicMock()
+    user_repo.get_by_user_name = AsyncMock(return_value=MagicMock(uid="admin-uid"))
+    return PolicyEventIngestService(
+        pin_repo=pin_repo,
+        event_pin_repo=event_pin_repo,
+        community_repo=community_repo,
+        cardnews_image_s3_repo=cardnews_repo,
+        user_repo=user_repo,
+    )
+
+
+class PolicyAdminRouteValidationTest(unittest.IsolatedAsyncioTestCase):
+    async def test_sync_invalid_date_returns_bad_request(self) -> None:
+        service = MagicMock()
+        service.sync_pipeline = AsyncMock()
+
+        with self.assertRaises(HTTPException) as ctx:
+            await sync_policy_pins(
+                service=service,
+                ingest_service=MagicMock(),
+                s3_util=MagicMock(),
+                _admin_uid="admin-uid",
+                start_date="20261301",
+                end_date="20261302",
+            )
+
+        self.assertEqual(ctx.exception.status_code, 400)
+        service.sync_pipeline.assert_not_awaited()
+
+
+class PolicyPinSchedulerShouldRunSyncTest(unittest.TestCase):
+    def test_runs_when_elapsed_within_one_hour_of_interval(self) -> None:
+        scheduler = PolicyPinSchedulerService(s3_util=MagicMock())
+        last_sync = datetime(2026, 6, 9, 1, 0, 5, tzinfo=_KST)
+        now = datetime(2026, 6, 12, 1, 0, 0, tzinfo=_KST)
+
+        with (
+            patch(
+                "app.services.internal.PolicyPinSchedulerService.POLICY_SYNC_META_PATH",
+                MagicMock(is_file=MagicMock(return_value=True)),
+            ),
+            patch(
+                "app.services.internal.PolicyPinSchedulerService.json.loads",
+                return_value={"last_sync_at": last_sync.isoformat()},
+            ),
+            patch(
+                "app.services.internal.PolicyPinSchedulerService.datetime") as mock_dt,
+            patch(
+                "app.services.internal.PolicyPinSchedulerService.settings.policy_sync_interval_days",
+                3,
+            ),
+        ):
+            mock_dt.now.return_value = now
+            mock_dt.fromisoformat = datetime.fromisoformat
+            self.assertTrue(scheduler._should_run_sync())
+
+    def test_skips_when_elapsed_more_than_one_hour_short_of_interval(self) -> None:
+        scheduler = PolicyPinSchedulerService(s3_util=MagicMock())
+        last_sync = datetime(2026, 6, 9, 1, 0, 5, tzinfo=_KST)
+        now = datetime(2026, 6, 11, 23, 0, 0, tzinfo=_KST)
+
+        with (
+            patch(
+                "app.services.internal.PolicyPinSchedulerService.POLICY_SYNC_META_PATH",
+                MagicMock(is_file=MagicMock(return_value=True)),
+            ),
+            patch(
+                "app.services.internal.PolicyPinSchedulerService.json.loads",
+                return_value={"last_sync_at": last_sync.isoformat()},
+            ),
+            patch(
+                "app.services.internal.PolicyPinSchedulerService.datetime") as mock_dt,
+            patch(
+                "app.services.internal.PolicyPinSchedulerService.settings.policy_sync_interval_days",
+                3,
+            ),
+        ):
+            mock_dt.now.return_value = now
+            mock_dt.fromisoformat = datetime.fromisoformat
+            self.assertFalse(scheduler._should_run_sync())
+
+
+class PolicyCardnewsHandoffPathTest(unittest.IsolatedAsyncioTestCase):
+    async def test_upload_local_handoff_path_accepts_absolute_path(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            slide_path = Path(tmp) / "slide_01.png"
+            slide_path.write_bytes(b"png-bytes")
+            s3_util = MagicMock()
+            s3_util.upload_bytes = AsyncMock(
+                return_value={"key": "policy/cardnews/1/slide_01.png", "url": "https://cdn.example/slide_01.png"},
+            )
+
+            result = await _upload_local_handoff_path(
+                s3_util,
+                content_id="1",
+                handoff_path=str(slide_path),
+            )
+
+            self.assertEqual(result["key"], "policy/cardnews/1/slide_01.png")
+            s3_util.upload_bytes.assert_awaited_once()
+
+    async def test_cleanup_local_cardnews_dir_ignores_empty_content_id(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp) / "policy_cardnews"
+            root.mkdir()
+
+            with patch("app.services.policy_cardnews.POLICY_CARDNEWS_OUTPUT_DIR", root):
+                cleanup_local_cardnews_dir("")
+
+            self.assertTrue(root.is_dir())
+
+
+class PolicyImportHandoffBatchTest(unittest.IsolatedAsyncioTestCase):
+    async def test_requested_batch_size_none_when_import_all(self) -> None:
+        service = _make_ingest_service()
+        handoff_rows = [
+            {
+                "contentid": "1",
+                "policy_api_id": 1,
+                "title": "정책1",
+                "pin_content": "본문1",
+                "cardnews_images": [{"key": "k1", "url": "https://cdn.example/1.png"}],
+            },
+        ]
+        with (
+            patch(
+                "app.services.PolicyEventIngestService.load_jsonl_rows",
+                return_value=handoff_rows,
+            ),
+            patch(
+                "app.services.PolicyEventIngestService.settings.policy_prune_pipeline_after_import",
+                False,
+            ),
+            patch.object(service, "_insert_policy_pin", AsyncMock(return_value=101)),
+        ):
+            result = await service.import_handoff_batch(import_all=True, limit=5)
+
+        self.assertIsNone(result.requested_batch_size)
+        self.assertEqual(result.inserted_count, 1)
+
+    async def test_requested_batch_size_limit_when_not_import_all(self) -> None:
+        service = _make_ingest_service()
+        handoff_rows = [
+            {
+                "contentid": "1",
+                "policy_api_id": 1,
+                "title": "정책1",
+                "pin_content": "본문1",
+                "cardnews_images": [{"key": "k1", "url": "https://cdn.example/1.png"}],
+            },
+        ]
+        with (
+            patch(
+                "app.services.PolicyEventIngestService.load_jsonl_rows",
+                return_value=handoff_rows,
+            ),
+            patch(
+                "app.services.PolicyEventIngestService.settings.policy_prune_pipeline_after_import",
+                False,
+            ),
+            patch.object(service, "_insert_policy_pin", AsyncMock(return_value=101)),
+        ):
+            result = await service.import_handoff_batch(import_all=False, limit=3)
+
+        self.assertEqual(result.requested_batch_size, 3)
+
+    async def test_get_imported_policy_api_ids_delegates_to_repo(self) -> None:
+        service = _make_ingest_service()
+        service._event_pin_repo.list_policy_api_ids = AsyncMock(return_value={1, 2})
+
+        result = await service.get_imported_policy_api_ids()
+
+        self.assertEqual(result, {1, 2})
+        service._event_pin_repo.list_policy_api_ids.assert_awaited_once()
+
+    async def test_import_handoff_batch_ok_when_handoff_empty_and_db_caught_up(self) -> None:
+        service = _make_ingest_service()
+        service._event_pin_repo.list_policy_api_ids = AsyncMock(return_value={1, 2})
+        documents = [
+            {"contentid": "1", "policy_api_id": 1, "pin_content": "a"},
+            {"contentid": "2", "policy_api_id": 2, "pin_content": "b"},
+        ]
+
+        with patch(
+            "app.services.PolicyEventIngestService.load_jsonl_rows",
+            side_effect=[[], documents],
+        ):
+            result = await service.import_handoff_batch(import_all=False, limit=5)
+
+        self.assertEqual(result.inserted_count, 0)
+        self.assertEqual(result.pending_import_count, 0)
+        self.assertIn("적재 완료", result.hint or "")
+
+    async def test_import_handoff_batch_raises_when_handoff_empty_but_pending_transform(self) -> None:
+        service = _make_ingest_service()
+        service._event_pin_repo.list_policy_api_ids = AsyncMock(return_value=set())
+        documents = [{"contentid": "1", "policy_api_id": 1, "pin_content": "a"}]
+
+        with patch(
+            "app.services.PolicyEventIngestService.load_jsonl_rows",
+            side_effect=[[], documents],
+        ):
+            with self.assertRaises(FileNotFoundError):
+                await service.import_handoff_batch(import_all=False, limit=5)
+
+    async def test_import_handoff_batch_skips_per_row_db_lookup(self) -> None:
+        service = _make_ingest_service()
+        service._event_pin_repo.get_by_policy_api_id = AsyncMock(return_value=None)
+        handoff_rows = [
+            {
+                "contentid": "1",
+                "policy_api_id": 1,
+                "title": "정책1",
+                "pin_content": "본문1",
+                "cardnews_images": [{"key": "k1", "url": "https://cdn.example/1.png"}],
+            },
+        ]
+        with (
+            patch(
+                "app.services.PolicyEventIngestService.load_jsonl_rows",
+                return_value=handoff_rows,
+            ),
+            patch(
+                "app.services.PolicyEventIngestService.settings.policy_prune_pipeline_after_import",
+                False,
+            ),
+            patch.object(service, "_insert_policy_pin", AsyncMock(return_value=101)),
+        ):
+            await service.import_handoff_batch(import_all=True)
+
+        service._event_pin_repo.get_by_policy_api_id.assert_not_awaited()
+
+    async def test_nested_failure_does_not_expunge_unrelated_session_state(self) -> None:
+        service = _make_ingest_service()
+        nested_ctx = MagicMock()
+        nested_ctx.__aenter__ = AsyncMock()
+        nested_ctx.__aexit__ = AsyncMock(return_value=False)
+        service._pin_repo.session.begin_nested = MagicMock(return_value=nested_ctx)
+        service._pin_repo.session.expunge = MagicMock()
+        service._pin_repo.session.expire = MagicMock()
+        handoff_rows = [
+            {
+                "contentid": "1",
+                "policy_api_id": 1,
+                "title": "정책1",
+                "pin_content": "본문1",
+                "cardnews_images": [{"key": "k1", "url": "https://cdn.example/1.png"}],
+            },
+        ]
+        with (
+            patch(
+                "app.services.PolicyEventIngestService.load_jsonl_rows",
+                return_value=handoff_rows,
+            ),
+            patch(
+                "app.services.PolicyEventIngestService.settings.policy_prune_pipeline_after_import",
+                False,
+            ),
+            patch.object(service, "_insert_policy_pin", AsyncMock(side_effect=RuntimeError("boom"))),
+        ):
+            result = await service.import_handoff_batch(import_all=True)
+
+        self.assertEqual(result.error_count, 1)
+        service._pin_repo.session.expunge.assert_not_called()
+        service._pin_repo.session.expire.assert_not_called()
+
+
+class PolicySyncPipelineAccumulationTest(unittest.IsolatedAsyncioTestCase):
+    async def test_import_results_accumulate_across_batches(self) -> None:
+        service = PolicyPinService()
+        ingest = _make_ingest_service()
+        ingest.get_imported_policy_api_ids = AsyncMock(return_value=set())
+
+        transform_batches = [
+            PolicyPinTransformResult(
+                input_path="in.jsonl",
+                output_path="out.jsonl",
+                processed_count=2,
+                error_count=0,
+                pins=[],
+                pending_count=3,
+                remaining_pending_count=1,
+            ),
+            PolicyPinTransformResult(
+                input_path="in.jsonl",
+                output_path="out.jsonl",
+                processed_count=1,
+                error_count=0,
+                pins=[],
+                pending_count=1,
+                remaining_pending_count=0,
+            ),
+        ]
+        import_batches = [
+            PolicyImportBatchResult(
+                inserted_count=2,
+                skipped_duplicate_count=0,
+                pending_import_count=1,
+                error_count=1,
+                errors=[{"policy_api_id": 1, "error": "e1"}],
+                items=[
+                    PolicyBatchItemResult(policy_api_id=1, action=PolicyBatchAction.CREATED),
+                    PolicyBatchItemResult(policy_api_id=2, action=PolicyBatchAction.ERROR, message="e1"),
+                ],
+                pin_ids=[11, 12],
+            ),
+            PolicyImportBatchResult(
+                inserted_count=1,
+                skipped_duplicate_count=0,
+                pending_import_count=0,
+                error_count=0,
+                errors=[],
+                items=[PolicyBatchItemResult(policy_api_id=3, action=PolicyBatchAction.CREATED)],
+                pin_ids=[13],
+            ),
+            PolicyImportBatchResult(
+                inserted_count=0,
+                skipped_duplicate_count=0,
+                pending_import_count=0,
+                error_count=0,
+                errors=[],
+                items=[],
+                pin_ids=[],
+            ),
+        ]
+
+        with (
+            patch(
+                "app.services.PolicyPinService.settings.policy_prune_pipeline_after_import",
+                False,
+            ),
+            patch.object(
+                service,
+                "search_and_save",
+                AsyncMock(
+                    return_value=PolicyPinSearchResult(
+                        query_start_date="20260610",
+                        query_end_date="20260612",
+                        count=0,
+                        pins=[],
+                        saved_documents_path="docs.jsonl",
+                        stats={},
+                    ),
+                ),
+            ),
+            patch.object(service, "transform_and_save", AsyncMock(side_effect=transform_batches)),
+            patch.object(ingest, "import_handoff_batch", AsyncMock(side_effect=import_batches)),
+            patch.object(PolicyEventIngestService, "write_sync_meta"),
+        ):
+            result = await service.sync_pipeline(
+                ingest_service=ingest,
+                s3_util=MagicMock(),
+                start_date="20260610",
+                end_date="20260612",
+                batch_size=2,
+            )
+
+        import_result = result.import_result
+        self.assertEqual(import_result.inserted_count, 3)
+        self.assertEqual(import_result.error_count, 1)
+        self.assertEqual(len(import_result.items), 3)
+        self.assertEqual(import_result.pin_ids, [11, 12, 13])
+        self.assertEqual(import_result.errors, [{"policy_api_id": 1, "error": "e1"}])
+        ingest.get_imported_policy_api_ids.assert_awaited_once()
+
+    async def test_import_drains_backlog_when_transform_produces_zero(self) -> None:
+        service = PolicyPinService()
+        ingest = _make_ingest_service()
+        ingest.get_imported_policy_api_ids = AsyncMock(return_value=set())
+
+        import_batches = [
+            PolicyImportBatchResult(
+                inserted_count=2,
+                skipped_duplicate_count=0,
+                pending_import_count=1,
+                error_count=0,
+                errors=[],
+                items=[
+                    PolicyBatchItemResult(policy_api_id=1, action=PolicyBatchAction.CREATED),
+                    PolicyBatchItemResult(policy_api_id=2, action=PolicyBatchAction.CREATED),
+                ],
+                pin_ids=[11, 12],
+            ),
+            PolicyImportBatchResult(
+                inserted_count=1,
+                skipped_duplicate_count=0,
+                pending_import_count=0,
+                error_count=0,
+                errors=[],
+                items=[PolicyBatchItemResult(policy_api_id=3, action=PolicyBatchAction.CREATED)],
+                pin_ids=[13],
+            ),
+        ]
+        import_mock = AsyncMock(side_effect=import_batches)
+
+        with (
+            patch(
+                "app.services.PolicyPinService.settings.policy_prune_pipeline_after_import",
+                False,
+            ),
+            patch.object(
+                service,
+                "search_and_save",
+                AsyncMock(
+                    return_value=PolicyPinSearchResult(
+                        query_start_date="20260610",
+                        query_end_date="20260612",
+                        count=0,
+                        pins=[],
+                        saved_documents_path="docs.jsonl",
+                        stats={},
+                    ),
+                ),
+            ),
+            patch.object(
+                service,
+                "transform_and_save",
+                AsyncMock(
+                    return_value=PolicyPinTransformResult(
+                        input_path="in.jsonl",
+                        output_path="out.jsonl",
+                        processed_count=0,
+                        error_count=0,
+                        pins=[],
+                        pending_count=0,
+                        remaining_pending_count=0,
+                    ),
+                ),
+            ),
+            patch.object(service, "handoff_path", return_value=Path("out.jsonl")),
+            patch(
+                "app.services.PolicyPinService.load_jsonl_rows",
+                return_value=[{"contentid": "1"}],
+            ),
+            patch.object(ingest, "import_handoff_batch", import_mock),
+            patch.object(PolicyEventIngestService, "write_sync_meta"),
+        ):
+            result = await service.sync_pipeline(
+                ingest_service=ingest,
+                s3_util=MagicMock(),
+                start_date="20260610",
+                end_date="20260612",
+                batch_size=2,
+            )
+
+        self.assertEqual(import_mock.await_count, 2)
+        import_result = result.import_result
+        self.assertEqual(import_result.inserted_count, 3)
+        self.assertEqual(import_result.pending_import_count, 0)
+        self.assertEqual(import_result.pin_ids, [11, 12, 13])
+
+    async def test_transform_loop_continues_while_remaining_pending(self) -> None:
+        service = PolicyPinService()
+        ingest = _make_ingest_service()
+        ingest.get_imported_policy_api_ids = AsyncMock(return_value=set())
+        transform_mock = AsyncMock(
+            side_effect=[
+                PolicyPinTransformResult(
+                    input_path="in.jsonl",
+                    output_path="out.jsonl",
+                    processed_count=20,
+                    error_count=5,
+                    pins=[],
+                    pending_count=25,
+                    remaining_pending_count=22,
+                ),
+                PolicyPinTransformResult(
+                    input_path="in.jsonl",
+                    output_path="out.jsonl",
+                    processed_count=22,
+                    error_count=0,
+                    pins=[],
+                    pending_count=22,
+                    remaining_pending_count=0,
+                ),
+            ],
+        )
+        import_mock = AsyncMock(
+            return_value=PolicyImportBatchResult(
+                inserted_count=1,
+                skipped_duplicate_count=0,
+                pending_import_count=0,
+                error_count=0,
+            ),
+        )
+
+        with (
+            patch(
+                "app.services.PolicyPinService.settings.policy_prune_pipeline_after_import",
+                False,
+            ),
+            patch.object(
+                service,
+                "search_and_save",
+                AsyncMock(
+                    return_value=PolicyPinSearchResult(
+                        query_start_date="20260610",
+                        query_end_date="20260612",
+                        count=42,
+                        pins=[],
+                        saved_documents_path="docs.jsonl",
+                        stats={},
+                    ),
+                ),
+            ),
+            patch.object(service, "transform_and_save", transform_mock),
+            patch.object(ingest, "import_handoff_batch", import_mock),
+            patch.object(PolicyEventIngestService, "write_sync_meta"),
+            patch.object(service, "handoff_path", return_value=Path("out.jsonl")),
+            patch("app.services.PolicyPinService.load_jsonl_rows", return_value=[]),
+        ):
+            result = await service.sync_pipeline(
+                ingest_service=ingest,
+                s3_util=MagicMock(),
+                batch_size=25,
+            )
+
+        self.assertEqual(transform_mock.await_count, 2)
+        self.assertEqual(result.transform.processed_count, 42)
+
+    async def test_sync_completes_when_transform_produces_zero_and_handoff_empty(self) -> None:
+        service = PolicyPinService()
+        ingest = _make_ingest_service()
+        ingest.get_imported_policy_api_ids = AsyncMock(return_value=set())
+        import_mock = AsyncMock()
+
+        with (
+            patch(
+                "app.services.PolicyPinService.settings.policy_prune_pipeline_after_import",
+                False,
+            ),
+            patch.object(
+                service,
+                "search_and_save",
+                AsyncMock(
+                    return_value=PolicyPinSearchResult(
+                        query_start_date="20260610",
+                        query_end_date="20260612",
+                        count=42,
+                        pins=[],
+                        saved_documents_path="docs.jsonl",
+                        stats={},
+                    ),
+                ),
+            ),
+            patch.object(
+                service,
+                "transform_and_save",
+                AsyncMock(
+                    return_value=PolicyPinTransformResult(
+                        input_path="in.jsonl",
+                        output_path="out.jsonl",
+                        processed_count=0,
+                        error_count=42,
+                        pins=[],
+                        pending_count=42,
+                        remaining_pending_count=42,
+                    ),
+                ),
+            ),
+            patch.object(service, "handoff_path", return_value=Path("out.jsonl")),
+            patch("app.services.PolicyPinService.load_jsonl_rows", return_value=[]),
+            patch.object(ingest, "import_handoff_batch", import_mock),
+            patch.object(PolicyEventIngestService, "write_sync_meta"),
+        ):
+            result = await service.sync_pipeline(
+                ingest_service=ingest,
+                s3_util=MagicMock(),
+            )
+
+        import_mock.assert_not_awaited()
+        self.assertIn("가공 0건", result.hint or "")
+        self.assertEqual(result.import_result.inserted_count, 0)
+
+    async def test_sync_hint_when_already_in_db(self) -> None:
+        service = PolicyPinService()
+        ingest = _make_ingest_service()
+        ingest.get_imported_policy_api_ids = AsyncMock(return_value={1, 2, 3})
+
+        with (
+            patch(
+                "app.services.PolicyPinService.settings.policy_prune_pipeline_after_import",
+                False,
+            ),
+            patch.object(
+                service,
+                "search_and_save",
+                AsyncMock(
+                    return_value=PolicyPinSearchResult(
+                        query_start_date="20260610",
+                        query_end_date="20260612",
+                        count=3,
+                        pins=[],
+                        saved_documents_path="docs.jsonl",
+                        stats={},
+                    ),
+                ),
+            ),
+            patch.object(
+                service,
+                "transform_and_save",
+                AsyncMock(
+                    return_value=PolicyPinTransformResult(
+                        input_path="in.jsonl",
+                        output_path="out.jsonl",
+                        processed_count=0,
+                        error_count=0,
+                        pins=[],
+                        remaining_pending_count=0,
+                    ),
+                ),
+            ),
+            patch.object(service, "handoff_path", return_value=Path("out.jsonl")),
+            patch("app.services.PolicyPinService.load_jsonl_rows", return_value=[]),
+            patch.object(PolicyEventIngestService, "write_sync_meta"),
+        ):
+            result = await service.sync_pipeline(
+                ingest_service=ingest,
+                s3_util=MagicMock(),
+            )
+
+        self.assertIn("이미 DB 반영 완료", result.hint or "")
+
+
+if __name__ == "__main__":
+    unittest.main()