diff --git a/app/comment/api/routers/v1/comment.py b/app/comment/api/routers/v1/comment.py
index eb2fc76..7ba9e7a 100644
--- a/app/comment/api/routers/v1/comment.py
+++ b/app/comment/api/routers/v1/comment.py
@@ -9,7 +9,9 @@ Comment API Router
     - DELETE /comment/{comment_id}: 본인 댓글 소프트 삭제 (로그인 필수)
 """
 
-from fastapi import APIRouter, Depends
+from typing import Literal
+
+from fastapi import APIRouter, Depends, Query
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.comment.schemas.comment_schema import (
@@ -60,12 +62,16 @@ router = APIRouter(prefix="/comment", tags=["Comment"])
 async def post_comment(
     video_id: int,
     body: CommentCreateRequest,
+    type: Literal["video", "ssul"] = Query(
+        default="video",
+        description="콘텐츠 종류. video.id 와 ssul_content.id 가 겹치므로 반드시 함께 보낼 것",
+    ),
     current_user: User = Depends(get_current_user),
     session: AsyncSession = Depends(get_session),
 ) -> CommentCreateResponse:
     logger.info(
-        f"[post_comment] START - video_id: {video_id}, user: {current_user.user_uuid}, "
-        f"parent_id: {body.parent_id}"
+        f"[post_comment] START - type: {type}, id: {video_id}, "
+        f"user: {current_user.user_uuid}, parent_id: {body.parent_id}"
     )
     comment = await create_comment(
         session=session,
@@ -74,6 +80,7 @@ async def post_comment(
         nickname=body.nickname,
         content=body.content,
         parent_id=body.parent_id,
+        content_type=type,
     )
     logger.info(f"[post_comment] SUCCESS - comment_id: {comment.id}")
     return CommentCreateResponse(
@@ -112,12 +119,16 @@ async def post_comment(
 )
 async def get_comments(
     video_id: int,
+    type: Literal["video", "ssul"] = Query(
+        default="video",
+        description="콘텐츠 종류. video.id 와 ssul_content.id 가 겹치므로 반드시 함께 보낼 것",
+    ),
     current_user: User | None = Depends(get_current_user_optional),
     session: AsyncSession = Depends(get_session),
     pagination: PaginationParams = Depends(get_pagination_params),
 ) -> PaginatedResponse[CommentItem]:
     logger.info(
-        f"[get_comments] START - video_id: {video_id}, "
+        f"[get_comments] START - type: {type}, id: {video_id}, "
         f"page: {pagination.page}, page_size: {pagination.page_size}"
     )
     current_user_uuid = current_user.user_uuid if current_user else None
@@ -127,6 +138,7 @@ async def get_comments(
         page=pagination.page,
         page_size=pagination.page_size,
         current_user_uuid=current_user_uuid,
+        content_type=type,
     )
     logger.info(f"[get_comments] SUCCESS - total: {result.total}, items: {len(result.items)}")
     return result
diff --git a/app/comment/services/comment.py b/app/comment/services/comment.py
index ac680c3..fee5dbb 100644
--- a/app/comment/services/comment.py
+++ b/app/comment/services/comment.py
@@ -1,5 +1,5 @@
 from collections import defaultdict
-from typing import List, Optional
+from typing import List, Literal, Optional
 
 from fastapi import HTTPException
 from sqlalchemy import exists, select
@@ -7,16 +7,54 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.comment.models import Comment
 from app.comment.schemas.comment_schema import CommentItem, ReplyItem
+from app.ssulbox.models import SsulContent
 from app.utils.pagination import PaginatedResponse
 from app.video.models import Video
 
+ContentType = Literal["video", "ssul"]
+
+
+def _target_col(content_type: ContentType):
+    """댓글이 달린 대상을 가리키는 컬럼.
+
+    `comment` 는 ADO2 영상과 썰박스를 함께 담고 `video_id` / `content_id` 중
+    정확히 하나만 채운다(CHECK 로 강제). 종류는 어느 컬럼이 채워졌는지가 결정한다.
+    """
+    return Comment.video_id if content_type == "video" else Comment.content_id
+
+
+async def _ensure_target_exists(
+    session: AsyncSession,
+    content_type: ContentType,
+    target_id: int,
+) -> None:
+    """댓글 대상(영상 또는 썰박스 콘텐츠)이 존재·공개 상태인지 확인.
+
+    id 가 종류별 독립 시퀀스라, 반대쪽 테이블에 같은 id 가 있어도 잡으면 안 된다.
+    """
+    if content_type == "ssul":
+        q = select(SsulContent.id).where(
+            SsulContent.id == target_id,
+            SsulContent.status == "done",
+            SsulContent.is_deleted.is_(False),
+        )
+    else:
+        q = select(Video.id).where(
+            Video.id == target_id,
+            Video.status == "completed",
+            Video.is_deleted.is_(False),
+        )
+    if (await session.execute(q)).scalar_one_or_none() is None:
+        raise HTTPException(status_code=404, detail="콘텐츠를 찾을 수 없습니다.")
+
 
 async def _validate_parent(
     session: AsyncSession,
     parent_id: int,
-    video_id: int,
+    content_type: ContentType,
+    target_id: int,
 ) -> None:
-    """2-depth 제한 + 동일 video 검증."""
+    """2-depth 제한 + 동일 대상 검증."""
     result = await session.execute(
         select(Comment).where(
             Comment.id == parent_id,
@@ -27,8 +65,11 @@ async def _validate_parent(
 
     if parent is None:
         raise HTTPException(status_code=400, detail="부모 댓글을 찾을 수 없습니다.")
-    if parent.video_id != video_id:
-        raise HTTPException(status_code=400, detail="다른 영상의 댓글에는 대댓글을 달 수 없습니다.")
+    # 부모가 같은 종류의 같은 대상에 달렸는지 본다. video_id 만 비교하면
+    # 썰박스 댓글(video_id=NULL)에 ADO2 대댓글이 달리는 교차를 못 막는다.
+    parent_target = parent.video_id if content_type == "video" else parent.content_id
+    if parent_target != target_id:
+        raise HTTPException(status_code=400, detail="다른 콘텐츠의 댓글에는 대댓글을 달 수 없습니다.")
     if parent.parent_id is not None:
         raise HTTPException(status_code=400, detail="대댓글에는 대댓글을 달 수 없습니다. (최대 2-depth)")
 
@@ -73,24 +114,19 @@ async def create_comment(
     nickname: str,
     content: str,
     parent_id: Optional[int],
+    content_type: ContentType = "video",
 ) -> Comment:
-    # Video 존재 확인
-    video_result = await session.execute(
-        select(Video).where(
-            Video.id == video_id,
-            Video.status == "completed",
-            Video.is_deleted == False,  # noqa: E712
-        )
-    )
-    if video_result.scalar_one_or_none() is None:
-        raise HTTPException(status_code=404, detail="영상을 찾을 수 없습니다.")
+    # 대상 존재 확인
+    await _ensure_target_exists(session, content_type, video_id)
 
     # parent_id 검증
     if parent_id is not None:
-        await _validate_parent(session, parent_id, video_id)
+        await _validate_parent(session, parent_id, content_type, video_id)
 
     comment = Comment(
-        video_id=video_id,
+        # 종류에 따라 둘 중 하나만 채운다 (CHECK ck_comment_one_target 이 강제)
+        video_id=video_id if content_type == "video" else None,
+        content_id=video_id if content_type == "ssul" else None,
         user_uuid=user_uuid,
         nickname=nickname,
         parent_id=parent_id,
@@ -108,6 +144,7 @@ async def list_comments(
     page: int,
     page_size: int,
     current_user_uuid: Optional[str],
+    content_type: ContentType = "video",
 ) -> PaginatedResponse[CommentItem]:
     offset = (page - 1) * page_size
 
@@ -121,9 +158,10 @@ async def list_comments(
         .correlate(Comment)
     )
 
-    # 최상위 댓글 필터: 삭제 안 됐거나 살아있는 대댓글이 있는 것
+    # 최상위 댓글 필터: 삭제 안 됐거나 살아있는 대댓글이 있는 것.
+    # 종류에 맞는 대상 컬럼으로 걸러야 같은 id 의 반대 종류 댓글이 섞이지 않는다.
     parent_where = [
-        Comment.video_id == video_id,
+        _target_col(content_type) == video_id,
         Comment.parent_id.is_(None),
         (Comment.is_deleted == False) | has_live_reply,  # noqa: E712
     ]
diff --git a/app/social/models.py b/app/social/models.py
index 185fa8e..a96ed65 100644
--- a/app/social/models.py
+++ b/app/social/models.py
@@ -7,7 +7,17 @@ Social Media Models
 from datetime import datetime
 from typing import TYPE_CHECKING, Optional
 
-from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, Integer, String, Text, func
+from sqlalchemy import (
+    BigInteger,
+    CheckConstraint,
+    DateTime,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+    func,
+)
 from sqlalchemy.dialects.mysql import JSON
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
@@ -53,6 +63,13 @@ class SocialUpload(Base):
 
     __tablename__ = "social_upload"
     __table_args__ = (
+        # ADO2 영상과 썰박스 콘텐츠를 함께 담는다(2026-07-30 병합,
+        # docs/manual_ddl/2026-07-30-social-upload-merge.sql).
+        # 대상은 video_id / content_id 중 **정확히 하나**만 채워진다.
+        CheckConstraint(
+            "(video_id IS NULL) <> (content_id IS NULL)",
+            name="ck_social_upload_one_target",
+        ),
         Index("idx_social_upload_user_uuid", "user_uuid"),
         Index("idx_social_upload_video_id", "video_id"),
         Index("idx_social_upload_social_account_id", "social_account_id"),
@@ -61,8 +78,14 @@ class SocialUpload(Base):
         Index("idx_social_upload_created_at", "created_at"),
         # 동일 영상+채널 조합 조회용 인덱스 (유니크 아님 - 여러 번 업로드 가능)
         Index("idx_social_upload_video_account", "video_id", "social_account_id"),
-        # 순번 조회용 인덱스
+        # 순번 조회용 인덱스 (종류별. content 쪽은 선행 컬럼이라 FK 인덱스도 겸한다)
         Index("idx_social_upload_seq", "video_id", "social_account_id", "upload_seq"),
+        Index(
+            "idx_social_upload_content_seq",
+            "content_id",
+            "social_account_id",
+            "upload_seq",
+        ),
         {
             "mysql_engine": "InnoDB",
             "mysql_charset": "utf8mb4",
@@ -91,11 +114,19 @@ class SocialUpload(Base):
         comment="사용자 UUID (User.user_uuid 참조)",
     )
 
-    video_id: Mapped[int] = mapped_column(
+    # 대상은 아래 둘 중 **정확히 하나**만 채워진다 (ck_social_upload_one_target).
+    video_id: Mapped[Optional[int]] = mapped_column(
         Integer,
         ForeignKey("video.id", ondelete="CASCADE"),
-        nullable=False,
-        comment="Video 외래키",
+        nullable=True,
+        comment="ADO2 영상 id (썰박스 업로드면 NULL)",
+    )
+    content_id: Mapped[Optional[int]] = mapped_column(
+        # ssul_content.id 는 BIGINT 다. INT 로 두면 FK 타입 불일치(errno 3780).
+        BigInteger,
+        ForeignKey("ssul_content.id", ondelete="CASCADE"),
+        nullable=True,
+        comment="썰박스 콘텐츠 id (ADO2 업로드면 NULL)",
     )
 
     social_account_id: Mapped[int] = mapped_column(
@@ -242,8 +273,10 @@ class SocialUpload(Base):
     # ==========================================================================
     # Relationships
     # ==========================================================================
-    video: Mapped["Video"] = relationship(
+    # 썰박스 업로드면 None 이다. 접근하는 쪽에서 반드시 방어할 것.
+    video: Mapped[Optional["Video"]] = relationship(
         "Video",
+        foreign_keys=[video_id],
         lazy="selectin",
     )
 
diff --git a/app/social/schemas/upload_schema.py b/app/social/schemas/upload_schema.py
index d33e167..f104e3a 100644
--- a/app/social/schemas/upload_schema.py
+++ b/app/social/schemas/upload_schema.py
@@ -3,7 +3,7 @@
 """
 
 from datetime import datetime
