From 4df6b26e1c9fa119b9cfbbce7ad9d11207586291 Mon Sep 17 00:00:00 2001
From: HWIYA <dlgnl117@gmail.com>
Date: Wed, 5 Aug 2026 14:47:42 +0900
Subject: [PATCH 1/4] =?UTF-8?q?feat:=20HF=20Intent=20=ED=95=98=EC=9D=B4?=
 =?UTF-8?q?=EB=B8=8C=EB=A6=AC=EB=93=9C=20=EB=B6=84=EB=A5=98=EA=B8=B0=20opt?=
 =?UTF-8?q?-in=20=EA=B3=A8=EA=B2=A9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

기본은 EXPIRY_RENEWAL stub 유지. FOWOCO_INTENT_MODEL_ENABLED=true 시 BERT(+선택 A.X)로 전환.
---
 app/agents/intent/__init__.py      |   6 +-
 app/agents/intent/guardrail.py     |  76 +++++++++++++++++
 app/agents/intent/hybrid.py        | 100 ++++++++++++++++++++++
 app/agents/intent/models_hf.py     | 132 +++++++++++++++++++++++++++++
 app/agents/intent/service.py       |  96 ++++++++++++++++++++-
 app/core/config.py                 |  14 +++
 pyproject.toml                     |   9 +-
 tests/agents/test_intent_hybrid.py |  53 ++++++++++++
 8 files changed, 482 insertions(+), 4 deletions(-)
 create mode 100644 app/agents/intent/guardrail.py
 create mode 100644 app/agents/intent/hybrid.py
 create mode 100644 app/agents/intent/models_hf.py
 create mode 100644 tests/agents/test_intent_hybrid.py

diff --git a/app/agents/intent/__init__.py b/app/agents/intent/__init__.py
index ed0a70f..8a67118 100644
--- a/app/agents/intent/__init__.py
+++ b/app/agents/intent/__init__.py
@@ -2,12 +2,16 @@
 
 from .service import (
     FixedExpiryRenewalIntentAgent,
+    HybridHfIntentAgent,
     IntentClassifier,
+    IntentResult,
     build_intent_agent,
 )
 
 __all__ = [
     "FixedExpiryRenewalIntentAgent",
+    "HybridHfIntentAgent",
     "IntentClassifier",
+    "IntentResult",
     "build_intent_agent",
-]
+]
\ No newline at end of file
diff --git a/app/agents/intent/guardrail.py b/app/agents/intent/guardrail.py
new file mode 100644
index 0000000..1c6efef
--- /dev/null
+++ b/app/agents/intent/guardrail.py
@@ -0,0 +1,76 @@
+# BERT 예측을 A.X로 넘길지 결정하는 라우팅 규칙
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+# 가드레일 라우팅 결과
+class RoutingResult:
+
+    should_route: bool
+    reason: str
+    category: str
+
+
+@dataclass
+# HR Intent BERT→A.X 라우팅 가드레일
+class HRRoutingGuardrail:
+
+    margin_threshold: float = 0.76
+    max_trained_labels: int = 3
+    label_prob_threshold: float = 0.55
+    status_kw: list[str] = field(
+        default_factory=lambda: ["없음", "완료", "이상없", "특이사항", "특이문의"]
+    )
+    action_kw: list[str] = field(default_factory=lambda: ["배치", "라인", "지시"])
+    doc_kw: list[str] = field(
+        default_factory=lambda: ["신청서", "서류", "챙겨", "접수", "명단확인"]
+    )
+
+    @staticmethod
+    # 공백 제거 정규화
+    def _normalize(text: str) -> str:
+        return text.replace(" ", "")
+
+    # A.X 호출 여부·사유 판정
+    def should_route_to_ax(
+        self, hr_input: str, probs: dict[str, float], margin: float
+    ) -> RoutingResult:
+        clean_input = self._normalize(hr_input)
+        activated_count = sum(1 for p in probs.values() if p >= self.label_prob_threshold)
+        if activated_count >= self.max_trained_labels:
+            return RoutingResult(
+                should_route=True,
+                reason=f"활성 label {activated_count}개 (학습 최댓값 {self.max_trained_labels}개 이상)",
+                category="OOD_Label_Count",
+            )
+        if any(kw in clean_input for kw in self.status_kw):
+            return RoutingResult(
+                should_route=True, reason="완료/상태 보고 키워드 감지", category="Rule_Status"
+            )
+        action_matches = sum(1 for kw in self.action_kw if kw in clean_input)
+        if action_matches >= 2:
+            return RoutingResult(
+                should_route=True,
+                reason=f"배치/라인/지시 키워드 {action_matches}개 감지",
+                category="Rule_Action",
+            )
+        if "급여계좌" in clean_input or ("급여" in clean_input and "확인" in clean_input):
+            return RoutingResult(
+                should_route=True, reason="급여계좌 관련 경계 키워드 감지", category="Rule_Salary"
+            )
+        if any(kw in clean_input for kw in self.doc_kw):
+            return RoutingResult(
+                should_route=True, reason="서류 확보 키워드 감지", category="Rule_Document"
+            )
+        if margin < self.margin_threshold:
+            return RoutingResult(
+                should_route=True,
+                reason=f"margin {margin:.3f} < {self.margin_threshold}",
+                category="Low_Margin",
+            )
+        return RoutingResult(
+            should_route=False, reason="BERT 신뢰 구간 통과", category="Pass_BERT"
+        )
diff --git a/app/agents/intent/hybrid.py b/app/agents/intent/hybrid.py
new file mode 100644
index 0000000..b0da104
--- /dev/null
+++ b/app/agents/intent/hybrid.py
@@ -0,0 +1,100 @@
+# BERT + 가드레일 + (선택) A.X 하이브리드 Intent 추론
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+
+from .guardrail import HRRoutingGuardrail
+from .models_hf import AxIntentModel, BertIntentModel
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+# 하이브리드 분류 한 건의 정규화 결과
+class HybridIntentPrediction:
+
+    intents: list[str]
+    scores: dict[str, float] = field(default_factory=dict)
+    evidence: dict[str, str | None] = field(default_factory=dict)
+    selected_model: str = "BERT"
+    degraded: bool = False
+
+
+# BERT 우선·필요 시 A.X 보조 파이프라인
+class HybridIntentPipeline:
+
+    # 설정값으로 BERT·가드레일·선택적 A.X 구성
+    def __init__(
+        self,
+        *,
+        bert_model_dir: str,
+        device: str = "cpu",
+        label_prob_threshold: float = 0.55,
+        margin_threshold: float = 0.76,
+        max_trained_labels: int = 3,
+        hf_token: str | None = None,
+        enable_ax: bool = False,
+        ax_base_model_name: str = "skt/A.X-4.0-Light",
+        ax_adapter_path: str = "fowoco/ax-intent-qlora",
+        ax_max_new_tokens: int = 96,
+    ) -> None:
+        self.bert = BertIntentModel(
+            model_dir=bert_model_dir,
+            device=device,
+            label_prob_threshold=label_prob_threshold,
+            hf_token=hf_token,
+        )
+        self.guardrail = HRRoutingGuardrail(
+            margin_threshold=margin_threshold,
+            max_trained_labels=max_trained_labels,
+            label_prob_threshold=label_prob_threshold,
+        )
+        self.ax: AxIntentModel | None = None
+        if enable_ax:
+            try:
+                self.ax = AxIntentModel(
+                    base_model_name=ax_base_model_name,
+                    adapter_path=ax_adapter_path,
+                    device=self.bert.device,
+                    max_new_tokens=ax_max_new_tokens,
+                    hf_token=hf_token,
+                )
+            except Exception:
+                logger.exception("A.X load failed — BERT-only degraded mode")
+
+    # instruction → 정규화 Intent 예측
+    def predict(self, instruction: str) -> HybridIntentPrediction:
+        probs, margin, bert_intents = self.bert.predict(instruction)
+        route = self.guardrail.should_route_to_ax(instruction, probs, margin)
+        if route.should_route and self.ax is not None:
+            try:
+                ax_items = self.ax.predict(instruction)
+                intents = [str(i.get("intent")) for i in ax_items if i.get("intent")]
+                evidence = {
+                    str(i.get("intent")): i.get("evidence")
+                    for i in ax_items
+                    if i.get("intent")
+                }
+                return HybridIntentPrediction(
+                    intents=intents or bert_intents,
+                    scores=probs,
+                    evidence=evidence,
+                    selected_model="AX",
+                    degraded=False,
+                )
+            except Exception:
+                logger.exception("A.X inference failed — BERT fallback")
+                return HybridIntentPrediction(
+                    intents=bert_intents,
+                    scores=probs,
+                    selected_model="BERT_FALLBACK",
+                    degraded=True,
+                )
+        return HybridIntentPrediction(
+            intents=bert_intents,
+            scores=probs,
+            selected_model="BERT",
+            degraded=False,
+        )
diff --git a/app/agents/intent/models_hf.py b/app/agents/intent/models_hf.py
new file mode 100644
index 0000000..338504f
--- /dev/null
+++ b/app/agents/intent/models_hf.py
@@ -0,0 +1,132 @@
+# klue/roberta-base Intent 분류 모델 로드·추론
+
+from __future__ import annotations
+
+from typing import Any
+
+
+# BERT multilabel Intent 분류기
+class BertIntentModel:
+
+    # HF Hub/로컬 경로에서 분류기 로드
+    def __init__(
+        self,
+        model_dir: str,
+        device: str,
+        label_prob_threshold: float = 0.55,
+        hf_token: str | None = None,
+    ) -> None:
+        import torch
+        from transformers import AutoModelForSequenceClassification, AutoTokenizer
+
+        self._torch = torch
+        self.device = (
+            "cuda"
+            if (device == "auto" and torch.cuda.is_available())
+            else (device if device != "auto" else "cpu")
+        )
+        self.tokenizer = AutoTokenizer.from_pretrained(model_dir, token=hf_token)
+        self.model = AutoModelForSequenceClassification.from_pretrained(
+            model_dir, token=hf_token
+        )
+        self.model.to(self.device).eval()
+        self.id2label = self.model.config.id2label
+        self.label_prob_threshold = label_prob_threshold
+
+    # 확률 dict·margin·활성 intent 목록
+    def predict(self, text: str) -> tuple[dict[str, float], float, list[str]]:
+        torch = self._torch
+        with torch.no_grad():
+            enc = self.tokenizer(
+                text, truncation=True, max_length=64, return_tensors="pt"
+            ).to(self.device)
+            logits = self.model(**enc).logits
+            probs_array = torch.sigmoid(logits)[0].cpu().numpy()
+
+        probs_dict = {self.id2label[i]: float(p) for i, p in enumerate(probs_array)}
+        activated = [p for p in probs_array if p >= self.label_prob_threshold]
+        not_activated = [p for p in probs_array if p < self.label_prob_threshold]
+        if not activated:
+            margin = float(max(probs_array)) - self.label_prob_threshold
+        else:
+            margin = float(min(activated)) - (
+                float(max(not_activated)) if not_activated else 0.0
+            )
+        picked = [
+            self.id2label[i]
+            for i, p in enumerate(probs_array)
+            if p >= self.label_prob_threshold
+        ]
+        if not picked:
+            picked = [self.id2label[int(probs_array.argmax())]]
+        return probs_dict, margin, picked
+
+
+# A.X-4.0-Light QLoRA Intent 보조 모델 (GPU·bitsandbytes 필요)
+class AxIntentModel:
+
+    _SYSTEM_PROMPT = (
+        "당신은 HR 업무 요청 문장(hr_input)을 분석하여 의도(Intent)를 분류하는 전문 AI 에이전트입니다.\n"
+        "Intent + evidence 추출까지가 책임입니다.\n"
+        '출력은 JSON만: {"intents": [{"intent": "INTENT_CODE", "evidence": "...|null"}]}'
+    )
+
+    # 4bit 베이스 + Peft 어댑터 로드
+    def __init__(
+        self,
+        base_model_name: str,
+        adapter_path: str,
+        device: str,
+        max_new_tokens: int = 96,
+        hf_token: str | None = None,
+    ) -> None:
+        import torch
+        from peft import PeftModel
+        from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
+
+        self._torch = torch
+        self.max_new_tokens = max_new_tokens
+        bnb_config = BitsAndBytesConfig(
+            load_in_4bit=True,
+            bnb_4bit_quant_type="nf4",
+            bnb_4bit_compute_dtype=torch.float16,
+        )
+        base_model = AutoModelForCausalLM.from_pretrained(
+            base_model_name,
+            quantization_config=bnb_config,
+            torch_dtype=torch.float16,
+            device_map={"": 0} if device != "cpu" else "cpu",
+            token=hf_token,
+        )
+        self.tokenizer = AutoTokenizer.from_pretrained(base_model_name, token=hf_token)
+        self.model = PeftModel.from_pretrained(base_model, adapter_path, token=hf_token)
+        self.model.eval()
+
+    # Intent 목록 [{"intent","evidence"}] — 실패 시 예외
+    def predict(self, hr_input: str) -> list[dict[str, Any]]:
+        import json
+        import re
+
+        messages = [
+            {"role": "system", "content": self._SYSTEM_PROMPT},
+            {"role": "user", "content": hr_input},
+        ]
+        inputs = self.tokenizer.apply_chat_template(
+            messages,
+            add_generation_prompt=True,
+            tokenize=True,
+            return_dict=True,
+            return_tensors="pt",
+        ).to(self.model.device)
+        with self._torch.no_grad():
+            output = self.model.generate(
+                **inputs, max_new_tokens=self.max_new_tokens, do_sample=False
+            )
+        raw = self.tokenizer.decode(
+            output[0][inputs["input_ids"].shape[-1] :], skip_special_tokens=True
+        )
+        match = re.search(r"\{.*\}", raw, re.DOTALL)
+        if not match:
+            raise ValueError(f"A.X output could not be parsed as JSON: {raw!r}")
+        parsed = json.loads(match.group(0))
+        return list(parsed.get("intents") or [])
diff --git a/app/agents/intent/service.py b/app/agents/intent/service.py
index 863cddd..82c3aa0 100644
--- a/app/agents/intent/service.py
+++ b/app/agents/intent/service.py
@@ -1,11 +1,14 @@
-# Intent·Slot — EXPIRY_RENEWAL 고정. WF 매핑은 Knowledge workflow_catalog 근거
+# Intent·Slot — HF 하이브리드 또는 EXPIRY_RENEWAL 고정. WF는 Knowledge catalog 근거
 
 from __future__ import annotations
 
