feat(publish): P0-1b enqueue 결선 — 저작 5경로 flag-gated (study→viewer)

study_question 발행 outbox enqueue 를 settings.study_publish_enabled 게이트로
5경로 결선(전부 같은 tx, caller commit = 콘텐츠 변경과 outbox INSERT 원자성):
- create_question_in_topic: 신규 문항 발행
- update_question: 문항 재투영(해설 ready 일 때만 동봉)
- soft_delete_question: tombstone(문항 + 해설 본문 존재 시 해설 kind)
- run_explanation_job (4-A 워커): 해설 ready → 문항+해설 발행
- generate_ai_explanation (실시간): 해설 ready → 문항+해설 발행

플래그 기본 false = 코드 inert(배포 후 GPU .env STUDY_PUBLISH_ENABLED 로 점등).
stale→tombstone 은 P1-3(해설 라이프사이클)로 분리.
검증: py_compile 6파일·결선 5곳 grep·플래그 기본 false.

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

app/api/study_questions.py

@@ -39,6 +39,8 @@ from services.study.explanation_rag import (
     gather_explanation_context,
     render_evidence_block,
 )
+from services.study.publish_enqueue import enqueue_publish, enqueue_question_publish
+from services.study.publish_projection import KIND_EXPLANATION, KIND_QUESTION
 
 logger = logging.getLogger(__name__)
 router = APIRouter()
@@ -543,6 +545,9 @@ async def create_question_in_topic(
     )
     session.add(q)
     await session.flush()
+    # 발행 outbox 적재(같은 tx, flag off 면 no-op) — 신규 문항 발행. P0-1b.
+    if settings.study_publish_enabled:
+        await enqueue_question_publish(session, q)
     await session.commit()
 
     stats = QuestionAttemptStats(attempt_count=0, correct_count=0, wrong_count=0)
@@ -908,6 +913,9 @@ async def update_question(
         await flag_cards_for_source(session, source_question_id=q.id, reason="source_changed")
 
     q.updated_at = datetime.now(timezone.utc)
+    # 발행 재투영(같은 tx) — 문항 갱신 반영. 해설은 ready 일 때만 동봉, stale→tombstone 은 P1-3. P0-1b.
+    if settings.study_publish_enabled:
+        await enqueue_question_publish(session, q)
     await session.commit()
 
     stats = await _attempt_stats(session, user.id, question_id)
@@ -971,6 +979,11 @@ async def soft_delete_question(
     # 공부 암기노트: 소스 문제 삭제 시 파생 암기카드를 검토 대기로 마킹(source_deleted).
     # study_questions 는 soft-delete 만이라 카드 FK CASCADE 는 미발동 — 이 훅이 실 경로.
     await flag_cards_for_source(session, source_question_id=q.id, reason="source_deleted")
+    # 발행 tombstone(같은 tx) — 삭제는 feed 1급 이벤트(raw DELETE 금지·워커 경유). 해설 본문 있으면 그 kind 도. P0-1b.
+    if settings.study_publish_enabled:
+        await enqueue_publish(session, kind=KIND_QUESTION, source_id=q.id, payload=None, deleted=True)
+        if q.ai_explanation:
+            await enqueue_publish(session, kind=KIND_EXPLANATION, source_id=q.id, payload=None, deleted=True)
     await session.commit()
 
 
@@ -1713,6 +1726,9 @@ async def generate_ai_explanation(
     primary_name = ai_client.ai.primary.model if hasattr(ai_client.ai.primary, "model") else "primary"
     q.ai_explanation_model = f"mlx:{primary_name}"
     q.updated_at = q.ai_explanation_generated_at
+    # 발행 재투영(같은 tx) — 실시간 해설 ready → 문항+해설 발행. P0-1b.
+    if settings.study_publish_enabled:
+        await enqueue_question_publish(session, q)
     await session.commit()
 
     return AIExplanationResponse(

app/workers/study_explanation_worker.py

@@ -33,6 +33,7 @@ from services.study.explanation_rag import (
     gather_explanation_context,
     render_evidence_block,
 )
+from services.study.publish_enqueue import enqueue_question_publish
 
 logger = logging.getLogger(__name__)
 
@@ -227,6 +228,10 @@ async def run_explanation_job(session: AsyncSession, job: StudyQuestionJob) -> N
         question.ai_explanation_model = f"mlx:{primary_name}"
         question.updated_at = question.ai_explanation_generated_at
 
+        # 발행 재투영(같은 tx, caller commit) — 4-A 해설 ready → 문항+해설 발행. P0-1b.
+        if settings.study_publish_enabled:
+            await enqueue_question_publish(session, question)
+
         job.status = "completed"
         job.completed_at = now()
         return