-from typing import Any, Optional
+from typing import Any, Literal, Optional
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -13,7 +13,14 @@ from app.social.constants import PrivacyStatus, UploadStatus
 class SocialUploadRequest(BaseModel):
     """소셜 업로드 요청"""
 
-    video_id: int = Field(..., description="업로드할 영상 ID")
+    # ⚠️ video_id 는 content_type 안에서만 유일하다 — video.id 와 ssul_content.id 는
+    #    각각 1부터 시작하는 독립 시퀀스라 값이 겹친다. content_type 없이 썰박스 id 를
+    #    보내면 **id 가 겹치는 남의 ADO2 영상이 업로드된다.**
+    content_type: Literal["video", "ssul"] = Field(
+        default="video",
+        description="콘텐츠 종류 (video: ADO2 영상, ssul: 썰박스)",
+    )
+    video_id: int = Field(..., description="업로드할 콘텐츠 ID (content_type 안에서만 유일)")
     social_account_id: int = Field(..., description="업로드할 소셜 계정 ID (연동 계정 목록의 id)")
     title: str = Field(..., min_length=1, max_length=100, description="영상 제목")
     description: Optional[str] = Field(
@@ -77,7 +84,9 @@ class SocialUploadStatusResponse(BaseModel):
     """업로드 상태 조회 응답"""
 
     upload_id: int = Field(..., description="업로드 작업 ID")
-    video_id: int = Field(..., description="영상 ID")
+    # 썰박스 업로드면 video_id 가 None 이고 content_id 가 채워진다 (정확히 하나만)
+    video_id: Optional[int] = Field(None, description="ADO2 영상 ID (썰박스면 None)")
+    content_id: Optional[int] = Field(None, description="썰박스 콘텐츠 ID (ADO2 면 None)")
     social_account_id: int = Field(..., description="소셜 계정 ID")
     upload_seq: int = Field(..., description="업로드 순번 (동일 영상+채널 조합 내 순번)")
     platform: str = Field(..., description="플랫폼명")
@@ -119,7 +128,9 @@ class SocialUploadHistoryItem(BaseModel):
     """업로드 이력 아이템"""
 
     upload_id: int = Field(..., description="업로드 작업 ID")
-    video_id: int = Field(..., description="영상 ID")
+    # 썰박스 업로드면 video_id 가 None 이고 content_id 가 채워진다 (정확히 하나만)
+    video_id: Optional[int] = Field(None, description="ADO2 영상 ID (썰박스면 None)")
+    content_id: Optional[int] = Field(None, description="썰박스 콘텐츠 ID (ADO2 면 None)")
     social_account_id: int = Field(..., description="소셜 계정 ID")
     upload_seq: int = Field(..., description="업로드 순번 (동일 영상+채널 조합 내 순번)")
     platform: str = Field(..., description="플랫폼명")
diff --git a/app/social/services/upload_service.py b/app/social/services/upload_service.py
index a2da831..023ffda 100644
--- a/app/social/services/upload_service.py
+++ b/app/social/services/upload_service.py
@@ -28,6 +28,8 @@ from app.social.schemas import (
 from app.social.services.account_service import SocialAccountService
 from app.social.worker.upload_task import process_social_upload
 from app.user.models import User
+from app.home.models import Project
+from app.ssulbox.models import SsulContent
 from app.video.models import Video
 
 logger = logging.getLogger(__name__)
@@ -55,26 +57,61 @@ class SocialUploadService:
         logger.info(
             f"[UPLOAD_SERVICE] 업로드 요청 - "
             f"user_uuid: {current_user.user_uuid}, "
-            f"video_id: {body.video_id}, "
+            f"type: {body.content_type}, id: {body.video_id}, "
             f"social_account_id: {body.social_account_id}"
         )
 
-        # 1. 영상 조회 및 검증
-        video_result = await session.execute(
-            select(Video).where(Video.id == body.video_id)
-        )
-        video = video_result.scalar_one_or_none()
+        # 1. 대상 조회 및 검증.
+        # video.id 와 ssul_content.id 는 값이 겹치는 독립 시퀀스라 content_type 으로
+        # 정확히 한 테이블만 봐야 한다 — 아니면 남의 다른 콘텐츠가 업로드된다.
+        # `target_col` 은 이후 중복 확인·채번에서도 같은 컬럼을 쓰기 위한 것이다.
+        # ⚠️ 소유자 필터가 필수다. 대상 id 는 클라이언트가 보내는 값이라, 안 거르면
+        # **남의 콘텐츠를 자기 SNS 채널에 업로드**할 수 있다(계정 소유권만 검증하고
+        # 콘텐츠 소유권을 안 보면 IDOR 이 된다).
+        if body.content_type == "ssul":
+            target_col = SocialUpload.content_id
+            content = (
+                await session.execute(
+                    select(SsulContent).where(
+                        SsulContent.id == body.video_id,
+                        SsulContent.user_uuid == current_user.user_uuid,
+                        SsulContent.is_deleted.is_(False),
+                    )
+                )
+            ).scalar_one_or_none()
+            if not content:
+                logger.warning(f"[UPLOAD_SERVICE] 썰박스 콘텐츠 없음 - id: {body.video_id}")
+                raise VideoNotFoundError(video_id=body.video_id)
+            if content.status != "done" or not content.video_url:
+                logger.warning(f"[UPLOAD_SERVICE] 썰박스 영상 미완성 - id: {body.video_id}")
+                raise VideoNotFoundError(
+                    video_id=body.video_id,
+                    detail="영상이 아직 준비되지 않았습니다. 영상 생성이 완료된 후 시도해주세요.",
+                )
+        else:
+            target_col = SocialUpload.video_id
+            # video 에는 user_uuid 가 없다 — 소유권은 project 에 있으므로 조인한다.
+            video = (
+                await session.execute(
+                    select(Video)
+                    .join(Project, Video.project_id == Project.id)
+                    .where(
+                        Video.id == body.video_id,
+                        Project.user_uuid == current_user.user_uuid,
+                    )
+                )
+            ).scalar_one_or_none()
 
-        if not video:
-            logger.warning(f"[UPLOAD_SERVICE] 영상 없음 - video_id: {body.video_id}")
-            raise VideoNotFoundError(video_id=body.video_id)
+            if not video:
+                logger.warning(f"[UPLOAD_SERVICE] 영상 없음 - video_id: {body.video_id}")
+                raise VideoNotFoundError(video_id=body.video_id)
 
-        if not video.result_movie_url:
-            logger.warning(f"[UPLOAD_SERVICE] 영상 URL 없음 - video_id: {body.video_id}")
-            raise VideoNotFoundError(
-                video_id=body.video_id,
-                detail="영상이 아직 준비되지 않았습니다. 영상 생성이 완료된 후 시도해주세요.",
-            )
+            if not video.result_movie_url:
+                logger.warning(f"[UPLOAD_SERVICE] 영상 URL 없음 - video_id: {body.video_id}")
+                raise VideoNotFoundError(
+                    video_id=body.video_id,
+                    detail="영상이 아직 준비되지 않았습니다. 영상 생성이 완료된 후 시도해주세요.",
+                )
 
         # 2. 소셜 계정 조회 및 소유권 검증
         account = await self._account_service.get_account_by_id(
@@ -96,7 +133,7 @@ class SocialUploadService:
         # 3-1. 진행 중인 업로드 확인 (즉시 pending 또는 uploading)
         in_progress_result = await session.execute(
             select(SocialUpload).where(
-                SocialUpload.video_id == body.video_id,
+                target_col == body.video_id,
                 SocialUpload.social_account_id == account.id,
                 SocialUpload.status.in_([UploadStatus.PENDING.value, UploadStatus.UPLOADING.value]),
                 or_(
@@ -122,7 +159,7 @@ class SocialUploadService:
         # 3-2. 미래 예약 업로드 확인
         scheduled_result = await session.execute(
             select(SocialUpload).where(
-                SocialUpload.video_id == body.video_id,
+                target_col == body.video_id,
                 SocialUpload.social_account_id == account.id,
                 SocialUpload.status == UploadStatus.PENDING.value,
                 SocialUpload.scheduled_at.isnot(None),
@@ -148,7 +185,7 @@ class SocialUploadService:
         # 4. 업로드 순번 계산
         max_seq_result = await session.execute(
             select(func.coalesce(func.max(SocialUpload.upload_seq), 0)).where(
-                SocialUpload.video_id == body.video_id,
+                target_col == body.video_id,
                 SocialUpload.social_account_id == account.id,
             )
         )
@@ -157,7 +194,9 @@ class SocialUploadService:
         # 5. 새 업로드 레코드 생성
         social_upload = SocialUpload(
             user_uuid=current_user.user_uuid,
-            video_id=body.video_id,
+            # 종류에 따라 둘 중 하나만 채운다 (CHECK ck_social_upload_one_target 이 강제)
+            video_id=body.video_id if body.content_type == "video" else None,
+            content_id=body.video_id if body.content_type == "ssul" else None,
             social_account_id=account.id,
             upload_seq=next_seq,
             platform=account.platform,
@@ -225,6 +264,7 @@ class SocialUploadService:
         return SocialUploadStatusResponse(
             upload_id=upload.id,
             video_id=upload.video_id,
+            content_id=upload.content_id,
             social_account_id=upload.social_account_id,
             upload_seq=upload.upload_seq,
             platform=upload.platform,
@@ -318,6 +358,7 @@ class SocialUploadService:
             SocialUploadHistoryItem(
                 upload_id=upload.id,
                 video_id=upload.video_id,
+                content_id=upload.content_id,
                 social_account_id=upload.social_account_id,
                 upload_seq=upload.upload_seq,
                 platform=upload.platform,
diff --git a/app/social/worker/upload_task.py b/app/social/worker/upload_task.py
index f2659a6..6c5a9c9 100644
--- a/app/social/worker/upload_task.py
+++ b/app/social/worker/upload_task.py
@@ -5,8 +5,6 @@ Social Upload Background Task
 """
 
 import logging
-import os
-import tempfile
 from pathlib import Path
 from typing import Optional
 
@@ -21,11 +19,12 @@ from config import social_upload_settings
 from app.dashboard.tasks import insert_dashboard
 from app.database.session import BackgroundSessionLocal
 from app.social.constants import SocialPlatform, UploadStatus
-from app.social.exceptions import TokenExpiredError, UploadError, UploadQuotaExceededError
+from app.social.exceptions import TokenExpiredError, UploadQuotaExceededError
 from app.social.models import SocialUpload
 from app.social.services import social_account_service
 from app.social.uploader import get_uploader
 from app.social.uploader.base import UploadMetadata
+from app.ssulbox.models import SsulContent
 from app.user.models import SocialAccount
 from app.video.models import Video
 
@@ -169,16 +168,27 @@ async def process_social_upload(upload_id: int) -> None:
                 logger.error(f"[SOCIAL_UPLOAD] 업로드 레코드 없음 - upload_id: {upload_id}")
                 return
 
-            # 2. Video 정보 조회
-            video_result = await session.execute(
-                select(Video).where(Video.id == upload.video_id)
-            )
-            video = video_result.scalar_one_or_none()
+            # 2. 대상 영상 조회.
+            # 병합 스키마: video_id / content_id 중 정확히 하나만 채워져 있다(CHECK).
+            # 채워진 쪽이 곧 종류다 — 썰박스면 ssul_content.video_url 을 쓴다.
+            if upload.content_id is not None:
+                content_result = await session.execute(
+                    select(SsulContent).where(SsulContent.id == upload.content_id)
+                )
+                content = content_result.scalar_one_or_none()
+                source_url = content.video_url if content else None
+            else:
+                video_result = await session.execute(
+                    select(Video).where(Video.id == upload.video_id)
+                )
+                video = video_result.scalar_one_or_none()
+                source_url = video.result_movie_url if video else None
 
-            if not video or not video.result_movie_url:
+            if not source_url:
                 logger.error(
                     f"[SOCIAL_UPLOAD] 영상 없음 또는 URL 없음 - "
-                    f"upload_id: {upload_id}, video_id: {upload.video_id}"
+                    f"upload_id: {upload_id}, video_id: {upload.video_id}, "
+                    f"content_id: {upload.content_id}"
                 )
                 await _update_upload_status(
                     upload_id=upload_id,
@@ -206,7 +216,7 @@ async def process_social_upload(upload_id: int) -> None:
                 return
 
             # 필요한 정보 저장
-            video_url = video.result_movie_url
+            video_url = source_url
             platform = SocialPlatform(upload.platform)
             upload_title = upload.title
             upload_description = upload.description
@@ -347,7 +357,7 @@ async def process_social_upload(upload_id: int) -> None:
                     f"upload_id: {upload_id}, error: {result.error_message}"
                 )
 
-    except UploadQuotaExceededError as e:
+    except UploadQuotaExceededError:
         logger.error(f"[SOCIAL_UPLOAD] API 할당량 초과 - upload_id: {upload_id}")
         await _update_upload_status(
             upload_id=upload_id,
diff --git a/app/video/api/routers/v1/video.py b/app/video/api/routers/v1/video.py
index 348c440..1131556 100644
--- a/app/video/api/routers/v1/video.py
+++ b/app/video/api/routers/v1/video.py
@@ -42,6 +42,7 @@ from app.database.like_cache import (
 from app.credit.exceptions import InsufficientCreditError
 from app.credit.services.credit_service import deduct_credit_for_job
 from app.ssulbox.constants import JOB_TYPE_VIDEO as CREDIT_JOB_TYPE_VIDEO
+from app.ssulbox.models import SsulContent
 from app.utils.logger import get_logger
 from app.video.models import Video, VideoReaction
 from app.video.services import unified_list
@@ -55,7 +56,10 @@ from app.video.schemas.video_schema import (
     VideoRenderData,
     VideoThumbnailItem,
 )
-from app.video.worker.video_task import download_and_upload_video_to_blob
+from app.video.worker.video_task import (
+    _fail_and_refund,
+    download_and_upload_video_to_blob,
+)
 
 
 from config import creatomate_settings
@@ -181,10 +185,17 @@ async def generate_video(
             # ===== 순차 쿼리 실행: Project, MarketingIntel, Lyric, Song, Image =====
             # Note: AsyncSession은 동일 세션에서 병렬 쿼리를 지원하지 않음
 
-            # Project 조회
+            # Project 조회 (본인 소유만).
+            # ⚠️ task_id 는 경로 파라미터라 남의 값을 넣을 수 있다. 소유자를 안 거르면
+            # 남의 프로젝트로 영상을 만들 수 있고, 더 나쁘게는 크레딧 원장에
+            # job_ref=피해자 task_id 로 차감이 기록돼 **피해자의 정상 생성이
+            # "이미 차감됨"으로 처리**된다(멱등 키 오염).
             project_result = await session.execute(
                 select(Project)
-                .where(Project.task_id == task_id)
+                .where(
+                    Project.task_id == task_id,
+                    Project.user_uuid == current_user.user_uuid,
+                )
                 .order_by(Project.created_at.desc())
                 .limit(1)
             )
@@ -521,17 +532,13 @@ async def generate_video(
         )
         import traceback
         logger.error(traceback.format_exc())
-        # 외부 API 실패 시 Video 상태를 failed로 업데이트
-        from app.database.session import AsyncSessionLocal
-
-        async with AsyncSessionLocal() as update_session:
-            video_result = await update_session.execute(
-                select(Video).where(Video.id == video_id)
-            )
-            video_to_update = video_result.scalar_one_or_none()
-            if video_to_update:
-                video_to_update.status = "failed"
-                await update_session.commit()
+        # 외부 API 실패 시 Video 상태를 failed로 갱신하고, 선차감한 크레딧을 환불한다.
+        # 크레딧은 1단계에서 이미 차감됐으므로 여기서 돌려주지 않으면 그대로 소멸된다.
+        await _fail_and_refund(
+            task_id,
+            user_uuid=current_user.user_uuid,
+            reason="영상 생성 실패 환불 (Creatomate 요청 오류)",
+        )
         return GenerateVideoResponse(
             success=False,
             task_id=task_id,
@@ -666,20 +673,41 @@ async def get_video_status(
         message = status_messages.get(status, f"상태: {status}")
         
         video_id = None
+
+        # ⚠️ 소유자 검증이 필수다. creatomate_render_id 는 클라이언트가 보내는 값이라
+        # 남의 렌더 ID 로 이 엔드포인트를 부를 수 있다. 소유자를 안 거르면
+        #   - 실패 분기: 남의 실패 건으로 **호출자에게 환불**이 나가고(크레딧 탈취),
+        #     원장 멱등 키(job_ref=피해자 task_id)가 소진돼 **피해자의 정당한 환불이 봉쇄**된다.
+        #   - 성공 분기: 남의 영상이 **호출자 UUID 경로의 Blob** 으로 업로드된다.
+        # video 에는 user_uuid 가 없으므로(소유권은 project 에 있다) Project 를 조인한다.
+        async def _load_owned_video() -> Video | None:
+            row = (
+                await session.execute(
+                    select(Video)
+                    .join(Project, Video.project_id == Project.id)
+                    .where(
+                        Video.creatomate_render_id == creatomate_render_id,
+                        Project.user_uuid == current_user.user_uuid,
+                    )
+                    .order_by(Video.created_at.desc())
+                    .limit(1)
+                )
+            ).scalar_one_or_none()
+            if row is None:
+                logger.warning(
+                    "[get_video_status] 소유자 아님 또는 영상 없음 — 후속 처리 생략, "
+                    f"creatomate_render_id: {creatomate_render_id}, "
+                    f"user: {current_user.user_uuid}"
+                )
+            return row
+
         # succeeded 상태인 경우 백그라운드 태스크 실행
         if status == "succeeded" and video_url:
-            # creatomate_render_id로 Video 조회하여 task_id 가져오기
-            video_result = await session.execute(
-                select(Video)
-                .where(Video.creatomate_render_id == creatomate_render_id)
-                .order_by(Video.created_at.desc())
-                .limit(1)
-            )
-            video = video_result.scalar_one_or_none()
-
-            video_id = video.id
+            # creatomate_render_id로 Video 조회하여 task_id 가져오기 (본인 것만)
+            video = await _load_owned_video()
 
             if video and video.status != "completed":
+                video_id = video.id
                 # 이미 완료된 경우 백그라운드 작업 중복 실행 방지
                 # 백그라운드 태스크로 MP4 다운로드 → Blob 업로드 → DB 업데이트 → 임시 파일 삭제
                 logger.info(
@@ -693,9 +721,25 @@ async def get_video_status(
                     user_uuid=current_user.user_uuid,
                 )
             elif video and video.status == "completed":
+                video_id = video.id
                 logger.debug(
                     f"[get_video_status] SKIPPED - Video already completed, creatomate_render_id: {creatomate_render_id}"
                 )
+        elif status == "failed":
+            # 렌더 실패는 Creatomate 가 명시적으로 알려준 시점에만 알 수 있다.
+            # 크레딧은 generate_video 에서 선차감됐으므로 여기서 돌려줘야 한다.
+            # 조회를 본인 소유로 제한했으므로 환불 대상이 곧 소유자다.
+            video = await _load_owned_video()
+
+            if video:
+                video_id = video.id
+                if video.status != "failed":
+                    await _fail_and_refund(
+                        video.task_id,
+                        creatomate_render_id=creatomate_render_id,
+                        user_uuid=current_user.user_uuid,
+                        reason="영상 생성 실패 환불 (Creatomate 렌더 실패)",
+                    )
 
         render_data = VideoRenderData(
             id=result.get("id"),
@@ -951,57 +995,77 @@ async def get_all_videos(
 )
 async def toggle_like(
     video_id: int,
+    type: Literal["video", "ssul"] = Query(
+        default="video",
+        description="콘텐츠 종류. video.id 와 ssul_content.id 가 겹치므로 반드시 함께 보낼 것",
+    ),
     current_user: User = Depends(get_current_user),
     session: AsyncSession = Depends(get_session),
 ) -> LikeToggleResponse:
-    """영상 좋아요를 토글합니다.
+    """영상/썰박스 좋아요를 토글합니다.
 
     Write-Behind 패턴:
     1. Redis user-set / count를 즉시 원자적으로 업데이트 (Lua script)
     2. dirty SET에 표시 → 스케줄러가 1분마다 MySQL에 반영
     DB write가 없으므로 고트래픽에서도 응답 지연 없음.
+
+    두 종류가 같은 테이블(video_reaction)·같은 Redis 로직을 쓰므로 엔드포인트도
+    하나다. type 은 ① 존재 확인 대상 ② Redis 키 접두 ③ backfill 컬럼만 가른다.
+    기본값이 "video" 라 기존 프론트 호출은 수정 없이 동작한다.
     """
-    logger.info(f"[toggle_like] START - video_id: {video_id}, user: {current_user.user_uuid}")
+    logger.info(
+        f"[toggle_like] START - type: {type}, id: {video_id}, user: {current_user.user_uuid}"
+    )
 
     try:
-        # 영상 존재 확인 (DB read는 유지 — 404 처리 필수)
-        video_result = await session.execute(
-            select(Video).where(
+        # 대상 존재 확인 (DB read는 유지 — 404 처리 필수).
+        # id 가 종류별 독립 시퀀스라 반대쪽 테이블에 같은 id 가 있어도 잡으면 안 된다.
+        if type == "ssul":
+            exists_q = select(SsulContent.id).where(
+                SsulContent.id == video_id,
+                SsulContent.status == "done",
+                SsulContent.is_deleted.is_(False),
+            )
+            # DB backfill 시 반응 행을 찾는 컬럼 — 썰박스 행은 content_id 가 채워져 있다
+            target_col = VideoReaction.content_id
+        else:
+            exists_q = select(Video.id).where(
                 Video.id == video_id,
                 Video.status == "completed",
-                Video.is_deleted == False,  # noqa: E712
+                Video.is_deleted.is_(False),
             )
-        )
-        if video_result.scalar_one_or_none() is None:
-            raise HTTPException(status_code=404, detail="영상을 찾을 수 없습니다.")
+            target_col = VideoReaction.video_id
+
+        if (await session.execute(exists_q)).scalar_one_or_none() is None:
+            raise HTTPException(status_code=404, detail="콘텐츠를 찾을 수 없습니다.")
 
         # Cold-start 보정: Redis에 데이터가 없으면 DB에서 backfill
-        count = await get_like_count(video_id)
+        count = await get_like_count(video_id, ctype=type)
         if count is None:
             # 카운트와 user-set 모두 없음 → DB에서 전체 복구
             user_uuids = (await session.execute(
-                select(VideoReaction.user_uuid)
-                .where(VideoReaction.video_id == video_id)
+                select(VideoReaction.user_uuid).where(target_col == video_id)
             )).scalars().all()
-            await backfill_user_set(video_id, list(user_uuids))
-            await set_like_count(video_id, len(user_uuids))
+            await backfill_user_set(video_id, list(user_uuids), ctype=type)
+            await set_like_count(video_id, len(user_uuids), ctype=type)
         elif count > 0:
-            if not await is_user_set_exists(video_id):
+            if not await is_user_set_exists(video_id, ctype=type):
                 # 카운트는 있지만 user-set이 증발한 경우 (부분 캐시 미스)
                 user_uuids = (await session.execute(
-                    select(VideoReaction.user_uuid)
-                    .where(VideoReaction.video_id == video_id)
+                    select(VideoReaction.user_uuid).where(target_col == video_id)
                 )).scalars().all()
-                await backfill_user_set(video_id, list(user_uuids))
+                await backfill_user_set(video_id, list(user_uuids), ctype=type)
 
         # Lua 스크립트로 원자적 토글 (race condition 방지)
-        is_liked, like_count = await toggle_like_atomic(video_id, current_user.user_uuid)
+        is_liked, like_count = await toggle_like_atomic(
+            video_id, current_user.user_uuid, ctype=type
+        )
 
         # dirty SET에 표시 → 스케줄러가 DB에 반영
-        await mark_dirty(video_id, current_user.user_uuid)
+        await mark_dirty(video_id, current_user.user_uuid, ctype=type)
 
         logger.info(
-            f"[toggle_like] SUCCESS - video_id: {video_id}, "
+            f"[toggle_like] SUCCESS - type: {type}, id: {video_id}, "
             f"is_liked: {is_liked}, count: {like_count}"
         )
         return LikeToggleResponse(video_id=video_id, is_liked=is_liked, like_count=like_count)