feat(music-lab): get_last_failed_step — 파이프라인 재개용 실패 step 판별

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

music-lab/app/db.py

@@ -1220,6 +1220,18 @@ def list_pipeline_jobs(pid: int) -> List[Dict[str, Any]]:
     return [dict(r) for r in rows]
 
 
+def get_last_failed_step(pid: int) -> Optional[str]:
+    """파이프라인의 가장 최근 status='failed' pipeline_job의 step. 없으면 None."""
+    with _conn() as conn:
+        row = conn.execute(
+            "SELECT step FROM pipeline_jobs "
+            "WHERE pipeline_id = ? AND status = 'failed' "
+            "ORDER BY id DESC LIMIT 1",
+            (pid,),
+        ).fetchone()
+    return row["step"] if row else None
+
+
 def get_youtube_setup() -> Dict[str, Any]:
     """youtube_setup의 기본 1행을 반환. 누락 시 자동 시드 후 재조회."""
     with _conn() as conn:

music-lab/tests/test_pipeline_retry.py

@@ -0,0 +1,24 @@
+import pytest
+from app import db
+
+
+@pytest.fixture
+def fresh_db(monkeypatch, tmp_path):
+    db_path = tmp_path / "music.db"
+    monkeypatch.setattr(db, "DB_PATH", str(db_path))
+    db.init_db()
+    return db_path
+
+
+def test_get_last_failed_step_returns_step(fresh_db):
+    pid = db.create_pipeline(track_id=1)
+    job_id = db.create_pipeline_job(pid, "video")
+    db.update_pipeline_job(job_id, status="failed", error="boom")
+    db.update_pipeline_state(pid, "failed", failed_reason="video: boom")
+    assert db.get_last_failed_step(pid) == "video"
+
+
+def test_get_last_failed_step_none_when_no_failure(fresh_db):
+    pid = db.create_pipeline(track_id=1)
+    db.create_pipeline_job(pid, "cover")
+    assert db.get_last_failed_step(pid) is None