Phase I 실행 완료 + 프로세스 재설계 (Stage 2.5 → Stage 5)

Phase I: 전수 정합성 복구 + 넘침 처리 패러다임 전환 (14개 항목)
- I-14: SSE 유틸 공통 추출 (src/sse_utils.py 신규, 3개 파일 중복 제거)
- I-13: dead code 3건 삭제 (_call_anthropic_direct, _extract_sse_text x2) + import anthropic 제거
- I-1: STEP_B_PROMPT purpose 가이드 미존재 블록 3개 → 실존 블록 교체
- I-2: catalog.yaml not_for 13건 미존재 블록 참조 교체/제거
- I-12: BLOCK_SLOTS 주석 개수 수정 (cards 9, visuals 6, emphasis 10)
- I-10: INDEX.md 38개 동기화 (삭제된 8개 블록 행 제거)
- I-11: README.md 38개 동기화 (_legacy 제거, 트리/개수 정리)
- I-3: PURPOSE_FALLBACK 상수 + purpose 기반 미등록 블록 교체
- I-7: compare-pill-pair 단독 사용 금지 검증
- I-4: 38개 블록 전체에 slot_desc 추가
- I-5: 편집자 프롬프트에 slot_desc 전달 로직
- I-6: 제목 유사도 70% 초과 시 자동 교정
- I-9: 넘침 판단 Kei API 호출 (KEI_OVERFLOW_PROMPT, call_kei_overflow_judgment)
- I-8: 대형 콘텐츠 정보 Kei overflow 프롬프트에 포함

프로세스 재설계:
- Stage 2.5 제거 → Stage 5에서 Sonnet 감지 + Kei 판단 통합
- _review_balance() 확장: zone 예산 + overflow_detected action 추가
- Stage 5 루프에 Kei 넘침 판단 호출 통합
- _apply_adjustments()에 kei_trim/kei_restructure action 추가
- _build_overflow_context(), _convert_kei_judgment() 헬퍼 함수 추가
- DOWNGRADE_MAP은 Kei API 실패 시 비상용으로만 잔존

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

src/content_editor.py

@@ -19,6 +19,7 @@ import httpx
 
 from src.config import settings
 from src.design_director import BLOCK_SLOTS
+from src.sse_utils import stream_sse_tokens
 
 logger = logging.getLogger(__name__)
 
@@ -92,6 +93,12 @@ async def fill_content(
                 f"  선택 슬롯: {slots.get('optional', [])}"
             )
 
+            # I-5: 슬롯 의미 설명 전달 (slot_desc가 있으면)
+            slot_desc = slots.get("slot_desc", {})
+            if slot_desc:
+                desc_lines = [f"    {k}: {v}" for k, v in slot_desc.items()]
+                req_text += "\n  슬롯 설명:\n" + "\n".join(desc_lines)
+
             if char_guide:
                 guide_lines = [f"    {k}: ~{v}자" for k, v in char_guide.items()]
                 req_text += "\n  글자 수 가이드 (참고, 의미 우선):\n" + "\n".join(guide_lines)
@@ -188,7 +195,7 @@ async def _call_kei_editor(prompt: str) -> str | None:
                     logger.warning(f"Kei API (editor) HTTP {response.status_code}")
                     return None
 
-                full_text = await _stream_sse_tokens(response)
+                full_text = await stream_sse_tokens(response)
 
             if full_text:
                 return full_text
@@ -201,65 +208,6 @@ async def _call_kei_editor(prompt: str) -> str | None:
         return None
 
 
-async def _stream_sse_tokens(response: httpx.Response) -> str:
-    """SSE 스트리밍 응답에서 토큰을 실시간 수집한다."""
-    tokens: list[str] = []
-    event_type = ""
-
-    async for line in response.aiter_lines():
-        line = line.strip()
-        if not line:
-            event_type = ""
-            continue
-        if line.startswith("event:"):
-            event_type = line[6:].strip()
-        elif line.startswith("data:"):
-            data = line[5:].strip()
-            if event_type == "token" and data:
-                try:
-                    token = json.loads(data)
-                    if isinstance(token, str):
-                        tokens.append(token)
-                except json.JSONDecodeError:
-                    tokens.append(data)
-            elif event_type == "done":
-                break
-            elif event_type == "error":
-                logger.warning(f"Kei API SSE 에러: {data}")
-                break
-
-    return "".join(tokens)
-
-
-def _extract_sse_text(raw: str) -> str:
-    """SSE 응답에서 토큰 텍스트를 수집한다."""
-    import re as _re
-    tokens = []
-    events = _re.split(r'\r?\n\r?\n', raw)
-    for event in events:
-        if not event.strip():
-            continue
-        event_type = ""
-        event_data = ""
-        for line in event.split('\n'):
-            line = line.strip('\r')
-            if line.startswith('event:'):
-                event_type = line[6:].strip()
-            elif line.startswith('data:'):
-                event_data = line[5:].strip()
-        if not event_data:
-            continue
-        if event_type == 'token':
-            try:
-                token = json.loads(event_data)
-                if isinstance(token, str):
-                    tokens.append(token)
-            except json.JSONDecodeError:
-                tokens.append(event_data)
-        elif event_type == 'done':
-            break
-    return "".join(tokens)
-
 
 def _apply_defaults(blocks: list[dict[str, Any]]) -> None:
     """실패 시 기본 데이터 적용."""

src/design_director.py

@@ -16,6 +16,7 @@ import httpx
 import yaml
 
 from src.config import settings
+from src.sse_utils import stream_sse_tokens
 
 logger = logging.getLogger(__name__)
 
