diff --git a/music-lab/app/pipeline/__init__.py b/music-lab/app/pipeline/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/music-lab/app/pipeline/state_machine.py b/music-lab/app/pipeline/state_machine.py
new file mode 100644
index 0000000..6e7271f
--- /dev/null
+++ b/music-lab/app/pipeline/state_machine.py
@@ -0,0 +1,41 @@
+"""파이프라인 상태 머신 — 전이 규칙 단일 소스."""
+
+STEPS = ["cover", "video", "thumb", "meta", "review", "publish"]
+USER_GATES = ["cover", "video", "thumb", "meta", "publish"]  # review는 자동
+
+_APPROVE_NEXT = {
+    "cover_pending":   "video_pending",
+    "video_pending":   "thumb_pending",
+    "thumb_pending":   "meta_pending",
+    "meta_pending":    "ai_review",      # 자동 검토 단계로
+    "publish_pending": "publishing",
+}
+
+TERMINAL_STATES = {"published", "cancelled", "failed", "awaiting_manual"}
+
+
+def next_state_on_approve(state: str) -> str:
+    if state not in _APPROVE_NEXT:
+        raise ValueError(f"승인 불가 상태: {state}")
+    return _APPROVE_NEXT[state]
+
+
+def next_state_on_reject(state: str) -> str:
+    if not state.endswith("_pending"):
+        raise ValueError(f"반려 불가 상태: {state}")
+    return state  # 같은 상태 유지 (재생성 후 다시 _pending)
+
+
+def can_transition(from_state: str, to_state: str) -> bool:
+    if from_state in TERMINAL_STATES:
+        return False
+    if to_state in {"cancelled", "failed", "awaiting_manual"}:
+        return True
+    if to_state == _APPROVE_NEXT.get(from_state):
+        return True
+    # 자동 전이 (ai_review → publish_pending, publishing → published)
+    auto_transitions = {
+        ("ai_review", "publish_pending"),
+        ("publishing", "published"),
+    }
+    return (from_state, to_state) in auto_transitions
diff --git a/music-lab/tests/test_state_machine.py b/music-lab/tests/test_state_machine.py
new file mode 100644
index 0000000..5c261a7
--- /dev/null
+++ b/music-lab/tests/test_state_machine.py
@@ -0,0 +1,49 @@
+import pytest
+from app.pipeline.state_machine import (
+    next_state_on_approve, next_state_on_reject, can_transition, STEPS, USER_GATES,
+)
+
+
+def test_steps_sequence():
+    assert STEPS == ["cover", "video", "thumb", "meta", "review", "publish"]
+
+
+def test_user_gates_excludes_review():
+    assert "review" not in USER_GATES
+    assert "publish" in USER_GATES
+    assert "cover" in USER_GATES
+
+
+def test_approve_progression():
+    assert next_state_on_approve("cover_pending") == "video_pending"
+    assert next_state_on_approve("video_pending") == "thumb_pending"
+    assert next_state_on_approve("thumb_pending") == "meta_pending"
+    assert next_state_on_approve("meta_pending") == "ai_review"
+    assert next_state_on_approve("publish_pending") == "publishing"
+
+
+def test_approve_invalid_state_raises():
+    with pytest.raises(ValueError):
+        next_state_on_approve("ai_review")  # 자동 전이 — approve 호출 자체가 무효
+
+
+def test_reject_keeps_same_state():
+    # 반려는 같은 *_pending 상태를 유지(재생성 트리거)
+    assert next_state_on_reject("cover_pending") == "cover_pending"
+    assert next_state_on_reject("publish_pending") == "publish_pending"
+
+
+def test_can_transition_blocks_terminal_states():
+    assert not can_transition("published", "cover_pending")
+    assert not can_transition("cancelled", "cover_pending")
+    assert not can_transition("failed", "cover_pending")
+
+
+def test_can_transition_allows_cancel_from_anywhere():
+    assert can_transition("cover_pending", "cancelled")
+    assert can_transition("publishing", "cancelled")
+
+
+def test_can_transition_allows_failed_from_pending():
+    assert can_transition("video_pending", "failed")
+    assert can_transition("publishing", "failed")