+import logging
 import re
 from dataclasses import dataclass, field
 from typing import Protocol
 
+logger = logging.getLogger(__name__)
+
 # Knowledge workflow_catalog.yaml: intent → workflow id (동일 intent면 catalog 등장 순)
 INTENT_TO_WORKFLOWS: dict[str, list[str]] = {
     "WORKER_ONBOARDING": ["WF-WRK-001"],
@@ -117,6 +120,95 @@ def classify(
         )
 
 
-# 재갱신 고정 Intent 분류기 생성
+# 멀티라벨 예측에서 대표 Intent·confidence 선택
+def _primary_intent(
+    intents: list[str], scores: dict[str, float]
+) -> tuple[str, float]:
+    if not intents:
+        return "OUT_OF_SCOPE", 0.0
+    if "OUT_OF_SCOPE" in intents and len(intents) == 1:
+        return "OUT_OF_SCOPE", float(scores.get("OUT_OF_SCOPE") or 0.5)
+    ranked = [i for i in intents if i != "OUT_OF_SCOPE"]
+    if not ranked:
+        return "OUT_OF_SCOPE", float(scores.get("OUT_OF_SCOPE") or 0.5)
+    if scores:
+        best = max(ranked, key=lambda name: float(scores.get(name) or 0.0))
+        return best, float(scores.get(best) or 0.0)
+    return ranked[0], 0.85
+
+
+# HF BERT(+선택 A.X) 하이브리드 → IntentClassifier
+class HybridHfIntentAgent:
+
+    # 파이프라인은 첫 classify 때 lazy 로드
+    def __init__(self, pipeline: object | None = None) -> None:
+        self._pipeline = pipeline
+        self._load_error: str | None = None
+
+    # 설정 기반 HybridIntentPipeline 확보
+    def _ensure_pipeline(self) -> object | None:
+        if self._pipeline is not None:
+            return self._pipeline
+        if self._load_error is not None:
+            return None
+        try:
+            import os
+
+            from app.core.config import get_settings
+
+            from .hybrid import HybridIntentPipeline
+
+            settings = get_settings()
+            token = settings.hf_token or os.environ.get("HF_TOKEN")
+            self._pipeline = HybridIntentPipeline(
+                bert_model_dir=settings.intent_bert_model_dir,
+                device=settings.intent_device,
+                label_prob_threshold=settings.intent_label_prob_threshold,
+                margin_threshold=settings.intent_margin_threshold,
+                max_trained_labels=settings.intent_max_trained_labels,
+                hf_token=token,
+                enable_ax=settings.intent_enable_ax,
+                ax_base_model_name=settings.intent_ax_base_model,
+                ax_adapter_path=settings.intent_ax_adapter_path,
+                ax_max_new_tokens=settings.intent_ax_max_new_tokens,
+            )
+        except Exception as exc:
+            self._load_error = str(exc)
+            logger.exception("HF Intent pipeline load failed — fixed EXPIRY fallback")
+            return None
+        return self._pipeline
+
+    # HF 분류 결과를 IntentResult로 변환
+    def classify(
+        self,
+        instruction: str,
+        *,
+        workflow_constraints: list[str] | None = None,
+    ) -> IntentResult:
+        pipeline = self._ensure_pipeline()
+        if pipeline is None:
+            return FixedExpiryRenewalIntentAgent().classify(
+                instruction, workflow_constraints=workflow_constraints
+            )
+        prediction = pipeline.predict(instruction)  # type: ignore[attr-defined]
+        intent, confidence = _primary_intent(prediction.intents, prediction.scores)
+        slots: dict[str, str] = {}
+        for name, evidence in (prediction.evidence or {}).items():
+            if evidence:
+                slots[f"evidence:{name}"] = str(evidence)
+        return IntentResult(
+            intent=intent,
+            confidence=max(0.0, min(1.0, confidence)),
+            workflow_id=resolve_workflow_id(intent, workflow_constraints),
+            extracted_slots=slots,
+        )
+
+
+# 설정에 따라 HF 하이브리드 또는 재갱신 고정 분류기 생성
 def build_intent_agent() -> IntentClassifier:
+    from app.core.config import get_settings
+
+    settings = get_settings()
+    if settings.intent_model_enabled:
+        return HybridHfIntentAgent()
     return FixedExpiryRenewalIntentAgent()
diff --git a/app/core/config.py b/app/core/config.py
index 972ebd3..1c90bbc 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -43,6 +43,20 @@ class Settings(BaseSettings):
     knowledge_enabled: bool = False
     knowledge_root: str | None = None
 
+    # Intent HF — true면 BERT(+선택 A.X) 분류, false면 EXPIRY_RENEWAL 고정
+    intent_model_enabled: bool = False
+    intent_bert_model_dir: str = "fowoco/klue-roberta-base-intent-classifier"
+    intent_ax_base_model: str = "skt/A.X-4.0-Light"
+    intent_ax_adapter_path: str = "fowoco/ax-intent-qlora"
+    intent_enable_ax: bool = False
+    intent_device: str = "cpu"
+    intent_margin_threshold: float = 0.76
+    intent_max_trained_labels: int = 3
+    intent_label_prob_threshold: float = 0.55
+    intent_ax_max_new_tokens: int = 96
+    # private HF 모델용. 미설정 시 환경변수 HF_TOKEN도 허용(코드에서 조회)
+    hf_token: str | None = None
+
     # Supervisor — rules(기본) | llm(FOWOCO_LLM_* 필요, 실패 시 rules 폴백)
     supervisor_mode: str = "rules"
 
diff --git a/pyproject.toml b/pyproject.toml
index 5fbe8e7..ed16622 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -33,7 +33,14 @@ dev = [
 ]
 # 로컬: pip install -e ../knowledge/fowoco-knowledge
 knowledge = []
-
+# HF Intent (BERT 필수, A.X는 FOWOCO_INTENT_ENABLE_AX=true 시 peft·bitsandbytes)
+intent = [
+  "torch>=2.2,<3",
+  "transformers>=4.46,<5",
+  "accelerate>=0.34,<2",
+  "peft>=0.13,<1",
+  "bitsandbytes>=0.44,<1",
+]
 [tool.setuptools.packages.find]
 where = ["."]
 include = ["app*"]
diff --git a/tests/agents/test_intent_hybrid.py b/tests/agents/test_intent_hybrid.py
new file mode 100644
index 0000000..e165442
--- /dev/null
+++ b/tests/agents/test_intent_hybrid.py
@@ -0,0 +1,53 @@
+# HF Intent 에이전트·대표 Intent 선택 단위 테스트
+
+from app.agents.intent.hybrid import HybridIntentPrediction
+from app.agents.intent.service import (
+    FixedExpiryRenewalIntentAgent,
+    HybridHfIntentAgent,
+    _primary_intent,
+    build_intent_agent,
+)
+
+
+# 점수 있을 때 최고 점수 Intent 선택
+def test_primary_intent_picks_highest_score() -> None:
+    intent, conf = _primary_intent(
+        ["DOCUMENT_REQUEST", "EXPIRY_RENEWAL"],
+        {"DOCUMENT_REQUEST": 0.4, "EXPIRY_RENEWAL": 0.91},
+    )
+    assert intent == "EXPIRY_RENEWAL"
+    assert conf == 0.91
+
+
+# OUT_OF_SCOPE 단독 유지
+def test_primary_intent_out_of_scope_alone() -> None:
+    intent, conf = _primary_intent(["OUT_OF_SCOPE"], {"OUT_OF_SCOPE": 0.8})
+    assert intent == "OUT_OF_SCOPE"
+    assert conf == 0.8
+
+
+# 기본 빌드는 재갱신 고정
+def test_build_intent_agent_defaults_to_fixed() -> None:
+    agent = build_intent_agent()
+    assert isinstance(agent, FixedExpiryRenewalIntentAgent)
+    result = agent.classify("아무 말")
+    assert result.intent == "EXPIRY_RENEWAL"
+    assert result.workflow_id in {"WF-STY-001", "WF-CON-001", ""}
+
+
+# 주입된 파이프라인으로 Hybrid 에이전트 분류
+def test_hybrid_agent_maps_pipeline_prediction() -> None:
+    class _FakePipe:
+        def predict(self, instruction: str) -> HybridIntentPrediction:
+            del instruction
+            return HybridIntentPrediction(
+                intents=["EXPIRY_RENEWAL"],
+                scores={"EXPIRY_RENEWAL": 0.93},
+                selected_model="BERT",
+            )
+
+    agent = HybridHfIntentAgent(pipeline=_FakePipe())
+    result = agent.classify("응웬반안 체류연장")
+    assert result.intent == "EXPIRY_RENEWAL"
+    assert result.confidence == 0.93
+    assert result.workflow_id == "WF-STY-001"

From 140a026a55fa4c938acf7d7254638799e02b4c68 Mon Sep 17 00:00:00 2001
From: HWIYA <dlgnl117@gmail.com>
Date: Wed, 5 Aug 2026 14:47:46 +0900
Subject: [PATCH 2/4] =?UTF-8?q?feat:=20analyses=20PLAN/ANALYZE=20=EA=B3=84?=
 =?UTF-8?q?=EC=95=BD=20(Issue=20#6)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

PLAN→CONTEXT_REQUIRED, ANALYZE→NEEDS_INFO|REVIEW_REQUIRED. HTTP 와이어 최소 페이로드와 맞춤.
---
 app/agents/pipeline.py                        | 296 +++++++++++-------
 app/api/routes/analyses.py                    |  10 +-
 app/api/schemas/analyses.py                   |  76 +++--
 docs/analyses-contract.md                     | 226 +++++++------
 examples/analyses/request_analyze.json        |  20 ++
 examples/analyses/request_plan.json           |   7 +
 .../analyses/response_context_required.json   |  29 ++
 examples/analyses/response_needs_info.json    |  20 +-
 .../analyses/response_review_required.json    |  12 +-
 tests/api/test_analyses_endpoint.py           | 223 ++++++-------
 tests/api/test_internal_handshake.py          |  29 +-
 11 files changed, 526 insertions(+), 422 deletions(-)
 create mode 100644 examples/analyses/request_analyze.json
 create mode 100644 examples/analyses/request_plan.json
 create mode 100644 examples/analyses/response_context_required.json

diff --git a/app/agents/pipeline.py b/app/agents/pipeline.py
index c833e95..a848a84 100644
--- a/app/agents/pipeline.py
+++ b/app/agents/pipeline.py
@@ -1,51 +1,92 @@
-# Analyses MVP 파이프라인 — Intent → Ambiguity → Workflow
+# Analyses 파이프라인 — PLAN(CONTEXT_REQUIRED) → ANALYZE(NEEDS_INFO|REVIEW_REQUIRED)
 
 from __future__ import annotations
 
+import re
 import time
 from uuid import uuid4
 
 from app import __version__
 from app.api.schemas.analyses import (
+    DEFAULT_CONTRACT_VERSION,
+    DEFAULT_KNOWLEDGE_VERSION,
     AnalysisCandidate,
+    AnalysisQuestion,
     AnalysisRequest,
     AnalysisResponse,
     AnalysisVersions,
+    ContextRequirement,
     WorkerContext,
 )
 
 from .ambiguity import AmbiguityAgent
 from .intent import IntentClassifier, build_intent_agent
-from .intent.service import public_workflow_id
 from .workflow import WorkflowAgent
 
-# requestedFields 맵에서 extractedSlots로 승격할 키
-_REQUESTED_FIELD_SLOT_KEYS = frozenset(
-    {
-        "legal_name",
-        "full_name",
-        "passport_number",
-        "phone",
-        "email",
-        "company_id",
-        "alien_registration_number",
-        "date_of_birth",
-        "nationality",
-        "wage",
-        "monthly_wage",
-    }
+# instruction 끝의 `, INTENT_TAG` 제거
+_INTENT_TAG_SUFFIX = re.compile(r",\s*[A-Z][A-Z0-9_]+\s*$")
+# 대상 이름 추정 시 끊을 토큰
+_NAME_STOP_PREFIXES = (
+    "체류",
+    "계약",
+    "서류",
+    "급여",
+    "연장",
+    "준비",
+    "요청",
+    "등록",
+    "변경",
+    "안내",
 )
 
-
-# WorkerContext·requestedFields를 초기 슬롯으로 시드
+_SLOT_PROMPTS: dict[str, str] = {
+    "worker_id": "대상 근로자를 지정해 주세요.",
+    "stay_expiry_date": "체류 만료일을 입력해 주세요.",
+    "contract_end_date": "계약 종료일을 입력해 주세요.",
+    "contract_start_date": "계약 시작일을 입력해 주세요.",
+    "legal_name": "여권상 성명을 입력해 주세요.",
+    "full_name": "성명을 입력해 주세요.",
+    "monthly_wage": "월 급여를 입력해 주세요.",
+    "wage": "급여를 입력해 주세요.",
+    "document_type": "서류 종류를 입력해 주세요.",
+    "pay_period": "급여 기간을 입력해 주세요.",
+    "change_type": "고용 변동 유형을 입력해 주세요.",
+}
+
+
+# 발화문에서 대상 표시 이름 추정
+def _guess_target_display_name(instruction: str) -> str:
+    text = _INTENT_TAG_SUFFIX.sub("", instruction).strip()
+    if not text:
+        return "unknown"
+    parts: list[str] = []
+    for tok in text.split():
+        if tok in {"의", "을", "를", "이", "가"}:
+            break
+        if any(tok.startswith(p) for p in _NAME_STOP_PREFIXES):
+            break
+        parts.append(tok)
+    name = " ".join(parts).strip()
+    return name or "unknown"
+
+
+# HR 질문 prompt 생성
+def _question_for(slot_key: str) -> AnalysisQuestion:
+    prompt = _SLOT_PROMPTS.get(slot_key) or f"{slot_key} 값을 입력해 주세요."
+    return AnalysisQuestion(slot_key=slot_key, prompt=prompt)
+
+
+# Worker requestedFields·식별자를 슬롯으로 시드
 def _seed_slots_from_worker(worker: WorkerContext) -> dict[str, str]:
     slots: dict[str, str] = {}
     if worker.worker_ref:
         slots["worker_id"] = worker.worker_ref
     if worker.display_name:
         slots["display_name"] = worker.display_name
+        slots.setdefault("full_name", worker.display_name)
     if worker.nationality_code:
         slots["nationality_code"] = worker.nationality_code
+        slots.setdefault("nationality", worker.nationality_code)
     if worker.stay_expiry_date:
         slots["stay_expiry_date"] = worker.stay_expiry_date
     if worker.contract_start_date:
@@ -53,20 +94,27 @@ def _seed_slots_from_worker(worker: WorkerContext) -> dict[str, str]:
     if worker.contract_end_date:
         slots["contract_end_date"] = worker.contract_end_date
     for key, value in worker.requested_fields.items():
-        if key in _REQUESTED_FIELD_SLOT_KEYS and value:
+        if value:
             slots.setdefault(key, value)
             if key == "legal_name":
                 slots.setdefault("full_name", value)
-    company_id = worker.requested_fields.get("company_id")
-    if company_id:
-        slots.setdefault("company_id", company_id)
     return slots
 
 
-# 의도 분류·슬롯 검사·워크플로 조회를 이어 붙인 분석기
+# 고정 버전 블록
+def _versions() -> AnalysisVersions:
+    return AnalysisVersions(
+        agent_version=__version__,
+        contract_version=DEFAULT_CONTRACT_VERSION,
+        workflow_catalog_version=DEFAULT_KNOWLEDGE_VERSION,
+        context_pack_version=DEFAULT_KNOWLEDGE_VERSION,
+    )
+
+
+# Intent → requiredFieldKeys / questions·candidates
 class AnalysisPipeline:
 
-    # 하위 에이전트 수신 안 주면 설정 기반 Intent + 기본 Ambiguity/Workflow 사용
+    # 하위 에이전트 미주입 시 기본 Intent·Ambiguity·Workflow
     def __init__(
         self,
         *,
@@ -78,103 +126,127 @@ def __init__(
         self._ambiguity = ambiguity_agent or AmbiguityAgent()
         self._workflow = workflow_agent or WorkflowAgent()
 
-    # 요청 실행 → outcome·candidates 목록 생성
+    # phase에 따라 CONTEXT_REQUIRED 또는 최종 outcome 반환
     def run(self, request: AnalysisRequest) -> AnalysisResponse:
         start = time.monotonic()
-        ai = request.analysis_input
-        instruction = ai.instruction
-        constraint_ids = [c.workflow_id for c in ai.workflow_constraints]
-        allowed_keys_by_constraint = {
-            c.workflow_id: c.allowed_slot_keys for c in ai.workflow_constraints
-        }
-
-        candidates: list[AnalysisCandidate] = []
-
-        for worker in ai.workers:
-            intent_result = self._intent.classify(
-                instruction, workflow_constraints=constraint_ids or None
-            )
+        if request.phase == "PLAN":
+            response = self._run_plan(request)
+        else:
+            response = self._run_analyze(request)
+        elapsed_ms = int((time.monotonic() - start) * 1000)
+        response.latency_ms = elapsed_ms
+        return response
+
+    # PLAN — Intent 확정 후 DB canonical key 요청
+    def _run_plan(self, request: AnalysisRequest) -> AnalysisResponse:
+        instruction = request.analysis_input.instruction
+        intent_result = self._intent.classify(instruction)
+        workflow_id = intent_result.workflow_id or ""
+        # Issue #6: Knowledge canonical key 전체 (worker_id 포함)
+        if intent_result.intent == "OUT_OF_SCOPE":
+            field_keys = ["worker_id"]
+        else:
+            required = self._required_slots_for(workflow_id)
+            field_keys = list(required) if required else ["worker_id", "stay_expiry_date"]
 
-            if not intent_result.workflow_id:
-                candidates.append(
-                    AnalysisCandidate(
-                        candidate_ref=f"candidate-{uuid4().hex[:8]}",
-                        worker_ref=worker.worker_ref,
-                        workflow_id="UNKNOWN",
-                        extracted_slots={},
-                        missing_slots=[],
-                        confidence=intent_result.confidence,
-                    )
-                )
-                continue
-
-            workflow = self._workflow.get_workflow(intent_result.workflow_id)
-            if workflow is None:
-                candidates.append(
-                    AnalysisCandidate(
-                        candidate_ref=f"candidate-{uuid4().hex[:8]}",
-                        worker_ref=worker.worker_ref,
-                        workflow_id="UNKNOWN",
-                        extracted_slots={},
-                        missing_slots=[],
-                        confidence=intent_result.confidence,
-                    )
-                )
-                continue
-
-            response_workflow_id = public_workflow_id(
-                internal_workflow_id=intent_result.workflow_id,
-                intent=intent_result.intent,
-                constraints=constraint_ids,
-            )
+        return AnalysisResponse(
+            request_id=request.request_id,
+            outcome="CONTEXT_REQUIRED",
+            context_requirement=ContextRequirement(
+                detected_intent=intent_result.intent or "UNKNOWN",
+                confidence=intent_result.confidence,
+                target_display_name=_guess_target_display_name(instruction),
+                extracted_slots=dict(intent_result.extracted_slots),
+                required_field_keys=field_keys,
+            ),
+            questions=[],
+            candidates=[],
+            validation_errors=[],
+            versions=_versions(),
+            provider_attempt_count=1,
+            latency_ms=0,
+        )
 
-            seeded = _seed_slots_from_worker(worker)
-            for key, value in seeded.items():
-                intent_result.extracted_slots.setdefault(key, value)
-
-            allowed_keys = allowed_keys_by_constraint.get(
-                response_workflow_id
-            ) or allowed_keys_by_constraint.get(intent_result.workflow_id)
-            filtered_slots = intent_result.extracted_slots
-            if allowed_keys:
-                # Step 3 조회 키는 allowed 목록과 무관하게 항상 유지
-                keep = {"worker_id", "company_id"}
-                filtered_slots = {
-                    k: v
-                    for k, v in intent_result.extracted_slots.items()
-                    if k in allowed_keys or k in keep
-                }
-
-            amb_result = self._ambiguity.check(
-                intent_result.workflow_id, filtered_slots, instruction
+    # ANALYZE — DB 보충값으로 NEEDS_INFO | REVIEW_REQUIRED
+    def _run_analyze(self, request: AnalysisRequest) -> AnalysisResponse:
+        ai = request.analysis_input
+        instruction = ai.instruction
+        intent_result = self._intent.classify(instruction)
+        workflow_id = intent_result.workflow_id or ""
+        public_wf = intent_result.intent or workflow_id or "UNKNOWN"
+
+        if not ai.workers:
+            return AnalysisResponse(
+                request_id=request.request_id,
+                outcome="NEEDS_INFO",
+                context_requirement=None,
+                questions=[_question_for("worker_id")],
+                candidates=[],
+                validation_errors=[],
+                versions=_versions(),
+                provider_attempt_count=1,
+                latency_ms=0,
             )
 
-            candidates.append(
-                AnalysisCandidate(
-                    candidate_ref=f"candidate-{uuid4().hex[:8]}",
-                    worker_ref=worker.worker_ref,
-                    workflow_id=response_workflow_id,
-                    extracted_slots=filtered_slots,
-                    missing_slots=amb_result.missing_slots,
-                    confidence=intent_result.confidence,
-                )
+        # MVP: 근로자 1명만
+        worker = ai.workers[0]
+        slots = _seed_slots_from_worker(worker)
+
+        # Server가 못 채운 PLAN 요청 키 → HR 질문 후보
+        hr_keys: list[str] = []
+        for key in ai.requested_field_keys:
+            if key not in worker.requested_fields and key not in slots:
+                hr_keys.append(key)
+
+        amb = self._ambiguity.check(workflow_id, slots, instruction)
+        for key in amb.missing_slots:
+            if key not in slots and key not in hr_keys:
+                hr_keys.append(key)
+
+        if hr_keys:
+            return AnalysisResponse(
+                request_id=request.request_id,
+                outcome="NEEDS_INFO",
+                context_requirement=None,
+                questions=[_question_for(k) for k in hr_keys],
+                candidates=[],
+                validation_errors=[],
+                versions=_versions(),
+                provider_attempt_count=1,
+                latency_ms=0,
             )
 
-        has_missing = any(c.missing_slots for c in candidates)
-        has_low_confidence = any(c.confidence < 0.65 for c in candidates)
-        outcome = "NEEDS_INFO" if (has_missing or has_low_confidence) else "REVIEW_REQUIRED"
-
-        elapsed_ms = int((time.monotonic() - start) * 1000)
-
+        # 응답 workflowId는 Intent형(서버 fixture) 우선
+        if intent_result.intent:
+            public_wf = intent_result.intent
+        elif self._workflow.get_workflow(workflow_id):
+            public_wf = workflow_id
+
+        candidate = AnalysisCandidate(
+            candidate_ref=f"candidate-{uuid4().hex[:8]}",
+            worker_ref=worker.worker_ref,
+            workflow_id=public_wf,
+            extracted_slots=slots,
+            missing_slots=[],
+            confidence=intent_result.confidence,
+        )
         return AnalysisResponse(
             request_id=request.request_id,
-            outcome=outcome,
-            candidates=candidates,
+            outcome="REVIEW_REQUIRED",
+            context_requirement=None,
+            questions=[],
+            candidates=[candidate],
             validation_errors=[],
-            versions=AnalysisVersions(
-                agent_version=__version__,
-                contract_version=request.contract_version,
-            ),
+            versions=_versions(),
             provider_attempt_count=1,
-            latency_ms=elapsed_ms,
+            latency_ms=0,
         )
+
+    # workflow 필수 슬롯 (Catalog → Ambiguity/Knowledge 폴백)
+    def _required_slots_for(self, workflow_id: str) -> list[str]:
+        if not workflow_id:
+            return []
+        info = self._workflow.get_workflow(workflow_id)
+        if info and info.required_slots:
+            return list(info.required_slots)
+        return self._ambiguity.check(workflow_id, {}, "").missing_slots
diff --git a/app/api/routes/analyses.py b/app/api/routes/analyses.py
index fac2c21..275ca19 100644
--- a/app/api/routes/analyses.py
+++ b/app/api/routes/analyses.py
@@ -14,15 +14,15 @@
 @router.post(
     "/analyses",
     response_model=AnalysisResponse,
-    summary="자연어 지시 분석",
+    summary="자연어 지시 분석 (PLAN / ANALYZE)",
     description=(
-        "Server가 analysisInput(instruction·WorkerContext·requestedFields)을 보내면 "
-        "Intent 분류, Slot 추출, 모호성 검사 후 candidate 목록 반환. "
-        "requestId/attemptId·Bearer(#8). 응답은 Server strict JSON 계약 필드만."
+        "phase=PLAN 이면 Intent 분류 후 CONTEXT_REQUIRED(requiredFieldKeys) 반환. "
+        "phase=ANALYZE 이면 DB 보충값으로 NEEDS_INFO(questions) 또는 REVIEW_REQUIRED(candidates). "
+        "HTTP는 requestId·phase·analysisInput만. Bearer(#8)."
     ),
     dependencies=[Depends(verify_internal_bearer)],
 )
-# 지시문 분석 → Intent·Slot·누락 정보 서버 응답
+# PLAN/ANALYZE 분기 → Server 계약 outcome 응답
 async def analyze(
     request: AnalysisRequest,
     pipeline: AnalysisPipeline = Depends(get_analysis_pipeline),  # noqa: B008
diff --git a/app/api/schemas/analyses.py b/app/api/schemas/analyses.py
index 2aa10d9..91355ee 100644
--- a/app/api/schemas/analyses.py
+++ b/app/api/schemas/analyses.py
@@ -1,62 +1,55 @@
-# POST /internal/v1/analyses 요청·응답 스키마 (Server #56 계약)
+# POST /internal/v1/analyses 요청·응답 스키마 (Server ai-runtime-contract)
 
 from __future__ import annotations
 
-from pydantic import BaseModel, Field
+from typing import Literal
 
+from pydantic import BaseModel, Field
 
-# 서버가 허용한 워크플로·슬롯 범위
-class WorkflowConstraint(BaseModel):
+AnalysisPhase = Literal["PLAN", "ANALYZE"]
+AnalysisOutcome = Literal["CONTEXT_REQUIRED", "NEEDS_INFO", "REVIEW_REQUIRED"]
 
-    workflow_id: str = Field(..., alias="workflowId")
-    allowed_slot_keys: list[str] = Field(default_factory=list, alias="allowedSlotKeys")
-
-    model_config = {"populate_by_name": True}
+DEFAULT_CONTRACT_VERSION = "1.0.0"
+DEFAULT_KNOWLEDGE_VERSION = "0.2.0"
 
 
-# 요청에 실린 근로자 컨텍스트 (Server WorkerContext)
+# HTTP 와이어 Worker — workerRef + requestedFields (나머지 필드는 선택·하위호환)
 class WorkerContext(BaseModel):
 
-    worker_ref: str = Field(..., alias="workerRef", description="서버 worker_id와 동일")
-    display_name: str = Field(..., alias="displayName")
+    worker_ref: str = Field(..., alias="workerRef", description="서버 worker_id")
+    display_name: str | None = Field(None, alias="displayName")
     nationality_code: str | None = Field(None, alias="nationalityCode")
-    preferred_language: str = Field("ko", alias="preferredLanguage")
-    work_status: str = Field("ACTIVE", alias="workStatus")
+    preferred_language: str | None = Field(None, alias="preferredLanguage")
+    work_status: str | None = Field(None, alias="workStatus")
     stay_expiry_date: str | None = Field(None, alias="stayExpiryDate")
     contract_start_date: str | None = Field(None, alias="contractStartDate")
     contract_end_date: str | None = Field(None, alias="contractEndDate")
-    # Agent가 요구한 field의 Server 원본값 (서비스 인증정보 금지)
     requested_fields: dict[str, str] = Field(default_factory=dict, alias="requestedFields")
 
     model_config = {"populate_by_name": True}
 
 
-# HR 지시문과 근로자·제약 목록
+# HR 지시 + PLAN/ANALYZE 문맥 (HTTP 최소 페이로드)
 class AnalysisInput(BaseModel):
 
     instruction: str
+    requested_field_keys: list[str] = Field(default_factory=list, alias="requestedFieldKeys")
     workers: list[WorkerContext] = Field(default_factory=list)
-    workflow_constraints: list[WorkflowConstraint] = Field(
-        default_factory=list, alias="workflowConstraints"
-    )
 
     model_config = {"populate_by_name": True}
 
 
-# Server → AI 분석 요청
+# Server → AI 분석 요청 (attemptId·version·deadline은 HTTP에 없음)
 class AnalysisRequest(BaseModel):
 
     request_id: str = Field(..., alias="requestId")
-    attempt_id: str = Field(..., alias="attemptId")
-    contract_version: str = Field("1.0.0", alias="contractVersion")
-    required_knowledge_version: str = Field("0.2.0", alias="requiredKnowledgeVersion")
-    deadline_ms: int = Field(10_000, alias="deadlineMs")
+    phase: AnalysisPhase
     analysis_input: AnalysisInput = Field(..., alias="analysisInput")
 
     model_config = {"populate_by_name": True}
 
 
-# 기계 판독용 검증 오류 (자유문 Provider 메시지 금지)
+# 기계 판독용 검증 오류
 class ValidationErrorItem(BaseModel):
 
     code: str
@@ -65,6 +58,27 @@ class ValidationErrorItem(BaseModel):
     model_config = {"populate_by_name": True}
 
 
+# PLAN 후 Server DB 조회 요청
+class ContextRequirement(BaseModel):
+
+    detected_intent: str = Field(..., alias="detectedIntent")
+    confidence: float = Field(..., ge=0.0, le=1.0)
+    target_display_name: str = Field(..., alias="targetDisplayName")
+    extracted_slots: dict[str, str] = Field(default_factory=dict, alias="extractedSlots")
+    required_field_keys: list[str] = Field(..., alias="requiredFieldKeys")
+
+    model_config = {"populate_by_name": True}
+
+
+# NEEDS_INFO 시 HR 질문 1건
+class AnalysisQuestion(BaseModel):
+
+    slot_key: str = Field(..., alias="slotKey")
+    prompt: str
+
+    model_config = {"populate_by_name": True}
+
+
 # 분석 결과 후보 1건 — Server AiCandidate 와이어 필드만
 class AnalysisCandidate(BaseModel):
 
@@ -86,18 +100,22 @@ class AnalysisVersions(BaseModel):
     model_name: str = Field("stub", alias="modelName")
     model_version: str = Field("stub", alias="modelVersion")
     prompt_version: str = Field("prompt-1", alias="promptVersion")
-    context_pack_version: str = Field("0.2.0", alias="contextPackVersion")
-    workflow_catalog_version: str = Field("0.2.0", alias="workflowCatalogVersion")
-    contract_version: str = Field("1.0.0", alias="contractVersion")
+    context_pack_version: str = Field(DEFAULT_KNOWLEDGE_VERSION, alias="contextPackVersion")
+    workflow_catalog_version: str = Field(
+        DEFAULT_KNOWLEDGE_VERSION, alias="workflowCatalogVersion"
+    )
+    contract_version: str = Field(DEFAULT_CONTRACT_VERSION, alias="contractVersion")
 
     model_config = {"populate_by_name": True}
 
 
-# AI → Server 분석 응답 (unknown field 금지 — Server FAIL_ON_UNKNOWN_PROPERTIES)
+# AI → Server 분석 응답 (unknown field 금지)
 class AnalysisResponse(BaseModel):
 
     request_id: str = Field(..., alias="requestId")
-    outcome: str = Field(..., description="NEEDS_INFO | REVIEW_REQUIRED")
+    outcome: AnalysisOutcome
+    context_requirement: ContextRequirement | None = Field(None, alias="contextRequirement")
+    questions: list[AnalysisQuestion] = Field(default_factory=list)
     candidates: list[AnalysisCandidate] = Field(default_factory=list)
     validation_errors: list[ValidationErrorItem] = Field(
         default_factory=list, alias="validationErrors"
diff --git a/docs/analyses-contract.md b/docs/analyses-contract.md
index 0db3991..c1cec31 100644
--- a/docs/analyses-contract.md
+++ b/docs/analyses-contract.md
@@ -1,7 +1,8 @@
 # Analyses Runtime 계약 (AI 소유)
 
-Server `docs/ai-runtime-contract.md`(PR #56)와 맞추기 위한 AI 쪽 계약 요약이다.
-Language Agent는 아래 응답 fixture를 임시 입력으로 써도 된다.
+Server `docs/ai-runtime-contract.md` + `AiRuntimeHttpRequest` (fowoco/server main)과 맞춘다.  
+**HTTP 와이어**는 최소 페이로드다. `attemptId` / version / deadline / `extractedSlots` /
+`workflowConstraints` 는 Server 내부 `AiAnalysisRequest`에만 있고 **요청 JSON에 실리지 않는다**.
 
 ## Endpoint
 
@@ -9,145 +10,170 @@ Language Agent는 아래 응답 fixture를 임시 입력으로 써도 된다.
 POST /internal/v1/analyses
 ```
 
-`/api/v1` prefix를 붙이지 않는다. (이전 `/api/v1/internal/v1/analyses` 는 제거)
+## 흐름
 
-## 요청 계약 (`analysisInput`)
+```text
+PLAN  → CONTEXT_REQUIRED (requiredFieldKeys)
+      → Server DB 조회
+ANALYZE → NEEDS_INFO (questions) | REVIEW_REQUIRED (candidates)
+```
+
+`CONTEXT_REQUIRED` / `NEEDS_INFO` / `REVIEW_REQUIRED` 는 모두 **성공 outcome** 이다 (`FAILED` 아님).
 
-옛 `maskedInput` / `maskedInstruction`은 사용하지 않는다.
+---
+
+## 1) PLAN 요청 (Server → AI)
 
 ```json
 {
   "requestId": "10000000-0000-0000-0000-000000000001",
-  "attemptId": "20000000-0000-0000-0000-000000000001",
-  "contractVersion": "1.0.0",
-  "requiredKnowledgeVersion": "0.2.0",
-  "deadlineMs": 10000,
+  "phase": "PLAN",
   "analysisInput": {
-    "instruction": "가상 근로자 응웬반안(010-1234-5678)의 체류연장 준비",
-    "workers": [
-      {
-        "workerRef": "30000000-0000-0000-0000-000000000001",
-        "displayName": "응웬반안",
-        "nationalityCode": "VN",
-        "preferredLanguage": "vi",
-        "workStatus": "ACTIVE",
-        "stayExpiryDate": "2026-12-31",
-        "contractStartDate": "2026-01-01",
-        "contractEndDate": "2026-12-31",
-        "requestedFields": {
-          "legal_name": "NGUYEN VAN AN",
-          "passport_number": "M12345678"
-        }
-      }
-    ],
-    "workflowConstraints": [
-      {
-        "workflowId": "EXPIRY_RENEWAL",
-        "allowedSlotKeys": ["stay_expiry_date", "contract_end_date", "monthly_wage"]
-      }
-    ]
+    "instruction": "응웬반안 체류연장 준비해줘"
   }
 }
 ```
 
-| 필드 | 의미 |
+| 필드 | 규칙 |
 |---|---|
-| `instruction` | HR 원문 (데모는 가상 데이터, `***`/`OOO` 치환 없음) |
-| `workers[].requestedFields` | Agent가 요구한 field의 **Server 원본값** 맵 |
-| `workflowConstraints` | Workflow·slot allow-list |
-
-요청 `requestedFields`(값 맵)와 #74 응답용 `requestedFields([{key,sourceHint}])`는 **이름이 같고 역할이 다르다**.
-현재 Server #56 응답 DTO에는 후보 `requestedFields`가 없으므로 Analyses **응답 와이어에는 넣지 않는다**.
-
-## 응답 계약 (strict JSON)
+| `requestId` | 필수. 응답에 그대로 에코 |
+| `phase` | `"PLAN"` |
+| `analysisInput.instruction` | HR 발화 **원문만** (Intent 태그·코드 미부착, Issue #6) |
+| workers / requestedFieldKeys | **보내지 않음** (PLAN) |
+| `intentHint` | **없음** (폐기) |
+| attemptId / contractVersion / deadlineMs 등 | HTTP에 **없음** (Server 내부) |
 
-Server `FAIL_ON_UNKNOWN_PROPERTIES`에 맞춰 **알 수 없는 필드를 넣지 않는다**.
-`attemptId`·`evidence`·`caseSignals`·후보 `requestedFields`는 Analyses 응답에 포함하지 않는다.
+## 2) CONTEXT_REQUIRED 응답 (AI → Server)
 
 ```json
 {
-  "requestId": "...",
-  "outcome": "NEEDS_INFO | REVIEW_REQUIRED",
-  "candidates": [
-    {
-      "candidateRef": "...",
-      "workerRef": "...",
-      "workflowId": "EXPIRY_RENEWAL",
-      "extractedSlots": {},
-      "missingSlots": [],
-      "confidence": 0.92
-    }
-  ],
+  "requestId": "10000000-0000-0000-0000-000000000001",
+  "outcome": "CONTEXT_REQUIRED",
+  "contextRequirement": {
+    "detectedIntent": "EXPIRY_RENEWAL",
+    "confidence": 0.94,
+    "targetDisplayName": "응웬반안",
+    "extractedSlots": {},
+    "requiredFieldKeys": ["worker_id", "stay_expiry_date"]
+  },
+  "questions": [],
+  "candidates": [],
   "validationErrors": [],
   "versions": { "...": "..." },
   "providerAttemptCount": 1,
-  "latencyMs": 245
+  "latencyMs": 120
 }
 ```
 
-`validationErrors` 항목은 `{ "code", "field" }` 객체다.
+| 규칙 | 내용 |
+|---|---|
+| `requiredFieldKeys` | 비어 있으면 Server 거부. Knowledge canonical key만 (`worker_id` 포함) |
+| `questions` / `candidates` | 비움 |
+| `confidence` | 0.0 ~ 1.0 |
+
+## 3) ANALYZE 요청 (Server → AI)
+
+```json
+{
+  "requestId": "10000000-0000-0000-0000-000000000001",
+  "phase": "ANALYZE",
+  "analysisInput": {
+    "instruction": "응웬반안 체류연장 준비해줘",
+    "requestedFieldKeys": ["worker_id", "stay_expiry_date"],
+    "workers": [
+      {
+        "workerRef": "30000000-0000-0000-0000-000000000001",
+        "requestedFields": {
+          "worker_id": "30000000-0000-0000-0000-000000000001",
+          "stay_expiry_date": "2026-12-31"
+        }
+      }
+    ]
+  }
+}
+```
 
-## workflowId
+| 필드 | 규칙 |
+|---|---|
+| `requestedFieldKeys` | PLAN에서 Agent가 요청한 **전체** key (DB 미조회여도 목록 유지) |
+| `workers[].requestedFields` | Server가 **실제로 찾은 값만** |
+| DB 미조회 키 | `requestedFieldKeys − requestedFields.keys` → HR 질문 후보 |
+| MVP | Worker **1명** |
+| HTTP에 안 실림 | `extractedSlots`, `workflowConstraints`, attemptId, versions, deadline |
 
-두 형태를 모두 받는다.
+> 이슈 댓글의 ANALYZE `extractedSlots` 와이어 추가는 **최종 HTTP 계약에서 제외**됨  
+> (`AiRuntimeHttpRequest` 주석·직렬화 기준).
 
-| 형태 | 예 | 언제 |
-|---|---|---|
-| Intent형 | `EXPIRY_RENEWAL` | Server 계약 fixture / intention projection |
-| Catalog형 | `WF-STY-001` | knowledge Workflow Catalog |
+## 4) ANALYZE 응답
 
-- 요청 `workflowConstraints[].workflowId`에 **Intent형**이 있으면, 응답 candidate의
-  `workflowId`도 **같은 문자열**을 되돌려 Server 검증을 통과시킨다.
-- Catalog형 constraint면 Catalog형 id를 그대로 반환한다.
-- constraint가 비어 있으면 내부 분류 결과인 Catalog형 id(`WF-…`)를 반환한다.
+### NEEDS_INFO
 
-내부 Ambiguity/Workflow 검증은 항상 Catalog형 id로 수행한다.
+- `contextRequirement`: null  
+- `candidates`: []  
+- `questions`: **1개 이상** `{ "slotKey", "prompt" }`
 
-## Step 3 DB 조회용 키 (Analyses → Server)
+### REVIEW_REQUIRED
 
-candidate / slots에서 Server가 worker·company 조회에 쓰는 값:
+- `contextRequirement`: null  
+- `questions`: []  
+- `candidates`: **1개 이상** (기존 AiCandidate 필드)
 
-| 키 | 출처 | 필수 |
-|---|---|---|
-| `workerRef` / `extractedSlots.worker_id` | 요청 workers[].workerRef | 필수 |
-| `extractedSlots.company_id` | workers[].requestedFields.company_id | 권장 |
-| `extractedSlots.stay_expiry_date` | workers[] 또는 requestedFields | 체류 경로 |
-| `extractedSlots.contract_end_date` | workers[] 또는 requestedFields | 계약 경로 |
+```json
+{
+  "candidateRef": "candidate-1",
+  "workerRef": "30000000-0000-0000-0000-000000000001",
+  "workflowId": "EXPIRY_RENEWAL",
+  "extractedSlots": { "stay_expiry_date": "2026-12-31" },
+  "missingSlots": ["contract_end_date", "monthly_wage"],
+  "confidence": 0.92
+}
+```
 
-`worker_id`·`company_id`는 `allowedSlotKeys`와 무관하게 응답 slots에 유지한다.
+공통 응답 필드: `validationErrors`, `versions`, `providerAttemptCount`, `latencyMs`.
 
-슬롯 부족 시 `missingSlots`로 알린다. Server 재조회·재호출 키 규약(#74):
-[slot-refill-contract.md](slot-refill-contract.md)
+### versions (응답 필수)
 
-## 핸드셰이크 (#8)
+Server가 내부 요청의 `contractVersion` / `requiredKnowledgeVersion` 과  
+응답 `versions.contractVersion` / `versions.workflowCatalogVersion` 을 대조한다.  
+HTTP 요청에 version이 없어도 AI는 기본값 **`1.0.0` / `0.2.0`** 을 맞춰야 한다.
 
-`requestId` / `attemptId` / Bearer 토큰: [ai-runtime-handshake.md](ai-runtime-handshake.md)
+---
 
-Server는 `Authorization: Bearer <AI_RUNTIME_SERVICE_CREDENTIAL>`과
-`X-Request-Id`(=requestId), 선택적 `traceparent`를 보낸다.
-AI는 `FOWOCO_INTERNAL_API_TOKEN`과 동일 값으로 검증한다.
+## 우리(AI) 구현 상태
 
-## Versions
+| 항목 | Server HTTP | AI (`schemas` / `pipeline`) |
+|---|---|---|
+| `phase` PLAN/ANALYZE | 필수 | **반영** |
+| `CONTEXT_REQUIRED` | 있음 | **반영** |
+| `questions` | NEEDS_INFO | **반영** |
+| ANALYZE `requestedFieldKeys` | 있음 | **반영** |
+| workers 최소 필드 | workerRef + requestedFields | **반영** (추가 필드는 선택) |
+| attemptId 등 | HTTP 미전송 | **요청에서 제거** |
+| 슬롯 기준 | Knowledge | Ambiguity/Workflow catalog |
+| versions | 응답 필수 | `1.0.0` / `0.2.0` 고정 |
+
+## Intent 분류기
+
+| 설정 | 동작 |
+|---|---|
+| `FOWOCO_INTENT_MODEL_ENABLED=false` (기본) | `EXPIRY_RENEWAL` 고정 stub |
+| `FOWOCO_INTENT_MODEL_ENABLED=true` | HF BERT(+선택 A.X) 하이브리드 |
 
-`contractVersion=1.0.0`, `requiredKnowledgeVersion=0.2.0` 을 기본으로 둔다.
-응답 `versions.workflowCatalogVersion`은 서버 `task.workflow_catalog_version`과 맞출 값이다.
-MVP 응답의 model* 필드는 `stub`이다.
+필요 시: `pip install -e ".[intent]"`, `.env`에 `FOWOCO_HF_TOKEN` 또는 `HF_TOKEN`,  
+`FOWOCO_INTENT_BERT_MODEL_DIR=fowoco/klue-roberta-base-intent-classifier`.  
+로컬 CPU는 `FOWOCO_INTENT_ENABLE_AX=false` 권장.
 
 ## Fixtures
 
 | 파일 | 용도 |
 |---|---|
-| `examples/analyses/request_expiry_renewal.json` | Server 스타일 요청 |
-| `examples/analyses/response_needs_info.json` | 안내문 생성이 필요한 응답 예시 |
-| `examples/analyses/response_review_required.json` | HR 검토용 응답 예시 |
-
-## Knowledge 연동 (선택)
+| `examples/analyses/request_plan.json` | PLAN 요청 |
+| `examples/analyses/response_context_required.json` | CONTEXT_REQUIRED |
+| `examples/analyses/request_analyze.json` | ANALYZE 요청 |
+| `examples/analyses/response_needs_info.json` | NEEDS_INFO (신계약) |
+| `examples/analyses/response_review_required.json` | REVIEW_REQUIRED |
+| `examples/analyses/request_expiry_renewal.json` | **구계약** 참고용 (폐기 예정) |
 
-기본은 builtin 규칙. `FOWOCO_KNOWLEDGE_ENABLED=true` 이면
-`fowoco-knowledge` 패키지에서 required_slots·ambiguity·workflow catalog를 읽는다.
+## 핸드셰이크 (#8)
 
-```powershell
-pip install -e ..\knowledge\fowoco-knowledge
-$env:FOWOCO_KNOWLEDGE_ENABLED="true"
-# 선택: $env:FOWOCO_KNOWLEDGE_ROOT="..\knowledge\fowoco-knowledge"
-```
+[ai-runtime-handshake.md](ai-runtime-handshake.md) — Bearer, `X-Request-Id` = requestId.
diff --git a/examples/analyses/request_analyze.json b/examples/analyses/request_analyze.json
new file mode 100644
index 0000000..0a061a8
--- /dev/null
+++ b/examples/analyses/request_analyze.json
@@ -0,0 +1,20 @@
+{
+  "requestId": "10000000-0000-0000-0000-000000000001",
+  "phase": "ANALYZE",
+  "analysisInput": {
+    "instruction": "응웬반안 체류연장 준비해줘",
+    "requestedFieldKeys": [
+      "worker_id",
+      "stay_expiry_date"
+    ],
+    "workers": [
+      {
+        "workerRef": "30000000-0000-0000-0000-000000000001",
+        "requestedFields": {
+          "worker_id": "30000000-0000-0000-0000-000000000001",
+          "stay_expiry_date": "2026-12-31"
+        }
+      }
+    ]
+  }
+}
diff --git a/examples/analyses/request_plan.json b/examples/analyses/request_plan.json
new file mode 100644
index 0000000..11fa3a9
--- /dev/null
+++ b/examples/analyses/request_plan.json
@@ -0,0 +1,7 @@
+{
+  "requestId": "10000000-0000-0000-0000-000000000001",
+  "phase": "PLAN",
+  "analysisInput": {
+    "instruction": "응웬반안 체류연장 준비해줘"
+  }
+}
diff --git a/examples/analyses/response_context_required.json b/examples/analyses/response_context_required.json
new file mode 100644
index 0000000..fe68250
--- /dev/null
+++ b/examples/analyses/response_context_required.json
@@ -0,0 +1,29 @@
+{
+  "requestId": "10000000-0000-0000-0000-000000000001",
+  "outcome": "CONTEXT_REQUIRED",
+  "contextRequirement": {
+    "detectedIntent": "EXPIRY_RENEWAL",
+    "confidence": 0.94,
+    "targetDisplayName": "응웬반안",
+    "extractedSlots": {},
+    "requiredFieldKeys": [
+      "worker_id",
+      "stay_expiry_date"
+    ]
+  },
+  "questions": [],
+  "candidates": [],
+  "validationErrors": [],
+  "versions": {
+    "agentVersion": "0.1.0",
+    "modelProvider": "stub",
+    "modelName": "stub",
+    "modelVersion": "stub",
+    "promptVersion": "prompt-1",
+    "contextPackVersion": "0.2.0",
+    "workflowCatalogVersion": "0.2.0",
+    "contractVersion": "1.0.0"
+  },
+  "providerAttemptCount": 1,
+  "latencyMs": 120
+}
diff --git a/examples/analyses/response_needs_info.json b/examples/analyses/response_needs_info.json
index 4fa7787..01eae79 100644
--- a/examples/analyses/response_needs_info.json
+++ b/examples/analyses/response_needs_info.json
@@ -1,22 +1,14 @@
 {
   "requestId": "10000000-0000-0000-0000-000000000001",
   "outcome": "NEEDS_INFO",
-  "candidates": [
+  "contextRequirement": null,
+  "questions": [
     {
-      "candidateRef": "candidate-1",
-      "workerRef": "30000000-0000-0000-0000-000000000001",
-      "workflowId": "EXPIRY_RENEWAL",
-      "extractedSlots": {
-        "stay_expiry_date": "2026-12-31",
-        "worker_id": "30000000-0000-0000-0000-000000000001"
-      },
-      "missingSlots": [
-        "contract_end_date",
-        "monthly_wage"
-      ],
-      "confidence": 0.8
+      "slotKey": "monthly_wage",
+      "prompt": "월 급여를 입력해 주세요."
     }
   ],
+  "candidates": [],
   "validationErrors": [],
   "versions": {
     "agentVersion": "0.1.0",
@@ -29,5 +21,5 @@
     "contractVersion": "1.0.0"
   },
   "providerAttemptCount": 1,
-  "latencyMs": 12
+  "latencyMs": 180
 }
diff --git a/examples/analyses/response_review_required.json b/examples/analyses/response_review_required.json
index 895d176..a8fd71e 100644
--- a/examples/analyses/response_review_required.json
+++ b/examples/analyses/response_review_required.json
@@ -1,17 +1,21 @@
 {
   "requestId": "10000000-0000-0000-0000-000000000001",
   "outcome": "REVIEW_REQUIRED",
+  "contextRequirement": null,
+  "questions": [],
   "candidates": [
     {
       "candidateRef": "candidate-1",
       "workerRef": "30000000-0000-0000-0000-000000000001",
       "workflowId": "EXPIRY_RENEWAL",
       "extractedSlots": {
-        "worker_id": "30000000-0000-0000-0000-000000000001",
         "stay_expiry_date": "2026-12-31"
       },
-      "missingSlots": [],
-      "confidence": 0.9
+      "missingSlots": [
+        "contract_end_date",
+        "monthly_wage"
+      ],
+      "confidence": 0.92
     }
   ],
   "validationErrors": [],
@@ -26,5 +30,5 @@
     "contractVersion": "1.0.0"
   },
   "providerAttemptCount": 1,
-  "latencyMs": 10
+  "latencyMs": 245
 }
diff --git a/tests/api/test_analyses_endpoint.py b/tests/api/test_analyses_endpoint.py
index 40eb7f0..0cc3fb9 100644
--- a/tests/api/test_analyses_endpoint.py
+++ b/tests/api/test_analyses_endpoint.py
@@ -1,4 +1,4 @@
-"""POST /internal/v1/analyses 엔드포인트 테스트."""
+# POST /internal/v1/analyses — PLAN / ANALYZE 계약 테스트
 
 import pytest
 from httpx import ASGITransport, AsyncClient
@@ -8,152 +8,135 @@
 ANALYSES_PATH = "/internal/v1/analyses"
 
 
-def _make_request(
-    instruction: str = "체류기간 연장 준비해줘",
+def _plan_body(instruction: str = "응웬반안 체류연장 준비해줘") -> dict:
+    return {
+        "requestId": "10000000-0000-0000-0000-000000000001",
+        "phase": "PLAN",
+        "analysisInput": {"instruction": instruction},
+    }
+
+
+def _analyze_body(
     *,
+    instruction: str = "응웬반안 체류연장 준비해줘",
     worker_ref: str = "30000000-0000-0000-0000-000000000001",
-    display_name: str = "테스트근로자",
-    workflow_constraints: list[dict] | None = None,
-    stay_expiry_date: str | None = "2026-12-31",
+    requested_field_keys: list[str] | None = None,
     requested_fields: dict[str, str] | None = None,
 ) -> dict:
-    workers = [
-        {
-            "workerRef": worker_ref,
-            "displayName": display_name,
-            "preferredLanguage": "vi",
-            "workStatus": "ACTIVE",
-            "requestedFields": requested_fields or {},
-        }
-    ]
-    if stay_expiry_date:
-        workers[0]["stayExpiryDate"] = stay_expiry_date
     return {
         "requestId": "10000000-0000-0000-0000-000000000001",
-        "attemptId": "20000000-0000-0000-0000-000000000001",
-        "contractVersion": "1.0.0",
-        "requiredKnowledgeVersion": "0.2.0",
-        "deadlineMs": 10000,
+        "phase": "ANALYZE",
         "analysisInput": {
             "instruction": instruction,
-            "workers": workers,
-            "workflowConstraints": workflow_constraints or [],
+            "requestedFieldKeys": requested_field_keys
+            or ["worker_id", "stay_expiry_date"],
+            "workers": [
+                {
+                    "workerRef": worker_ref,
+                    "requestedFields": requested_fields
+                    or {
+                        "worker_id": worker_ref,
+                        "stay_expiry_date": "2026-12-31",
+                    },
+                }
+            ],
         },
     }
 
 
 @pytest.mark.asyncio
-async def test_analyses_returns_review_required_for_complete_request() -> None:
-    body = _make_request(
-        "WRK-012 체류기간 연장 준비 2026-12-31",
-        worker_ref="WRK-012",
-        display_name="WRK-012",
-        workflow_constraints=[
-            {"workflowId": "WF-STY-001", "allowedSlotKeys": ["stay_expiry_date", "worker_id"]},
-        ],
-    )
+async def test_plan_returns_context_required() -> None:
     async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
-        resp = await client.post(ANALYSES_PATH, json=body)
+        resp = await client.post(ANALYSES_PATH, json=_plan_body())
 
     assert resp.status_code == 200
     data = resp.json()
     assert data["requestId"] == "10000000-0000-0000-0000-000000000001"
-    assert data["outcome"] in ("REVIEW_REQUIRED", "NEEDS_INFO")
+    assert data["outcome"] == "CONTEXT_REQUIRED"
+    assert data["candidates"] == []
+    assert data["questions"] == []
+    ctx = data["contextRequirement"]
+    assert ctx["detectedIntent"] == "EXPIRY_RENEWAL"
+    assert ctx["targetDisplayName"] == "응웬반안"
+    assert "stay_expiry_date" in ctx["requiredFieldKeys"]
+    assert "worker_id" in ctx["requiredFieldKeys"]
+    assert data["versions"]["contractVersion"] == "1.0.0"
+    assert data["versions"]["workflowCatalogVersion"] == "0.2.0"
     assert "attemptId" not in data
-    assert len(data["candidates"]) == 1
-    candidate = data["candidates"][0]
-    assert candidate["workerRef"] == "WRK-012"
-    assert candidate["workflowId"] == "WF-STY-001"
-    assert candidate["confidence"] > 0
-    assert "requestedFields" not in candidate
-    assert "evidence" not in candidate
-    assert "caseSignals" not in candidate
-    assert "agentVersion" in data["versions"]
 
 
 @pytest.mark.asyncio
-async def test_analyses_echoes_intent_style_workflow_constraint() -> None:
-    """Server 계약 fixture처럼 Intent형 workflowId를 요청하면 응답에도 동일 id를 쓴다."""
-    body = _make_request(
-        "체류기간 연장 준비해줘",
-        workflow_constraints=[
-            {
-                "workflowId": "EXPIRY_RENEWAL",
-                "allowedSlotKeys": [
-                    "stay_expiry_date",
-                    "contract_end_date",
-                    "monthly_wage",
-                ],
-            }
-        ],
-    )
+async def test_analyze_returns_review_required_when_slots_filled() -> None:
     async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
-        resp = await client.post(ANALYSES_PATH, json=body)
+        resp = await client.post(ANALYSES_PATH, json=_analyze_body())
 
     assert resp.status_code == 200
     data = resp.json()
-    assert data["candidates"][0]["workflowId"] == "EXPIRY_RENEWAL"
-    assert "stay_expiry_date" in data["candidates"][0]["extractedSlots"]
+    assert data["outcome"] == "REVIEW_REQUIRED"
+    assert data["contextRequirement"] is None
+    assert data["questions"] == []
+    assert len(data["candidates"]) == 1
+    candidate = data["candidates"][0]
+    assert candidate["workerRef"] == "30000000-0000-0000-0000-000000000001"
+    assert candidate["workflowId"] == "EXPIRY_RENEWAL"
+    assert candidate["extractedSlots"]["stay_expiry_date"] == "2026-12-31"
+    assert candidate["extractedSlots"]["worker_id"] == (
+        "30000000-0000-0000-0000-000000000001"
+    )
 
 
 @pytest.mark.asyncio
-async def test_analyses_accepts_server_requested_fields_map() -> None:
-    body = _make_request(
-        "응웬반안 체류연장 준비해줘",
-        display_name="응웬반안",
-        requested_fields={
-            "legal_name": "NGUYEN VAN AN",
-            "passport_number": "M12345678",
-        },
-        workflow_constraints=[
-            {
-                "workflowId": "EXPIRY_RENEWAL",
-                "allowedSlotKeys": [
-                    "stay_expiry_date",
-                    "contract_end_date",
-                    "monthly_wage",
-                ],
-            }
-        ],
+async def test_analyze_returns_needs_info_when_db_fields_missing() -> None:
+    body = _analyze_body(
+        requested_field_keys=["worker_id", "stay_expiry_date"],
+        requested_fields={"worker_id": "30000000-0000-0000-0000-000000000001"},
     )
     async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
         resp = await client.post(ANALYSES_PATH, json=body)
 
     assert resp.status_code == 200
-    slots = resp.json()["candidates"][0]["extractedSlots"]
-    assert slots["worker_id"] == "30000000-0000-0000-0000-000000000001"
-    assert "stay_expiry_date" in slots
+    data = resp.json()
+    assert data["outcome"] == "NEEDS_INFO"
+    assert data["candidates"] == []
+    assert data["contextRequirement"] is None
+    keys = {q["slotKey"] for q in data["questions"]}
+    assert "stay_expiry_date" in keys
+    assert all("prompt" in q for q in data["questions"])
 
 
 @pytest.mark.asyncio
-async def test_analyses_returns_needs_info_when_slots_missing() -> None:
-    body = _make_request(
-        "서류 요청해줘",
-        stay_expiry_date=None,
+async def test_analyze_mvp_uses_first_worker_only() -> None:
+    body = _analyze_body()
+    body["analysisInput"]["workers"].append(
+        {
+            "workerRef": "30000000-0000-0000-0000-000000000002",
+            "requestedFields": {"stay_expiry_date": "2026-11-30"},
+        }
     )
     async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
         resp = await client.post(ANALYSES_PATH, json=body)
 
     assert resp.status_code == 200
     data = resp.json()
-    assert data["outcome"] == "NEEDS_INFO"
+    assert data["outcome"] == "REVIEW_REQUIRED"
     assert len(data["candidates"]) == 1
-    assert len(data["candidates"][0]["missingSlots"]) > 0
+    assert data["candidates"][0]["workerRef"] == (
+        "30000000-0000-0000-0000-000000000001"
+    )
 
 
 @pytest.mark.asyncio
-async def test_analyses_fixed_intent_treats_unrelated_as_expiry_renewal() -> None:
-    """기본 Intent 고정 모드에서는 무관 문장도 EXPIRY_RENEWAL로 본다."""
-    body = _make_request("오늘 날씨 어때?")
+async def test_plan_fixed_intent_even_for_unrelated_instruction() -> None:
     async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
-        resp = await client.post(ANALYSES_PATH, json=body)
+        resp = await client.post(
+            ANALYSES_PATH, json=_plan_body("오늘 날씨 어때?")
+        )
 
     assert resp.status_code == 200
     data = resp.json()
-    candidate = data["candidates"][0]
-    assert candidate["confidence"] == 1.0
-    assert candidate["workflowId"] in {"EXPIRY_RENEWAL", "WF-STY-001", "WF-CON-001"}
-    assert data["outcome"] in {"NEEDS_INFO", "REVIEW_REQUIRED"}
+    assert data["outcome"] == "CONTEXT_REQUIRED"
+    assert data["contextRequirement"]["detectedIntent"] == "EXPIRY_RENEWAL"
+    assert data["contextRequirement"]["confidence"] == 1.0
 
 
 @pytest.mark.asyncio
@@ -167,56 +150,26 @@ async def test_analyses_endpoint_in_openapi() -> None:
 
 
 @pytest.mark.asyncio
-async def test_analyses_multiple_workers() -> None:
+async def test_analyses_rejects_legacy_masked_input() -> None:
     body = {
-        "requestId": "10000000-0000-0000-0000-000000000099",
-        "attemptId": "20000000-0000-0000-0000-000000000099",
-        "contractVersion": "1.0.0",
-        "requiredKnowledgeVersion": "0.2.0",
-        "deadlineMs": 10000,
-        "analysisInput": {
-            "instruction": "체류기간 연장 준비",
-            "workers": [
-                {
-                    "workerRef": "w-001",
-                    "displayName": "근로자1",
-                    "preferredLanguage": "vi",
-                    "workStatus": "ACTIVE",
-                    "stayExpiryDate": "2026-12-31",
-                    "requestedFields": {},
-                },
-                {
-                    "workerRef": "w-002",
-                    "displayName": "근로자2",
-                    "preferredLanguage": "th",
-                    "workStatus": "ACTIVE",
-                    "stayExpiryDate": "2026-11-30",
-                    "requestedFields": {},
-                },
-            ],
-            "workflowConstraints": [],
+        "requestId": "10000000-0000-0000-0000-000000000001",
+        "phase": "PLAN",
+        "maskedInput": {
+            "maskedInstruction": "체류연장",
+            "workers": [],
         },
     }
     async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
         resp = await client.post(ANALYSES_PATH, json=body)
 
-    assert resp.status_code == 200
-    data = resp.json()
-    assert len(data["candidates"]) == 2
-    refs = {c["workerRef"] for c in data["candidates"]}
-    assert refs == {"w-001", "w-002"}
+    assert resp.status_code == 422
 
 
 @pytest.mark.asyncio
-async def test_analyses_rejects_legacy_masked_input() -> None:
+async def test_analyses_requires_phase() -> None:
     body = {
         "requestId": "10000000-0000-0000-0000-000000000001",
-        "attemptId": "20000000-0000-0000-0000-000000000001",
-        "maskedInput": {
-            "maskedInstruction": "체류연장",
-            "workers": [],
-            "workflowConstraints": [],
-        },
+        "analysisInput": {"instruction": "체류연장"},
     }
     async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
         resp = await client.post(ANALYSES_PATH, json=body)
diff --git a/tests/api/test_internal_handshake.py b/tests/api/test_internal_handshake.py
index 4a41fd3..5036a52 100644
--- a/tests/api/test_internal_handshake.py
+++ b/tests/api/test_internal_handshake.py
@@ -15,24 +15,11 @@ def test_requested_fields_for_api_maps_source_hints() -> None:
     assert by_key["passport_number"] == "DOCUMENT_OCR"
 
 
-def _analysis_body(*, request_id: str = "req-open", attempt_id: str = "att-1") -> dict:
+def _analysis_body(*, request_id: str = "req-open") -> dict:
     return {
         "requestId": request_id,
-        "attemptId": attempt_id,
-        "analysisInput": {
-            "instruction": "체류연장",
-            "workers": [
-                {
-                    "workerRef": "worker-001",
-                    "displayName": "테스트",
-                    "stayExpiryDate": "2026-12-31",
-                    "requestedFields": {},
-                }
-            ],
-            "workflowConstraints": [
-                {"workflowId": "EXPIRY_RENEWAL", "allowedSlotKeys": []}
-            ],
-        },
+        "phase": "PLAN",
+        "analysisInput": {"instruction": "체류연장"},
     }
 
 
@@ -57,19 +44,15 @@ def test_internal_api_requires_bearer_when_token_set(monkeypatch) -> None:
         "/internal/v1/analyses",
         json={
             "requestId": "req-auth",
-            "attemptId": "att-2",
-            "analysisInput": {
-                "instruction": "체류연장",
-                "workers": [],
-                "workflowConstraints": [],
-            },
+            "phase": "PLAN",
+            "analysisInput": {"instruction": "체류연장"},
         },
     )
     assert denied.status_code == 401
     ok = client.post(
         "/internal/v1/analyses",
         headers={"Authorization": "Bearer secret-token"},
-        json=_analysis_body(request_id="req-auth", attempt_id="att-2"),
+        json=_analysis_body(request_id="req-auth"),
     )
     assert ok.status_code == 200
     get_settings.cache_clear()

From 43cfbf9fc9a551c254bc50fdf982ee1d40e4c0d1 Mon Sep 17 00:00:00 2001
From: HWIYA <dlgnl117@gmail.com>
Date: Wed, 5 Aug 2026 14:55:19 +0900
Subject: [PATCH 3/4] =?UTF-8?q?test:=20Intent=20opt-in=C2=B7=ED=8F=B4?=
 =?UTF-8?q?=EB=B0=B1=C2=B7=EA=B0=80=EB=93=9C=EB=A0=88=EC=9D=BC=20=EC=BB=A4?=
 =?UTF-8?q?=EB=B2=84=EB=A6=AC=EC=A7=80=20=EB=B3=B4=EA=B0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Windows용 BERT-only `.[intent]`와 A.X용 `.[intent-ax]` extras 분리.
---
 pyproject.toml                     |  6 +++-
 tests/agents/test_intent_hybrid.py | 49 ++++++++++++++++++++++++++++++
 2 files changed, 54 insertions(+), 1 deletion(-)

diff --git a/pyproject.toml b/pyproject.toml
index ed16622..eb068ba 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -33,11 +33,15 @@ dev = [
 ]
 # 로컬: pip install -e ../knowledge/fowoco-knowledge
 knowledge = []
-# HF Intent (BERT 필수, A.X는 FOWOCO_INTENT_ENABLE_AX=true 시 peft·bitsandbytes)
+# HF Intent BERT (Windows CPU 권장: pip install -e ".[intent]")
 intent = [
   "torch>=2.2,<3",
   "transformers>=4.46,<5",
   "accelerate>=0.34,<2",
+]
+# A.X 경로 (Linux/CUDA 권장). Windows에선 bitsandbytes 설치 실패 흔함
+intent-ax = [
+  "fowoco-ai[intent]",
   "peft>=0.13,<1",
   "bitsandbytes>=0.44,<1",
 ]
diff --git a/tests/agents/test_intent_hybrid.py b/tests/agents/test_intent_hybrid.py
index e165442..eedf66c 100644
--- a/tests/agents/test_intent_hybrid.py
+++ b/tests/agents/test_intent_hybrid.py
@@ -1,5 +1,6 @@
 # HF Intent 에이전트·대표 Intent 선택 단위 테스트
 
+from app.agents.intent.guardrail import HRRoutingGuardrail
 from app.agents.intent.hybrid import HybridIntentPrediction
 from app.agents.intent.service import (
     FixedExpiryRenewalIntentAgent,
@@ -7,6 +8,7 @@
     _primary_intent,
     build_intent_agent,
 )
+from app.core.config import get_settings
 
 
 # 점수 있을 때 최고 점수 Intent 선택
@@ -51,3 +53,50 @@ def predict(self, instruction: str) -> HybridIntentPrediction:
     assert result.intent == "EXPIRY_RENEWAL"
     assert result.confidence == 0.93
     assert result.workflow_id == "WF-STY-001"
+
+
+# INTENT_MODEL_ENABLED=true → HybridHfIntentAgent
+def test_build_intent_agent_enabled_returns_hybrid(monkeypatch) -> None:
+    monkeypatch.setenv("FOWOCO_INTENT_MODEL_ENABLED", "true")
+    get_settings.cache_clear()
+    try:
+        agent = build_intent_agent()
+        assert isinstance(agent, HybridHfIntentAgent)
+    finally:
+        get_settings.cache_clear()
+
+
+# 파이프라인 로드 실패 시 재갱신 고정 폴백
+def test_hybrid_load_failure_falls_back_to_fixed(monkeypatch) -> None:
+    def _boom(*_args, **_kwargs):  # noqa: ANN002, ANN003
+        raise RuntimeError("model unavailable")
+
+    monkeypatch.setattr("app.agents.intent.hybrid.HybridIntentPipeline", _boom)
+    agent = HybridHfIntentAgent()
+    result = agent.classify("체류연장 준비해줘")
+    assert result.intent == "EXPIRY_RENEWAL"
+    assert agent._load_error is not None
+
+
+# margin 통과 시 BERT 유지
+def test_guardrail_pass_bert_when_confident() -> None:
+    gate = HRRoutingGuardrail(margin_threshold=0.76)
+    out = gate.should_route_to_ax(
+        "응웬반안 체류연장 준비",
+        {"EXPIRY_RENEWAL": 0.92, "DOCUMENT_REQUEST": 0.05},
+        margin=0.87,
+    )
+    assert out.should_route is False
+    assert out.category == "Pass_BERT"
+
+
+# 서류 키워드면 A.X 라우팅
+def test_guardrail_routes_on_document_keyword() -> None:
+    gate = HRRoutingGuardrail()
+    out = gate.should_route_to_ax(
+        "신청서 서류 챙겨줘",
+        {"DOCUMENT_REQUEST": 0.8},
+        margin=0.9,
+    )
+    assert out.should_route is True
+    assert out.category == "Rule_Document"

From 65b62532c7f5583d8e5c82a4654404c09d95c611 Mon Sep 17 00:00:00 2001
From: HWIYA <dlgnl117@gmail.com>
Date: Wed, 5 Aug 2026 14:55:20 +0900
Subject: [PATCH 4/4] =?UTF-8?q?test:=20analyses=20=ED=8C=8C=EC=9D=B4?=
 =?UTF-8?q?=ED=94=84=EB=9D=BC=EC=9D=B8=C2=B7fixture=20=EC=8A=A4=ED=82=A4?=
 =?UTF-8?q?=EB=A7=88=20=EC=A0=95=ED=95=A9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

구계약 orphan fixture 제거. REVIEW_REQUIRED missingSlots는 빈 배열로 맞춤.
---
 docs/analyses-contract.md                     |  20 ++-
 examples/analyses/request_expiry_renewal.json |  38 ------
 .../analyses/response_requested_fields.json   |  29 -----
 .../analyses/response_review_required.json    |   9 +-
 tests/agents/test_analysis_pipeline.py        | 116 ++++++++++++++++++
 tests/contracts/test_analyses_fixtures.py     |  41 +++++++
 6 files changed, 175 insertions(+), 78 deletions(-)
 delete mode 100644 examples/analyses/request_expiry_renewal.json
 delete mode 100644 examples/analyses/response_requested_fields.json
 create mode 100644 tests/agents/test_analysis_pipeline.py
 create mode 100644 tests/contracts/test_analyses_fixtures.py

diff --git a/docs/analyses-contract.md b/docs/analyses-contract.md
index c1cec31..dd56d6c 100644
--- a/docs/analyses-contract.md
+++ b/docs/analyses-contract.md
@@ -123,12 +123,19 @@ ANALYZE → NEEDS_INFO (questions) | REVIEW_REQUIRED (candidates)
   "candidateRef": "candidate-1",
   "workerRef": "30000000-0000-0000-0000-000000000001",
   "workflowId": "EXPIRY_RENEWAL",
-  "extractedSlots": { "stay_expiry_date": "2026-12-31" },
-  "missingSlots": ["contract_end_date", "monthly_wage"],
+  "extractedSlots": {
+    "worker_id": "30000000-0000-0000-0000-000000000001",
+    "stay_expiry_date": "2026-12-31",
+    "full_name": "NGUYEN VAN AN"
+  },
+  "missingSlots": [],
   "confidence": 0.92
 }
 ```
 
+`missingSlots`는 REVIEW 직전 필수 슬롯이 모두 채워졌을 때 **빈 배열**이다.  
+남은 HR 입력은 `NEEDS_INFO.questions`로 보낸다.
+
 공통 응답 필드: `validationErrors`, `versions`, `providerAttemptCount`, `latencyMs`.
 
 ### versions (응답 필수)
@@ -159,9 +166,11 @@ HTTP 요청에 version이 없어도 AI는 기본값 **`1.0.0` / `0.2.0`** 을 
 | `FOWOCO_INTENT_MODEL_ENABLED=false` (기본) | `EXPIRY_RENEWAL` 고정 stub |
 | `FOWOCO_INTENT_MODEL_ENABLED=true` | HF BERT(+선택 A.X) 하이브리드 |
 
-필요 시: `pip install -e ".[intent]"`, `.env`에 `FOWOCO_HF_TOKEN` 또는 `HF_TOKEN`,  
+필요 시 BERT만: `pip install -e ".[intent]"` (Windows CPU 권장).  
+A.X까지: `pip install -e ".[intent-ax]"` (Linux/CUDA; Windows에선 `bitsandbytes` 실패 흔함).  
+`.env`에 `FOWOCO_HF_TOKEN` 또는 `HF_TOKEN`,  
 `FOWOCO_INTENT_BERT_MODEL_DIR=fowoco/klue-roberta-base-intent-classifier`.  
-로컬 CPU는 `FOWOCO_INTENT_ENABLE_AX=false` 권장.
+로컬 CPU는 `FOWOCO_INTENT_ENABLE_AX=false` 권장. `.env.example`은 두지 않음(로컬 `.env`만).
 
 ## Fixtures
 
@@ -170,9 +179,8 @@ HTTP 요청에 version이 없어도 AI는 기본값 **`1.0.0` / `0.2.0`** 을 
 | `examples/analyses/request_plan.json` | PLAN 요청 |
 | `examples/analyses/response_context_required.json` | CONTEXT_REQUIRED |
 | `examples/analyses/request_analyze.json` | ANALYZE 요청 |
-| `examples/analyses/response_needs_info.json` | NEEDS_INFO (신계약) |
+| `examples/analyses/response_needs_info.json` | NEEDS_INFO |
 | `examples/analyses/response_review_required.json` | REVIEW_REQUIRED |
-| `examples/analyses/request_expiry_renewal.json` | **구계약** 참고용 (폐기 예정) |
 
 ## 핸드셰이크 (#8)
 
diff --git a/examples/analyses/request_expiry_renewal.json b/examples/analyses/request_expiry_renewal.json
deleted file mode 100644
index 2eb86da..0000000
--- a/examples/analyses/request_expiry_renewal.json
+++ /dev/null
@@ -1,38 +0,0 @@
-{
-  "requestId": "10000000-0000-0000-0000-000000000001",
-  "attemptId": "20000000-0000-0000-0000-000000000001",
-  "contractVersion": "1.0.0",
-  "requiredKnowledgeVersion": "0.2.0",
-  "deadlineMs": 10000,
-  "analysisInput": {
-    "instruction": "가상 근로자 응웬반안(010-1234-5678)의 체류연장 준비",
-    "workers": [
-      {
-        "workerRef": "30000000-0000-0000-0000-000000000001",
-        "displayName": "응웬반안",
-        "nationalityCode": "VN",
-        "preferredLanguage": "vi",
-        "workStatus": "ACTIVE",
-        "stayExpiryDate": "2026-12-31",
-        "contractStartDate": "2026-01-01",
-        "contractEndDate": "2026-12-31",
-        "requestedFields": {
-          "legal_name": "NGUYEN VAN AN",
-          "passport_number": "M12345678",
-          "phone": "010-1234-5678",
-          "email": "worker@example.com"
-        }
-      }
-    ],
-    "workflowConstraints": [
-      {
-        "workflowId": "EXPIRY_RENEWAL",
-        "allowedSlotKeys": [
-          "stay_expiry_date",
-          "contract_end_date",
-          "monthly_wage"
-        ]
-      }
-    ]
-  }
-}
diff --git a/examples/analyses/response_requested_fields.json b/examples/analyses/response_requested_fields.json
deleted file mode 100644
index 95f77a3..0000000
--- a/examples/analyses/response_requested_fields.json
+++ /dev/null
@@ -1,29 +0,0 @@
-{
-  "requestId": "10000000-0000-0000-0000-000000000001",
-  "outcome": "NEEDS_INFO",
-  "candidates": [
-    {
-      "candidateRef": "candidate-demo01",
-      "workerRef": "30000000-0000-0000-0000-000000000001",
-      "workflowId": "EXPIRY_RENEWAL",
-      "extractedSlots": {
-        "worker_id": "30000000-0000-0000-0000-000000000001"
-      },
-      "missingSlots": ["stay_expiry_date", "wage"],
-      "confidence": 0.72
-    }
-  ],
-  "validationErrors": [],
-  "versions": {
-    "agentVersion": "0.1.0",
-    "modelProvider": "stub",
-    "modelName": "stub",
-    "modelVersion": "stub",
-    "promptVersion": "prompt-1",
-    "contextPackVersion": "0.2.0",
-    "workflowCatalogVersion": "0.2.0",
-    "contractVersion": "1.0.0"
-  },
-  "providerAttemptCount": 1,
-  "latencyMs": 12
-}
diff --git a/examples/analyses/response_review_required.json b/examples/analyses/response_review_required.json
index a8fd71e..52fb79b 100644
--- a/examples/analyses/response_review_required.json
+++ b/examples/analyses/response_review_required.json
@@ -9,12 +9,11 @@
       "workerRef": "30000000-0000-0000-0000-000000000001",
       "workflowId": "EXPIRY_RENEWAL",
       "extractedSlots": {
-        "stay_expiry_date": "2026-12-31"
+        "worker_id": "30000000-0000-0000-0000-000000000001",
+        "stay_expiry_date": "2026-12-31",
+        "full_name": "NGUYEN VAN AN"
       },
-      "missingSlots": [
-        "contract_end_date",
-        "monthly_wage"
-      ],
+      "missingSlots": [],
       "confidence": 0.92
     }
   ],
diff --git a/tests/agents/test_analysis_pipeline.py b/tests/agents/test_analysis_pipeline.py
new file mode 100644
index 0000000..1bdd29d
--- /dev/null
+++ b/tests/agents/test_analysis_pipeline.py
@@ -0,0 +1,116 @@
+# AnalysisPipeline PLAN/ANALYZE 단위 테스트
+
+from uuid import uuid4
+
+from app.agents.intent.service import IntentResult
+from app.agents.pipeline import AnalysisPipeline
+from app.api.schemas.analyses import AnalysisInput, AnalysisRequest, WorkerContext
+
+
+# IntentClassifier Protocol용 고정 분류기
+class _FakeIntent:
+    def __init__(self, *, intent: str, confidence: float = 0.9, workflow_id: str = "") -> None:
+        self.intent = intent
+        self.confidence = confidence
+        self.workflow_id = workflow_id
+
+    def classify(
+        self,
+        instruction: str,
+        *,
+        workflow_constraints: list[str] | None = None,
+    ) -> IntentResult:
+        del instruction, workflow_constraints
+        return IntentResult(
+            intent=self.intent,
+            confidence=self.confidence,
+            workflow_id=self.workflow_id,
+            extracted_slots={},
+        )
+
+
+# PLAN → CONTEXT_REQUIRED + requiredFieldKeys
+def test_plan_returns_context_required_for_expiry() -> None:
+    pipe = AnalysisPipeline(
+        intent_agent=_FakeIntent(intent="EXPIRY_RENEWAL", workflow_id="WF-STY-001")
+    )
+    req = AnalysisRequest(
+        requestId=str(uuid4()),
+        phase="PLAN",
+        analysisInput=AnalysisInput(instruction="응웬반안 체류연장 준비해줘"),
+    )
+    res = pipe.run(req)
+    assert res.outcome == "CONTEXT_REQUIRED"
+    assert res.context_requirement is not None
+    assert res.context_requirement.detected_intent == "EXPIRY_RENEWAL"
+    assert "worker_id" in res.context_requirement.required_field_keys
+
+
+# OUT_OF_SCOPE PLAN은 worker_id만 요청
+def test_plan_out_of_scope_requests_worker_id_only() -> None:
+    pipe = AnalysisPipeline(intent_agent=_FakeIntent(intent="OUT_OF_SCOPE", confidence=0.7))
+    req = AnalysisRequest(
+        requestId=str(uuid4()),
+        phase="PLAN",
+        analysisInput=AnalysisInput(instruction="오늘 날씨 어때"),
+    )
+    res = pipe.run(req)
+    assert res.outcome == "CONTEXT_REQUIRED"
+    assert res.context_requirement is not None
+    assert res.context_requirement.required_field_keys == ["worker_id"]
+
+
+# ANALYZE workers 없으면 NEEDS_INFO
+def test_analyze_without_workers_needs_info() -> None:
+    pipe = AnalysisPipeline(
+        intent_agent=_FakeIntent(intent="EXPIRY_RENEWAL", workflow_id="WF-STY-001")
+    )
+    req = AnalysisRequest(
+        requestId=str(uuid4()),
+        phase="ANALYZE",
+        analysisInput=AnalysisInput(
+            instruction="체류연장",
+            requestedFieldKeys=["worker_id", "stay_expiry_date"],
+            workers=[],
+        ),
+    )
+    res = pipe.run(req)
+    assert res.outcome == "NEEDS_INFO"
+    assert any(q.slot_key == "worker_id" for q in res.questions)
+
+
+# ANALYZE 슬롯 충족 시 REVIEW_REQUIRED + missingSlots 빈 목록
+def test_analyze_filled_slots_review_required() -> None:
+    pipe = AnalysisPipeline(
+        intent_agent=_FakeIntent(
+            intent="EXPIRY_RENEWAL", confidence=0.91, workflow_id="WF-STY-001"
+        )
+    )
+    worker = WorkerContext(
+        workerRef="30000000-0000-0000-0000-000000000001",
+        requestedFields={
+            "worker_id": "30000000-0000-0000-0000-000000000001",
+            "stay_expiry_date": "2026-12-31",
+            "contract_end_date": "2026-12-31",
+            "legal_name": "NGUYEN VAN AN",
+            "passport_number": "M12345678",
+            "alien_registration_number": "123456-7890123",
+            "date_of_birth": "1990-01-01",
+            "nationality": "VN",
+            "full_name": "NGUYEN VAN AN",
+        },
+    )
+    req = AnalysisRequest(
+        requestId=str(uuid4()),
+        phase="ANALYZE",
+        analysisInput=AnalysisInput(
+            instruction="응웬반안 체류연장",
+            requestedFieldKeys=list(worker.requested_fields.keys()),
+            workers=[worker],
+        ),
+    )
+    res = pipe.run(req)
+    assert res.outcome == "REVIEW_REQUIRED"
+    assert len(res.candidates) == 1
+    assert res.candidates[0].missing_slots == []
+    assert res.candidates[0].workflow_id == "EXPIRY_RENEWAL"
diff --git a/tests/contracts/test_analyses_fixtures.py b/tests/contracts/test_analyses_fixtures.py
new file mode 100644
index 0000000..c3b7fb3
--- /dev/null
+++ b/tests/contracts/test_analyses_fixtures.py
@@ -0,0 +1,41 @@
+# examples/analyses fixture ↔ Pydantic 스키마 roundtrip
+
+from pathlib import Path
+
+import pytest
+
+from app.api.schemas.analyses import AnalysisRequest, AnalysisResponse
+
+_FIXTURES = Path(__file__).resolve().parents[2] / "examples" / "analyses"
+
+_REQUESTS = ("request_plan.json", "request_analyze.json")
+_RESPONSES = (
+    "response_context_required.json",
+    "response_needs_info.json",
+    "response_review_required.json",
+)
+
+
+# PLAN/ANALYZE 요청 fixture 역직렬화
+@pytest.mark.parametrize("name", _REQUESTS)
+def test_analysis_request_fixtures_parse(name: str) -> None:
+    raw = (_FIXTURES / name).read_text(encoding="utf-8")
+    req = AnalysisRequest.model_validate_json(raw)
+    assert req.phase in {"PLAN", "ANALYZE"}
+    assert req.analysis_input.instruction
+
+
+# 성공 outcome 응답 fixture 역직렬화
+@pytest.mark.parametrize("name", _RESPONSES)
+def test_analysis_response_fixtures_parse(name: str) -> None:
+    raw = (_FIXTURES / name).read_text(encoding="utf-8")
+    res = AnalysisResponse.model_validate_json(raw)
+    assert res.outcome in {"CONTEXT_REQUIRED", "NEEDS_INFO", "REVIEW_REQUIRED"}
+    if res.outcome == "REVIEW_REQUIRED":
+        assert res.candidates
+        assert res.candidates[0].missing_slots == []
+    if res.outcome == "NEEDS_INFO":
+        assert res.questions
+        assert res.candidates == []
+    if res.outcome == "CONTEXT_REQUIRED":
+        assert res.context_requirement is not None