@@ -24,49 +25,285 @@ logger = logging.getLogger(__name__)
 # ──────────────────────────────────────
 BLOCK_SLOTS = {
     # headers/ (5개)
-    "section-title-with-bg": {"required": ["title_ko"], "optional": ["title_en", "breadcrumb", "bg_image"]},
-    "section-header-bar": {"required": ["title"], "optional": ["subtitle"]},
-    "topic-left-right": {"required": ["title", "description"], "optional": []},
-    "topic-center": {"required": ["title"], "optional": ["subtitle", "description"]},
-    "topic-numbered": {"required": ["number", "title"], "optional": ["description", "color"]},
-    # cards/ (10개)
-    "card-image-3col": {"required": ["cards"], "optional": []},
-    "card-dark-overlay": {"required": ["cards"], "optional": []},
-    "card-tag-image": {"required": ["cards"], "optional": []},
-    "card-icon-desc": {"required": ["cards"], "optional": []},
-    "card-compare-3col": {"required": ["cards"], "optional": []},
-    "card-step-vertical": {"required": ["steps"], "optional": []},
-    "card-image-round": {"required": ["cards"], "optional": []},
-    "card-stat-number": {"required": ["stats"], "optional": []},
-    "card-numbered": {"required": ["items"], "optional": []},
+    "section-title-with-bg": {
+        "required": ["title_ko"], "optional": ["title_en", "breadcrumb", "bg_image"],
+        "slot_desc": {
+            "title_ko": "한글 메인 타이틀",
+            "title_en": "영문 서브 타이틀 (없으면 생략)",
+            "breadcrumb": "상위 카테고리 경로 (예: 디지털전환 > BIM)",
+            "bg_image": "배경 이미지 경로",
+        },
+    },
+    "section-header-bar": {
+        "required": ["title"], "optional": ["subtitle"],
+        "slot_desc": {
+            "title": "섹션 제목 (짧고 굵게)",
+            "subtitle": "보조 설명 (한 줄)",
+        },
+    },
+    "topic-left-right": {
+        "required": ["title", "description"], "optional": [],
+        "slot_desc": {
+            "title": "꼭지 제목 (좌측, 굵게)",
+            "description": "꼭지 설명 (우측, 2~3줄)",
+        },
+    },
+    "topic-center": {
+        "required": ["title"], "optional": ["subtitle", "description"],
+        "slot_desc": {
+            "title": "중앙 정렬 대제목",
+            "subtitle": "부제목 (작은 글씨)",
+            "description": "추가 설명 (1~2줄)",
+        },
+    },
+    "topic-numbered": {
+        "required": ["number", "title"], "optional": ["description", "color"],
+        "slot_desc": {
+            "number": "순번 (1, 2, 3 등)",
+            "title": "단계/항목 제목",
+            "description": "설명 텍스트",
+            "color": "원형 번호 색상 (CSS 색상값)",
+        },
+    },
+    # cards/ (9개)
+    "card-image-3col": {
+        "required": ["cards"], "optional": [],
+        "slot_desc": {
+            "cards": "카드 배열. 각 카드: {image: '이미지 경로', title: '제목', title_en: '영문', bullets: ['항목1', '항목2']}. 3장.",
+        },
+    },
+    "card-dark-overlay": {
+        "required": ["cards"], "optional": [],
+        "slot_desc": {
+            "cards": "카드 배열. 각 카드: {image: '배경 이미지', title: '키워드', description: '짧은 설명'}. 3~5장.",
+        },
+    },
+    "card-tag-image": {
+        "required": ["cards"], "optional": [],
+        "slot_desc": {
+            "cards": "카드 배열. 각 카드: {tag: '카테고리 라벨', tag_color: '색상', image: '이미지', title: '제목', description: '설명'}. 3장.",
+        },
+    },
+    "card-icon-desc": {
+        "required": ["cards"], "optional": [],
+        "slot_desc": {
+            "cards": "카드 배열. 각 카드: {icon: '이모지', title: '제목', description: '설명 (2~3줄)'}. 2~4장.",
+        },
+    },
+    "card-compare-3col": {
+        "required": ["cards"], "optional": [],
+        "slot_desc": {
+            "cards": "비교 카드 배열. 각 카드: {header: '카테고리명', header_color: '색상', bullets: ['항목1', '항목2']}. 3장.",
+        },
+    },
+    "card-step-vertical": {
+        "required": ["steps"], "optional": [],
+        "slot_desc": {
+            "steps": "단계 배열. 각 단계: {number: '01', title: '단계명', description: '설명', image: '이미지(선택)'}. 3~5단계.",
+        },
+    },
+    "card-image-round": {
+        "required": ["cards"], "optional": [],
+        "slot_desc": {
+            "cards": "카드 배열. 각 카드: {image: '원형 이미지', title: '제목', description: '설명'}. 2~3장.",
+        },
+    },
+    "card-stat-number": {
+        "required": ["stats"], "optional": [],
+        "slot_desc": {
+            "stats": "통계 배열. 각 항목: {number: '85', unit: '%', label: '비용 절감율'}. 2~4개. 숫자는 출처 있는 것만!",
+        },
+    },
+    "card-numbered": {
+        "required": ["items"], "optional": [],
+        "slot_desc": {
+            "items": "항목 배열. 각 항목: {title: '항목 제목', description: '설명'}. 3~5개.",
+        },
+    },
     # tables/ (3개)
-    "compare-3col-badge": {"required": ["headers", "rows"], "optional": []},
-    "compare-2col-split": {"required": ["left_title", "right_title", "rows"], "optional": []},
-    "table-simple-striped": {"required": ["headers", "rows"], "optional": []},
-    # visuals/ (10개)
-    "venn-diagram": {"required": ["center_label", "items"], "optional": ["center_sub", "description"]},
-    "circle-gradient": {"required": ["label"], "optional": ["sub_label"]},
-    "compare-pill-pair": {"required": ["left_label", "right_label"], "optional": ["left_sub", "right_sub"]},
-    "process-horizontal": {"required": ["steps"], "optional": []},
-    "flow-arrow-horizontal": {"required": ["steps"], "optional": []},
-    "keyword-circle-row": {"required": ["keywords"], "optional": []},
-    # emphasis/ (12개)
-    "quote-big-mark": {"required": ["quote_text"], "optional": ["source"]},
-    "quote-question": {"required": ["question"], "optional": ["description"]},
-    "comparison-2col": {"required": ["left_title", "left_content", "right_title", "right_content"], "optional": ["left_subtitle", "right_subtitle"]},
-    "banner-gradient": {"required": ["text"], "optional": ["sub_text"]},
-    "dark-bullet-list": {"required": ["bullets"], "optional": ["title"]},
-    "highlight-strip": {"required": ["segments"], "optional": []},
-    "callout-solution": {"required": ["title", "description"], "optional": ["icon", "source"]},
-    "callout-warning": {"required": ["title", "description"], "optional": ["icon"]},
-    "tab-label-row": {"required": ["tabs"], "optional": []},
-    "divider-text": {"required": ["text"], "optional": []},
+    "compare-3col-badge": {
+        "required": ["headers", "rows"], "optional": [],
+        "slot_desc": {
+            "headers": "3개 열 헤더 배열: ['항목', 'A 대상', 'B 대상']",
+            "rows": "비교 행 배열. 각 행: {criteria: '비교 기준', left: 'A 내용', right: 'B 내용'}. 최소 3행.",
+        },
+    },
+    "compare-2col-split": {
+        "required": ["left_title", "right_title", "rows"], "optional": [],
+        "slot_desc": {
+            "left_title": "왼쪽 열 헤더",
+            "right_title": "오른쪽 열 헤더",
+            "rows": "비교 행 배열. 각 행: {criteria: '비교 기준', left: '왼쪽 내용', right: '오른쪽 내용'}. 최소 3행.",
+        },
+    },
+    "table-simple-striped": {
+        "required": ["headers", "rows"], "optional": [],
+        "slot_desc": {
+            "headers": "열 헤더 배열: ['열1', '열2', '열3']",
+            "rows": "데이터 행 배열. 각 행: ['셀1', '셀2', '셀3']. 행 수 자유.",
+        },
+    },
+    # visuals/ (6개)
+    "venn-diagram": {
+        "required": ["center_label", "items"], "optional": ["center_sub", "description"],
+        "slot_desc": {
+            "center_label": "중앙 교집합 라벨 (핵심 키워드)",
+            "items": "원 배열. 각 원: {label: '영역명', sub: '설명'}. 2~5개.",
+            "center_sub": "중앙 부가 설명",
+            "description": "다이어그램 하단 설명",
+        },
+    },
+    "circle-gradient": {
+        "required": ["label"], "optional": ["sub_label"],
+        "slot_desc": {
+            "label": "원 중앙 메인 텍스트 (키워드, 1~2단어)",
+            "sub_label": "원 아래 보조 텍스트",
+        },
+    },
+    "compare-pill-pair": {
+        "required": ["left_label", "right_label"], "optional": ["left_sub", "right_sub"],
+        "slot_desc": {
+            "left_label": "왼쪽 개념명 (1~2단어)",
+            "right_label": "오른쪽 개념명 (1~2단어)",
+            "left_sub": "왼쪽 보조 설명",
+            "right_sub": "오른쪽 보조 설명",
+        },
+    },
+    "process-horizontal": {
+        "required": ["steps"], "optional": [],
+        "slot_desc": {
+            "steps": "단계 배열. 각 단계: {number: '01', title: '단계명', description: '설명'}. 3~5단계.",
+        },
+    },
+    "flow-arrow-horizontal": {
+        "required": ["steps"], "optional": [],
+        "slot_desc": {
+            "steps": "흐름 배열. 각 항목: {label: '단계명'}. 3~5개. 화살표로 연결됨.",
+        },
+    },
+    "keyword-circle-row": {
+        "required": ["keywords"], "optional": [],
+        "slot_desc": {
+            "keywords": "키워드 배열. 각 항목: {letter: '약어 (G)', label: '풀네임', description: '설명'}. 3~5개.",
+        },
+    },
+    # emphasis/ (10개)
+    "quote-big-mark": {
+        "required": ["quote_text"], "optional": ["source"],
+        "slot_desc": {
+            "quote_text": "인용할 본문 텍스트 (핵심 발언, 1~3문장)",
+            "source": "출처 (예: 국토교통부, 2024). 꼭지 제목이 아님!",
+        },
+    },
+    "quote-question": {
+        "required": ["question"], "optional": ["description"],
+        "slot_desc": {
+            "question": "독자에게 던지는 질문 (1문장, 물음표로 끝)",
+            "description": "질문에 대한 부연 (1~2줄)",
+        },
+    },
+    "comparison-2col": {
+        "required": ["left_title", "left_content", "right_title", "right_content"],
+        "optional": ["left_subtitle", "right_subtitle"],
+        "slot_desc": {
+            "left_title": "왼쪽 개념 제목 (파란색)",
+            "left_content": "왼쪽 본문 (불릿 또는 문장)",
+            "right_title": "오른쪽 개념 제목 (빨간색)",
+            "right_content": "오른쪽 본문 (불릿 또는 문장)",
+            "left_subtitle": "왼쪽 보조 제목",
+            "right_subtitle": "오른쪽 보조 제목",
+        },
+    },
+    "banner-gradient": {
+        "required": ["text"], "optional": ["sub_text"],
+        "slot_desc": {
+            "text": "핵심 결론 한 줄 (굵은 대형 텍스트. 가장 중요한 메시지)",
+            "sub_text": "부연 설명 (작은 보조 텍스트. text보다 덜 중요)",
+        },
+    },
+    "dark-bullet-list": {
+        "required": ["bullets"], "optional": ["title"],
+        "slot_desc": {
+            "title": "리스트 상단 제목 (파란색, 선택)",
+            "bullets": "불릿 항목 배열: ['핵심 포인트 1', '핵심 포인트 2']. 3~5개.",
+        },
+    },
+    "highlight-strip": {
+        "required": ["segments"], "optional": [],
+        "slot_desc": {
+            "segments": "색상 구간 배열. 각 구간: {label: '카테고리명', color: '색상'}. 3~5개.",
+        },
+    },
+    "callout-solution": {
+        "required": ["title", "description"], "optional": ["icon", "source"],
+        "slot_desc": {
+            "title": "솔루션/방향성 제목",
+            "description": "상세 설명 (2~3줄)",
+            "icon": "아이콘 이모지 (예: 💡)",
+            "source": "출처 (있으면)",
+        },
+    },
+    "callout-warning": {
+        "required": ["title", "description"], "optional": ["icon"],
+        "slot_desc": {
+            "title": "문제점/경고 제목",
+            "description": "상세 설명 (2~3줄)",
+            "icon": "아이콘 이모지 (예: ⚠️)",
+        },
+    },
+    "tab-label-row": {
+        "required": ["tabs"], "optional": [],
+        "slot_desc": {
+            "tabs": "탭 배열. 각 탭: {label: '탭 이름', active: true/false}. 3~5개. 하나만 active.",
+        },
+    },
+    "divider-text": {
+        "required": ["text"], "optional": [],
+        "slot_desc": {
+            "text": "구분선 중앙 텍스트 (짧은 전환 문구, 1~5단어)",
+        },
+    },
     # media/ (5개)
-    "image-row-2col": {"required": ["images"], "optional": []},
-    "image-grid-2x2": {"required": ["images"], "optional": []},
-    "image-side-text": {"required": ["image_src"], "optional": ["image_alt", "title", "description", "bullets"]},
-    "image-full-caption": {"required": ["src"], "optional": ["alt", "caption"]},
-    "image-before-after": {"required": ["before_src", "after_src"], "optional": ["before_label", "after_label", "caption"]},
+    "image-row-2col": {
+        "required": ["images"], "optional": [],
+        "slot_desc": {
+            "images": "이미지 배열. 각 항목: {src: '이미지 경로', alt: '설명', caption: '캡션'}. 2장.",
+        },
+    },
+    "image-grid-2x2": {
+        "required": ["images"], "optional": [],
+        "slot_desc": {
+            "images": "이미지 배열. 각 항목: {src: '이미지 경로', alt: '설명'}. 4장 (2x2).",
+        },
+    },
+    "image-side-text": {
+        "required": ["image_src"], "optional": ["image_alt", "title", "description", "bullets"],
+        "slot_desc": {
+            "image_src": "좌측 이미지 경로",
+            "image_alt": "이미지 대체 텍스트",
+            "title": "우측 제목",
+            "description": "우측 설명 텍스트",
+            "bullets": "우측 불릿 항목 배열: ['항목1', '항목2']",
+        },
+    },
+    "image-full-caption": {
+        "required": ["src"], "optional": ["alt", "caption"],
+        "slot_desc": {
+            "src": "전체 너비 이미지 경로",
+            "alt": "이미지 대체 텍스트",
+            "caption": "이미지 하단 캡션",
+        },
+    },
+    "image-before-after": {
+        "required": ["before_src", "after_src"], "optional": ["before_label", "after_label", "caption"],
+        "slot_desc": {
+            "before_src": "Before 이미지 경로",
+            "after_src": "After 이미지 경로",
+            "before_label": "Before 라벨 (기본: Before)",
+            "after_label": "After 라벨 (기본: After)",
+            "caption": "비교 설명 캡션",
+        },
+    },
 }
 
 # ──────────────────────────────────────
@@ -264,11 +501,11 @@ header/footer는 고정이므로 건드리지 않는다.
 ## purpose 기반 블록 선택 가이드 (참고, 강제 아님)
 각 꼭지의 purpose에 맞는 블록 계열을 선택하라:
 - 문제제기 → callout-warning, quote-big-mark, quote-question
-- 근거사례 → quote-left-border (출처 포함), card-text-grid (항목 나열)
+- 근거사례 → quote-big-mark (출처 포함), card-icon-desc (항목 나열)
 - 핵심전달 → comparison-2col, compare-pill-pair, compare-2col-split
-- 용어정의 → card-text-grid (정의+출처), card-numbered (순서 있으면)
+- 용어정의 → card-icon-desc (정의+출처), card-numbered (순서 있으면)
 - 결론강조 → banner-gradient (footer)
-- 구조시각화 → venn-diagram, layer-diagram (단독 배치)
+- 구조시각화 → venn-diagram (단독 배치)
 
 ## 허용된 블록 id 목록 (이 목록에 없는 블록은 절대 선택하지 마라)
 {allowed_ids}
@@ -364,7 +601,7 @@ async def _opus_block_recommendation(
                     logger.warning(f"[Step A-2] Kei API HTTP {response.status_code}")
                     return None
 
-                full_text = await _stream_sse_tokens(response)
+                full_text = await stream_sse_tokens(response)
 
             if not full_text:
                 logger.warning("[Step A-2] Kei API 응답 텍스트 없음")
@@ -386,35 +623,6 @@ async def _opus_block_recommendation(
         return None
 
 
-async def _stream_sse_tokens(response: httpx.Response) -> str:
-    """SSE 스트리밍 응답에서 토큰을 실시간 수집한다."""
-    tokens: list[str] = []
-    event_type = ""
-
-    async for line in response.aiter_lines():
-        line = line.strip()
-        if not line:
-            event_type = ""
-            continue
-        if line.startswith("event:"):
-            event_type = line[6:].strip()
-        elif line.startswith("data:"):
-            data = line[5:].strip()
-            if event_type == "token" and data:
-                try:
-                    token = json.loads(data)
-                    if isinstance(token, str):
-                        tokens.append(token)
-                except json.JSONDecodeError:
-                    tokens.append(data)
-            elif event_type == "done":
-                break
-            elif event_type == "error":
-                logger.warning(f"Kei API SSE 에러: {data}")
-                break
-
-    return "".join(tokens)
-
 
 async def create_layout_concept(
     content: str,
@@ -567,11 +775,13 @@ async def create_layout_concept(
             for block in blocks:
                 block_type = block.get("type", "")
                 if block_type and block_type not in registered_ids:
+                    purpose = block.get("purpose", "")
+                    fallback = PURPOSE_FALLBACK.get(purpose, "callout-solution")
                     logger.warning(
                         f"[Step B 검증] 미등록 블록 '{block_type}' 거부 → "
-                        f"'callout-solution'으로 교체"
+                        f"'{fallback}'으로 교체 (purpose={purpose})"
                     )
-                    block["type"] = "callout-solution"
+                    block["type"] = fallback
 
             # area명 검증: 프리셋 zone에 없으면 기본 zone으로 매핑
             valid_zones = {z for z in preset["zones"] if z != "header"}
@@ -597,13 +807,14 @@ async def create_layout_concept(
                         )
                         block["area"] = "footer"
 
-            # 5번: zone별 height_cost 합산 검증 — 초과 시 큰 블록 교체
-            _validate_height_budget(blocks, preset)
+            # 5번: zone별 height_cost 합산 검증 (I-9: overflow 수집, 블록 교체 안 함)
+            overflows = _validate_height_budget(blocks, preset)
 
             logger.info(
                 f"[Step B] 블록 매핑 완료: {preset_name}, {len(blocks)}개 블록"
+                + (f", overflow {len(overflows)}건" if overflows else "")
             )
-            return {
+            result = {
                 "title": analysis.get("title", "슬라이드"),
                 "pages": [{
                     "grid_areas": preset["grid_areas"],
@@ -612,6 +823,9 @@ async def create_layout_concept(
                     "blocks": blocks,
                 }],
             }
+            if overflows:
+                result["overflow"] = overflows
+            return result
         else:
             logger.warning("블록 매핑 JSON 파싱 실패. fallback.")
 
@@ -670,6 +884,16 @@ HEIGHT_COST_PX = {
     "xlarge": 400,
 }
 
+# 미등록 블록 거부 시 purpose 기반 대체 (I-3)
+PURPOSE_FALLBACK = {
+    "문제제기": "callout-warning",
+    "근거사례": "quote-big-mark",
+    "핵심전달": "comparison-2col",
+    "용어정의": "card-icon-desc",
+    "결론강조": "banner-gradient",
+    "구조시각화": "card-icon-desc",
+}
+
 # body/sidebar/footer zone에서 사용 금지인 블록 → 교체
 BODY_FORBIDDEN_MAP = {
     "section-title-with-bg": "topic-center",  # 500px 블록 → compact 헤더로
@@ -708,11 +932,16 @@ def _load_catalog_map_for_height() -> dict[str, str]:
         return {}
 
 
-def _validate_height_budget(blocks: list[dict], preset: dict) -> None:
-    """zone별 height_cost 합산을 검증하고, 초과 시 큰 블록을 교체한다.
+def _validate_height_budget(blocks: list[dict], preset: dict) -> list[dict]:
+    """zone별 height_cost 합산을 검증한다. (I-9 개정)
 
-    또한 body/sidebar/footer zone에서 금지된 블록을 교체한다.
-    코드 레벨 검증 — Sonnet이 높이 예산을 안 지켜도 강제 교정.
+    금지 블록 교체, pill-pair 단독 검증은 수행하되,
+    높이 초과 시 블록을 자동 교체하지 않는다.
+    대신 overflow 정보를 수집하여 반환 → pipeline에서 Kei에게 판단 요청.
+    DOWNGRADE_MAP은 Kei API 실패 시 비상용으로만 사용.
+
+    Returns:
+        overflow 정보 리스트. 초과 없으면 빈 리스트.
     """
     zones = preset.get("zones", {})
     gap_px = 20  # --spacing-block
@@ -736,11 +965,24 @@ def _validate_height_budget(blocks: list[dict], preset: dict) -> None:
             )
             block["type"] = replacement
 
+    # compare-pill-pair 단독 사용 금지 (I-7)
+    COMPARISON_BLOCKS = {"compare-2col-split", "compare-3col-badge", "comparison-2col"}
+    for area, area_blocks in zone_blocks.items():
+        types = {b.get("type") for b in area_blocks}
+        if "compare-pill-pair" in types and not types & COMPARISON_BLOCKS:
+            for block in area_blocks:
+                if block.get("type") == "compare-pill-pair":
+                    block["type"] = "comparison-2col"
+                    logger.warning(
+                        "[pill-pair 단독 금지] compare-pill-pair → comparison-2col"
+                    )
+
+    # 높이 예산 검증 — 초과 시 overflow 정보 수집 (블록 교체 안 함)
+    overflows: list[dict] = []
     for area, area_blocks in zone_blocks.items():
         zone_info = zones.get(area, {})
         budget = zone_info.get("budget_px", 490)
 
-        # 총 높이 계산
         total = sum(_get_block_height(b.get("type", "")) for b in area_blocks)
         total += gap_px * max(0, len(area_blocks) - 1)
 
@@ -752,8 +994,39 @@ def _validate_height_budget(blocks: list[dict], preset: dict) -> None:
             f"블록: {[b.get('type') for b in area_blocks]}"
         )
 
-        # 가장 큰 블록부터 교체 시도
-        area_blocks.sort(key=lambda b: _get_block_height(b.get("type", "")), reverse=True)
+        overflows.append({
+            "area": area,
+            "overflow_px": total - budget,
+            "budget_px": budget,
+            "total_px": total,
+            "blocks": [
+                {
+                    "type": b.get("type", ""),
+                    "purpose": b.get("purpose", ""),
+                    "topic_id": b.get("topic_id"),
+                    "height_px": _get_block_height(b.get("type", "")),
+                }
+                for b in area_blocks
+            ],
+        })
+
+    return overflows
+
+
+def _downgrade_fallback(blocks: list[dict], overflows: list[dict]) -> None:
+    """Kei API 실패 시 비상용 기계적 블록 교체.
+
+    기존 DOWNGRADE_MAP 로직. 정상 경로가 아닌 비상 경로.
+    """
+    for overflow in overflows:
+        area = overflow["area"]
+        area_blocks = [b for b in blocks if b.get("area") == area]
+        area_blocks.sort(
+            key=lambda b: _get_block_height(b.get("type", "")), reverse=True
+        )
+
+        total = overflow["total_px"]
+        budget = overflow["budget_px"]
 
         for block in area_blocks:
             block_type = block.get("type", "")
@@ -767,8 +1040,8 @@ def _validate_height_budget(blocks: list[dict], preset: dict) -> None:
                 block["type"] = replacement
                 total = total - old_height + new_height
 
-                logger.info(
-                    f"[높이 교체] {block_type}({old_height}px) → "
+                logger.warning(
+                    f"[DOWNGRADE 비상] {block_type}({old_height}px) → "
                     f"{replacement}({new_height}px). 잔여: {total}px/{budget}px"
                 )
 

src/kei_client.py

@@ -10,10 +10,10 @@ import logging
 import re
 from typing import Any
 
-import anthropic
 import httpx
 
 from src.config import settings
+from src.sse_utils import stream_sse_tokens
 
 logger = logging.getLogger(__name__)
 
@@ -167,7 +167,7 @@ async def refine_concepts(
                     logger.warning(f"[1단계-B] Kei API HTTP {response.status_code}")
                     return analysis
 
-                full_text = await _stream_sse_tokens(response)
+                full_text = await stream_sse_tokens(response)
 
         if not full_text:
             logger.warning("[1단계-B] 응답 텍스트 없음. 1단계-A 결과 유지.")
@@ -214,7 +214,7 @@ async def _call_kei_api(content: str) -> dict[str, Any] | None:
                     logger.warning(f"Kei API HTTP {response.status_code}")
                     return None
 
-                full_text = await _stream_sse_tokens(response)
+                full_text = await stream_sse_tokens(response)
 
             if not full_text:
                 logger.warning("Kei API 응답에서 텍스트 추출 실패")
@@ -232,128 +232,105 @@ async def _call_kei_api(content: str) -> dict[str, Any] | None:
         return None
 
 
-async def _stream_sse_tokens(response: httpx.Response) -> str:
-    """SSE 스트리밍 응답에서 토큰을 실시간 수집한다.
 
-    persona_agent의 SSE 이벤트:
-    - token: 텍스트 토큰 수집
-    - done: 완료, 중단
-    - error: 에러, 즉시 중단
-    - planning/planning_done/research_progress/warning: 스킵
+
+# ──────────────────────────────────────
+# I-9: Kei 넘침 판단 호출
+# ──────────────────────────────────────
+
+KEI_OVERFLOW_PROMPT = """당신은 슬라이드 콘텐츠 전문가이다.
+디자인 팀장이 배치한 블록들이 컨테이너(zone)의 높이 예산을 초과했다.
+콘텐츠의 중요도와 전달 메시지를 기준으로 어떻게 처리할지 판단하라.
+
+## 판단 기준
+- 텍스트 분량만 줄이면 현재 블록 구조 안에서 해결되는가? → "trim"
+- 콘텐츠 자체가 컨테이너에 담기엔 본질적으로 많은가? → "restructure"
+- 중요도가 높은 콘텐츠를 무리하게 축소하면 안 된다
+- 부가/상세 정보는 팝업(detail page)으로 분리할 수 있다
+
+## 출력 (JSON만. 설명 없이.)
+
+Option 1 (텍스트 축약):
+{"decision": "trim", "trim_targets": [{"topic_id": 1, "max_chars": 200, "reason": "부연 설명 축약 가능"}]}
+
+Option 2 (핵심 재구성 + 팝업 분리):
+{"decision": "restructure", "core_topics": [1, 2], "detail_topics": [3], "reason": "12행 비교표는 팝업으로 분리"}
+"""
+
+
+async def call_kei_overflow_judgment(
+    overflows: list[dict],
+    content: str,
+    analysis: dict[str, Any],
+) -> dict[str, Any] | None:
+    """Kei API에 넘침 상황을 전달하고 판단을 받는다.
+
+    반드시 Kei API 경유. Anthropic 직접 호출 절대 금지.
+    fallback: None 반환 → pipeline에서 DOWNGRADE 비상 작동.
     """
-    tokens: list[str] = []
-    event_type = ""
+    kei_url = getattr(settings, "kei_api_url", "http://localhost:8000")
 
-    async for line in response.aiter_lines():
-        line = line.strip()
-        if not line:
-            event_type = ""
-            continue
-        if line.startswith("event:"):
-            event_type = line[6:].strip()
-        elif line.startswith("data:"):
-            data = line[5:].strip()
-            if event_type == "token" and data:
-                try:
-                    token = json.loads(data)
-                    if isinstance(token, str):
-                        tokens.append(token)
-                except json.JSONDecodeError:
-                    tokens.append(data)
-            elif event_type == "done":
-                break
-            elif event_type == "error":
-                logger.warning(f"Kei API SSE 에러: {data}")
-                break
+    overflow_desc = json.dumps(overflows, ensure_ascii=False, indent=2)
+    topics_desc = json.dumps(
+        [
+            {
+                "id": t.get("id"),
+                "title": t.get("title", ""),
+                "purpose": t.get("purpose", ""),
+                "summary": t.get("summary", "")[:100],
+            }
+            for t in analysis.get("topics", [])
+        ],
+        ensure_ascii=False,
+    )
 
-    return "".join(tokens)
+    # I-8: 대형 콘텐츠 정보 포함
+    extra_info = ""
+    tables = analysis.get("tables", [])
+    if tables:
+        extra_info += f"\n\n## 테이블 정보\n{json.dumps(tables, ensure_ascii=False)}"
+    images = analysis.get("images", [])
+    if images:
+        extra_info += f"\n\n## 이미지 정보\n{json.dumps(images, ensure_ascii=False)}"
 
-
-def _extract_sse_text(raw: str) -> str:
-    """SSE 응답에서 토큰 텍스트를 수집한다. CRLF/LF 모두 처리."""
-    tokens = []
-    # CRLF 또는 LF로 이벤트 분리
-    events = re.split(r'\r?\n\r?\n', raw)
-
-    for event in events:
-        if not event.strip():
-            continue
-
-        event_type = ""
-        event_data = ""
-
-        for line in event.split('\n'):
-            line = line.strip('\r')
-            if line.startswith('event:'):
-                event_type = line[6:].strip()
-            elif line.startswith('data:'):
-                event_data = line[5:].strip()
-
-        if not event_data:
-            continue
-
-        if event_type == 'token':
-            try:
-                token = json.loads(event_data)
-                if isinstance(token, str):
-                    tokens.append(token)
-            except json.JSONDecodeError:
-                tokens.append(event_data)
-        elif event_type == 'done':
-            break
-
-    return "".join(tokens)
-
-
-async def _call_anthropic_direct(content: str) -> dict[str, Any] | None:
-    """Anthropic API 직접 호출 (Kei API fallback)."""
-    if not settings.anthropic_api_key:
-        return None
-
-    system_prompt = (
-        "당신은 콘텐츠를 분석하여 슬라이드 구조를 설계하는 실장이다.\n\n"
-        "## 핵심 원칙\n"
-        "- 원본의 논리 흐름과 정보를 빠뜨리지 마라\n"
-        "- 원본에 있는 내용을 임의로 제거하거나 다른 의미로 바꾸지 마라\n"
-        "- 슬라이드에 맞게 정리하되, 원본이 말하려는 흐름은 유지\n\n"
-        "## 꼭지 추출 규칙\n"
-        "- 본문에서 2~5개의 핵심 꼭지를 추출한다\n"
-        "- 참조 정보는 role: 'reference', 본문 흐름은 role: 'flow'로 표시\n"
-        "- 1페이지 적정 꼭지 수: 5개\n"
-        "- 초과 시 2페이지 분리\n"
-        "- 이미지가 있으면 images[]에, 표가 있으면 tables[]에 판단 기록\n\n"
-        "## 출력 형식 (JSON만. 설명 없이.)\n"
-        '{"title": "제목", "total_pages": 1, '
-        '"info_structure": "정보 구조 설명", '
-        '"topics": ['
-        '{"id": 1, "title": "꼭지 제목", "summary": "요약", '
-        '"layer": "intro|core|supporting|conclusion", '
-        '"role": "flow|reference", '
-        '"emphasis": true, "direction": "vertical|horizontal|flexible", '
-        '"content_type": "text|image|table|mixed", '
-        '"detail_target": false, "page": 1}], '
-        '"images": [], "tables": []}'
+    prompt = (
+        KEI_OVERFLOW_PROMPT + "\n\n"
+        f"## 넘침 현황\n{overflow_desc}\n\n"
+        f"## 꼭지 목록\n{topics_desc}"
+        f"{extra_info}\n\n"
+        f"## 원본 콘텐츠 요약\n{content[:2000]}"
     )
 
     try:
-        client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
-        response = await client.messages.create(
-            model="claude-sonnet-4-20250514",
-            max_tokens=2048,
-            system=system_prompt,
-            messages=[{"role": "user", "content": f"다음 콘텐츠의 꼭지를 추출해줘:\n\n{content}"}],
-        )
+        async with httpx.AsyncClient(timeout=None) as client:
+            async with client.stream(
+                "POST",
+                f"{kei_url}/api/message",
+                json={
+                    "message": prompt,
+                    "session_id": "design-agent-overflow",
+                    "mode_hint": "chat",
+                },
+                timeout=None,
+            ) as response:
+                if response.status_code != 200:
+                    logger.warning(f"Kei API (overflow) HTTP {response.status_code}")
+                    return None
+                full_text = await stream_sse_tokens(response)
 
-        result_text = response.content[0].text
-        result = _parse_json(result_text)
-
-        if result and "topics" in result:
-            return result
+        if full_text:
+            result = _parse_json(full_text)
+            if result and "decision" in result:
+                logger.info(f"[Kei 넘침 판단] decision={result['decision']}")
+                return result
+            logger.warning("[Kei 넘침 판단] JSON 파싱 실패 또는 decision 없음")
+            return None
 
+        logger.warning("Kei API (overflow) 텍스트 추출 실패")
         return None
 
     except Exception as e:
-        logger.warning(f"Anthropic 직접 호출 실패: {e}")
+        logger.warning(f"Kei API (overflow) 호출 실패: {e}")
         return None
 
 

src/pipeline.py

@@ -15,8 +15,8 @@ from typing import Any, AsyncIterator
 
 import anthropic
 
-from src.kei_client import classify_content, manual_classify, refine_concepts
-from src.design_director import create_layout_concept, LAYOUT_PRESETS, select_preset
+from src.kei_client import classify_content, manual_classify, refine_concepts, call_kei_overflow_judgment
+from src.design_director import create_layout_concept, LAYOUT_PRESETS, select_preset, _downgrade_fallback
 from src.content_editor import fill_content
 from src.renderer import render_slide
 from src.image_utils import get_image_sizes, embed_images
@@ -55,6 +55,20 @@ async def generate_slide(
         analysis = await refine_concepts(content, analysis)
         logger.info("1단계-B 완료: 컨셉 구체화")
 
+        # I-6: 슬라이드 제목 ↔ 첫 꼭지 제목 중복 검증
+        from difflib import SequenceMatcher
+        title = analysis.get("title", "")
+        topics = analysis.get("topics", [])
+        if topics:
+            first_title = topics[0].get("title", "")
+            similarity = SequenceMatcher(None, title, first_title).ratio()
+            if similarity > 0.7:
+                purpose = topics[0].get("purpose", "문제제기")
+                topics[0]["title"] = f"{purpose}: {topics[0].get('summary', '')[:30]}"
+                logger.warning(
+                    f"[제목 중복 교정] 유사도 {similarity:.0%} → 첫 꼭지 제목 변경"
+                )
+
         # 이미지 크기 측정 (base_path 있을 때만)
         image_sizes = get_image_sizes(content, base_path)
         if image_sizes:
@@ -92,7 +106,9 @@ async def generate_slide(
         yield {"event": "progress", "data": "5/5 디자인 팀장이 전체 균형을 검토 중..."}
 
         for review_round in range(MAX_REVIEW_ROUNDS):
-            review_result = await _review_balance(html, layout_concept, content)
+            review_result = await _review_balance(
+                html, layout_concept, content, analysis
+            )
 
             if not review_result or not review_result.get("needs_adjustment"):
                 if review_round == 0:
@@ -107,6 +123,31 @@ async def generate_slide(
                 f"조정 필요 — {issues}"
             )
 
+            # overflow_detected가 있으면 Kei에게 판단 요청 (Sonnet은 감지만, 판단은 Kei)
+            overflow_adjs = [
+                adj for adj in review_result.get("adjustments", [])
+                if adj.get("action") == "overflow_detected"
+            ]
+            if overflow_adjs:
+                overflow_context = _build_overflow_context(
+                    layout_concept, overflow_adjs
+                )
+                kei_judgment = await call_kei_overflow_judgment(
+                    overflow_context, content, analysis
+                )
+
+                if kei_judgment is None:
+                    logger.warning("[DOWNGRADE 비상] Kei API 실패 → 기계적 교체")
+                    for page in layout_concept.get("pages", []):
+                        _downgrade_fallback(
+                            page.get("blocks", []), overflow_context
+                        )
+                else:
+                    _convert_kei_judgment(review_result, kei_judgment)
+                    logger.info(
+                        f"[Kei 넘침 판단] decision={kei_judgment.get('decision')}"
+                    )
+
             layout_concept = await _apply_adjustments(
                 layout_concept, review_result, content
             )
@@ -237,13 +278,15 @@ async def _review_balance(
     html: str,
     layout_concept: dict[str, Any],
     content: str,
+    analysis: dict[str, Any] | None = None,
 ) -> dict[str, Any] | None:
-    """5단계: 디자인 팀장이 1차 조립 결과를 재검토한다.
+    """5단계: 디자인 팀장이 조립 결과를 재검토한다.
 
-    HTML 코드 기반으로 구조적 점검:
+    HTML 코드 기반으로 구조적 점검 + 높이 넘침 감지:
     - 빈 블록 감지
     - 블록 간 채움 비율 불균형
     - 이미지/표 크기 적절성
+    - 높이 초과 감지 → overflow_detected (Kei 판단 필요)
     """
     try:
         client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
@@ -259,28 +302,62 @@ async def _review_balance(
                     f"데이터 {text_len}자"
                 )
 
+        # zone 예산 정보 (analysis에서 프리셋 추출)
+        zone_budget_text = ""
+        overflow_hint_text = ""
+        if analysis:
+            preset_name = select_preset(analysis)
+            preset = LAYOUT_PRESETS.get(preset_name, {})
+            zone_lines = [
+                f"- {name}: ~{z['budget_px']}px (너비 {z['width_pct']}%)"
+                for name, z in preset.get("zones", {}).items()
+            ]
+            zone_budget_text = (
+                "\n\n## zone별 높이 예산\n" + "\n".join(zone_lines)
+            )
+
+            # Stage 2에서 감지한 예상 overflow 힌트
+            overflow_hint = layout_concept.get("overflow", [])
+            if overflow_hint:
+                hint_lines = [
+                    f"- {o['area']}: 예상 {o['total_px']}px > 예산 {o['budget_px']}px "
+                    f"(+{o['overflow_px']}px 초과)"
+                    for o in overflow_hint
+                ]
+                overflow_hint_text = (
+                    "\n\n## 높이 초과 힌트 (2단계 예상치, 참고용)\n"
+                    + "\n".join(hint_lines)
+                )
+
         system = (
-            "당신은 디자인 팀장이다. 1차 조립 결과(HTML)를 검토하여 균형을 점검한다.\n\n"
+            "당신은 디자인 팀장이다. 조립 결과(HTML)를 검토하여 균형과 높이 제약을 점검한다.\n\n"
             "## 점검 항목\n"
             "1. 빈 블록: 데이터가 없거나 극히 적은 블록\n"
             "2. 채움 불균형: 한 블록은 빽빽하고 다른 블록은 비어있음\n"
             "3. 이미지/표: 너무 작거나 큰 것은 없는지\n"
             "4. 전체 정보량: 한 페이지에 너무 많거나 적은지\n"
-            "5. HTML 구조: 블록이 영역 안에 잘 배치되었는지\n\n"
+            "5. HTML 구조: 블록이 영역 안에 잘 배치되었는지\n"
+            "6. 높이 초과: 각 zone의 블록+텍스트가 예산을 초과하는가?\n"
+            "   - 텍스트 양/블록 수를 보고 판단\n"
+            "   - shrink로 해결 가능하면 shrink 사용\n"
+            "   - 불가능 (콘텐츠가 본질적으로 큼) → overflow_detected\n\n"
             "## 조정 action 설명\n"
-            "- expand: 텍스트를 늘린다. target_ratio로 얼마나 늘릴지 지정 (예: 1.3 = 30% 증가)\n"
-            "- shrink: 텍스트를 줄인다. target_ratio로 얼마나 줄일지 지정 (예: 0.7 = 30% 감소)\n"
-            "- rewrite: 텍스트를 완전히 재작성한다. detail에 재작성 방향 명시.\n\n"
+            "- expand: 텍스트를 늘린다. target_ratio로 지정 (예: 1.3 = 30% 증가)\n"
+            "- shrink: 텍스트를 줄인다. target_ratio로 지정 (예: 0.7 = 30% 감소)\n"
+            "- rewrite: 텍스트를 완전히 재작성한다. detail에 방향 명시.\n"
+            "- overflow_detected: 높이 초과로 콘텐츠 판단 필요. 해당 zone과 초과 블록을 detail에 명시.\n\n"
             "## 출력 형식 (JSON만)\n"
             '{"needs_adjustment": true/false, '
             '"issues": ["이슈1", "이슈2"], '
-            '"adjustments": [{"block_area": "...", "action": "expand|shrink|rewrite", '
+            '"adjustments": [{"block_area": "...", "action": "expand|shrink|rewrite|overflow_detected", '
             '"target_ratio": 1.3, "detail": "..."}]}'
         )
 
         user_prompt = (
-            f"## 1차 조립 HTML\n{html}\n\n"
+            f"## 조립 HTML\n{html}\n\n"
             f"## 블록별 데이터 양\n" + "\n".join(block_summary) +
+            zone_budget_text +
+            overflow_hint_text +
             f"\n\n## 레이아웃 구조\n"
             f"페이지 수: {len(layout_concept.get('pages', []))}\n"
             f"총 블록 수: {sum(len(p.get('blocks', [])) for p in layout_concept.get('pages', []))}\n\n"
@@ -344,11 +421,105 @@ async def _apply_adjustments(
                     block["reason"] = f"재작성: {detail}"
                     logger.info(f"조정: {area} → rewrite ({detail})")
 
+                elif action == "kei_trim":
+                    max_chars = adj.get("max_chars", 200)
+                    if "char_guide" not in block:
+                        block["char_guide"] = {}
+                    for key in block.get("char_guide", {}):
+                        block["char_guide"][key] = min(
+                            block["char_guide"][key], max_chars
+                        )
+                    if not block["char_guide"]:
+                        block["char_guide"] = {"text": max_chars}
+                    logger.info(
+                        f"조정: {area} → kei_trim max_chars={max_chars} "
+                        f"({detail})"
+                    )
+
+                elif action == "kei_restructure":
+                    block["detail_target"] = True
+                    if "data" in block:
+                        del block["data"]
+                    block["reason"] = f"재구성: {detail}"
+                    logger.info(
+                        f"조정: {area} → kei_restructure (detail_target)"
+                    )
+
     # 조정된 가이드로 재편집
     layout_concept = await fill_content(content, layout_concept)
     return layout_concept
 
 
+def _build_overflow_context(
+    layout_concept: dict[str, Any],
+    overflow_adjs: list[dict],
+) -> list[dict]:
+    """Sonnet이 감지한 overflow_detected를 Kei에게 전달할 형태로 변환한다.
+
+    실제 채워진 블록 데이터(텍스트)를 포함하여 Kei가 판단할 수 있도록 한다.
+    """
+    overflows = []
+    for adj in overflow_adjs:
+        area = adj.get("block_area", "")
+        # 해당 zone의 블록 정보 + 실제 텍스트 추출
+        area_blocks = []
+        for page in layout_concept.get("pages", []):
+            for block in page.get("blocks", []):
+                if block.get("area") == area:
+                    data = block.get("data", {})
+                    text_preview = json.dumps(data, ensure_ascii=False)[:300]
+                    area_blocks.append({
+                        "type": block.get("type", ""),
+                        "purpose": block.get("purpose", ""),
+                        "topic_id": block.get("topic_id"),
+                        "text_preview": text_preview,
+                    })
+        overflows.append({
+            "area": area,
+            "detail": adj.get("detail", ""),
+            "blocks": area_blocks,
+        })
+    return overflows
+
+
+def _convert_kei_judgment(
+    review_result: dict[str, Any],
+    kei_judgment: dict[str, Any],
+) -> None:
+    """Kei의 trim/restructure 판단을 review_result.adjustments에 반영한다.
+
+    기존 overflow_detected 항목을 kei_trim 또는 kei_restructure로 교체.
+    """
+    decision = kei_judgment.get("decision", "")
+    new_adjs = []
+
+    for adj in review_result.get("adjustments", []):
+        if adj.get("action") == "overflow_detected":
+            # overflow_detected → Kei 판단으로 교체
+            if decision == "trim":
+                for target in kei_judgment.get("trim_targets", []):
+                    new_adjs.append({
+                        "block_area": adj.get("block_area", ""),
+                        "action": "kei_trim",
+                        "max_chars": target.get("max_chars", 200),
+                        "topic_id": target.get("topic_id"),
+                        "detail": target.get("reason", ""),
+                    })
+            elif decision == "restructure":
+                for tid in kei_judgment.get("detail_topics", []):
+                    new_adjs.append({
+                        "block_area": adj.get("block_area", ""),
+                        "action": "kei_restructure",
+                        "topic_id": tid,
+                        "detail": kei_judgment.get("reason", ""),
+                    })
+        else:
+            # 기존 expand/shrink/rewrite는 그대로 유지
+            new_adjs.append(adj)
+
+    review_result["adjustments"] = new_adjs
+
+
 def _parse_json(text: str) -> dict[str, Any] | None:
     """텍스트에서 JSON을 추출한다."""
     patterns = [

src/sse_utils.py

@@ -0,0 +1,50 @@
+"""SSE 스트리밍 공통 유틸리티.
+
+persona_agent의 SSE 이벤트를 수신하여 토큰을 수집한다.
+kei_client, content_editor, design_director에서 공통 사용.
+"""
+from __future__ import annotations
+
+import json
+import logging
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+
+async def stream_sse_tokens(response: httpx.Response) -> str:
+    """SSE 스트리밍 응답에서 토큰을 실시간 수집한다.
+
+    persona_agent의 SSE 이벤트:
+    - token: 텍스트 토큰 수집
+    - done: 완료, 중단
+    - error: 에러, 즉시 중단
+    - planning/planning_done/research_progress/warning: 스킵
+    """
+    tokens: list[str] = []
+    event_type = ""
+
+    async for line in response.aiter_lines():
+        line = line.strip()
+        if not line:
+            event_type = ""
+            continue
+        if line.startswith("event:"):
+            event_type = line[6:].strip()
+        elif line.startswith("data:"):
+            data = line[5:].strip()
+            if event_type == "token" and data:
+                try:
+                    token = json.loads(data)
+                    if isinstance(token, str):
+                        tokens.append(token)
+                except json.JSONDecodeError:
+                    tokens.append(data)
+            elif event_type == "done":
+                break
+            elif event_type == "error":
+                logger.warning(f"Kei API SSE 에러: {data}")
+                break
+
+    return "".join(tokens)