diff --git a/Makefile b/Makefile
index 1ab9232..5c7c353 100644
--- a/Makefile
+++ b/Makefile
@@ -7,7 +7,10 @@ install:  ## Install runtime + dev dependencies
 	pip install -e ".[dev]"
 	pre-commit install
 
-run:  ## Run the API locally with hot reload
+run:  ## Run the API locally
+	uvicorn app.main:app --host 0.0.0.0 --port 8000
+
+dev:  ## Run the API locally with hot reload
 	uvicorn app.main:app --reload --reload-dir app --reload-exclude .venv \
 		--host 0.0.0.0 --port 8000
 
diff --git a/app/api/v1/endpoints/stages.py b/app/api/v1/endpoints/stages.py
index b7f8a09..3d26229 100644
--- a/app/api/v1/endpoints/stages.py
+++ b/app/api/v1/endpoints/stages.py
@@ -194,9 +194,18 @@ async def stage_content_fetch_extract(
             title=features.title,
             has_password_field=features.has_password_field,
             has_meta_refresh=features.has_meta_refresh,
+            has_password_form_external_action=features.has_password_form_external_action,
+            has_external_meta_refresh=features.has_external_meta_refresh,
             external_link_ratio=features.external_link_ratio,
             image_alts=features.image_alts,
             is_spa_shell=features.is_spa_shell,
+            body_text_snippets=features.body_text_snippets,
+            form_field_summaries=features.form_field_summaries,
+            cta_texts=features.cta_texts,
+            download_links=features.download_links,
+            sensitive_field_types=features.sensitive_field_types,
+            korean_lure_keywords=features.korean_lure_keywords,
+            public_agency_keywords=features.public_agency_keywords,
         ),
         html_preview=fetch_result.html[:_HTML_PREVIEW_CHARS],
     )
diff --git a/app/core/config.py b/app/core/config.py
index c5fd216..028f021 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -119,16 +119,21 @@ def alembic_database_url(self) -> str:
 
     # 도메인 휴리스틱 점수
     score_weight_ip_direct: int = 40
-    score_weight_typo_domain: int = 40
+    score_weight_typo_domain: int = 30
     score_weight_punycode_idn: int = 35
-    score_weight_no_https: int = 30
-    score_weight_new_domain: int = 30
-    score_weight_subdomain_overuse: int = 25
-    score_weight_open_redirect_param: int = 20
+    score_weight_no_https: int = 20
+    score_weight_new_domain: int = 25
+    score_weight_subdomain_overuse: int = 20
+    score_weight_open_redirect_param: int = 30
     score_weight_hyphen_overuse: int = 20
-    score_weight_suspicious_tld: int = 20
-    score_weight_dga_like: int = 15
-    score_weight_hosting_platform: int = 15
+    score_weight_suspicious_tld: int = 25
+    score_weight_dga_like: int = 10
+    score_weight_hosting_platform: int = 20
+    score_weight_url_userinfo: int = 45
+    score_weight_brand_in_url: int = 30
+    score_weight_free_hosting_lure: int = 25
+    score_weight_sensitive_path: int = 20
+    score_weight_url_shortener: int = 25
 
     # 휴리스틱 점수 캡 — GSB/URLhaus(각 50) high-confidence 시그널을 합산이 압도하지 않도록 클램프
     domain_heuristic_score_cap: int = 80
@@ -145,8 +150,8 @@ def alembic_database_url(self) -> str:
     domain_label_length_threshold: int = 20
 
     # 페이지 콘텐츠 정적 분석 (4단계)
-    content_fetch_timeout_seconds: float = 4.0
-    content_fetch_connect_timeout_seconds: float = 3.0
+    content_fetch_timeout_seconds: float = 3.5
+    content_fetch_connect_timeout_seconds: float = 2.0
     content_fetch_max_bytes: int = 2 * 1024 * 1024  # 2MiB 이상이면 끊고 분석
     # DNS rebinding 잔여 위험은 앱 레벨 사전 해석만으로 완전히 닫을 수 없다. 운영에서 분석 전용
     # egress 프록시를 두는 경우 이 값으로 fetch 트래픽을 강제 경유시킨다.
@@ -175,19 +180,32 @@ def alembic_database_url(self) -> str:
     # BS4 파싱은 입력 본문 대비 ~10배 메모리를 점유한다. FastAPI 동시성에 곱셈으로 폭주하는
     # 걸 막기 위해 추출 단계에 글로벌 세마포어를 둔다 — 동시성 N 이 와도 피크는 이 값 * per-page.
     content_extract_concurrency: int = 8
+    # 정확도 우선 slow path. 정적 HTML 로 판정이 애매하거나 SPA 셸인 경우에만 브라우저
+    # 렌더링 DOM 을 추가 분석한다. Playwright 미설치 환경에서는 degraded no-op.
+    content_precision_enabled: bool = True
+    content_precision_min_score: int = 20
+    content_render_timeout_seconds: float = 4.0
+    content_render_settle_ms: int = 500
+    content_render_concurrency: int = 2
 
     # 콘텐츠 분석 점수
     score_weight_brand_impersonation: int = 50
-    score_weight_logo_alt_impersonation: int = 30
+    score_weight_logo_alt_impersonation: int = 10
     score_weight_credential_form_external: int = 45
+    score_weight_pii_collection_form: int = 20
+    score_weight_sensitive_id_field: int = 30
+    score_weight_financial_field: int = 25
+    score_weight_risky_download_link: int = 20
+    score_weight_public_agency_lure: int = 20
+    score_weight_korean_lure: int = 15
     score_weight_meta_refresh: int = 20
     score_weight_external_meta_refresh: int = 25
-    score_weight_external_link_overuse: int = 15
+    score_weight_external_link_overuse: int = 5
     # 도달 실패(timeout/connect_error/HTTP 5xx 등)에 대한 보수적 가산.
     # not_html(이미지·PDF), too_large(대용량 정상 페이지), unexpected_redirect(unchainer 누락)는
     # 정상 컨텐츠 또는 파이프라인 정합성 문제로 보고 점수 가산 없이 시그널만 남긴다.
-    score_weight_content_fetch_failed: int = 10
-    score_weight_ai_phishing: int = 40
+    score_weight_content_fetch_failed: int = 15
+    score_weight_ai_phishing: int = 45
     score_weight_ai_suspicious: int = 20
     # 4단계 단독 캡 — 컨텐츠 분석 단계 안에서만 적용된다. 전 단계 합산은 별도로 score_total_cap 에서
     # 다시 100 으로 클램프되므로, 여기를 낮춰도 합산 상한이 자동으로 같이 낮아지는 게 아니다.
@@ -201,6 +219,14 @@ def alembic_database_url(self) -> str:
     # 단계별 점수 합산이 100 을 넘지 않도록 종합 점수에서 캡 적용.
     score_total_cap: int = 100
 
+    # 동기 분석 SLA. 단계별 작업은 이 총 예산 안에서만 실행되어 단일 요청이 20초를 넘기지
+    # 않도록 한다. 각 stage budget 은 외부 네트워크 timeout 이 순차 누적되는 것을 막기 위한 상한.
+    pipeline_total_timeout_seconds: float = 20.0
+    pipeline_unchain_timeout_seconds: float = 4.0
+    pipeline_reputation_timeout_seconds: float = 5.0
+    pipeline_domain_timeout_seconds: float = 4.0
+    pipeline_content_timeout_seconds: float = 8.0
+
     # 4단계 AI 프로바이더 선택.
     # auto   : OPENAI_API_KEY 있으면 OpenAIProvider, 없으면 NullAIProvider
     # openai : 강제 OpenAIProvider (키 없으면 경고 + NullAIProvider 폴백)
diff --git a/app/core/tld.py b/app/core/tld.py
index dbdebfe..c3d7186 100644
--- a/app/core/tld.py
+++ b/app/core/tld.py
@@ -1,9 +1,100 @@
 from __future__ import annotations
 
-import tldextract
+from dataclasses import dataclass
+from urllib.parse import urlparse
 
-_EXTRACTOR = tldextract.TLDExtract(cache_dir=None, suffix_list_urls=())
+_MULTI_LABEL_SUFFIXES = frozenset(
+    {
+        "ac.kr",
+        "co.in",
+        "co.jp",
+        "co.kr",
+        "co.uk",
+        "com.au",
+        "com.br",
+        "com.cn",
+        "com.hk",
+        "com.mx",
+        "com.my",
+        "com.sg",
+        "com.tr",
+        "com.tw",
+        "edu.au",
+        "go.kr",
+        "gov.au",
+        "gov.uk",
+        "ne.kr",
+        "net.cn",
+        "or.kr",
+        "org.cn",
+        "re.kr",
+        # Private hosting suffixes that phishing kits commonly abuse.
+        "amazonaws.com",
+        "azurewebsites.net",
+        "backblazeb2.com",
+        "duckdns.org",
+        "firebaseapp.com",
+        "fly.dev",
+        "free.nf",
+        "github.io",
+        "gitlab.io",
+        "glitch.me",
+        "godaddysites.com",
+        "herokuapp.com",
+        "my.id",
+        "netlify.app",
+        "pages.dev",
+        "railway.app",
+        "render.com",
+        "replit.app",
+        "surge.sh",
+        "vercel.app",
+        "wasmer.app",
+        "web.app",
+        "workers.dev",
+    }
+)
 
 
-def extract_url_parts(url: str) -> tldextract.ExtractResult:
-    return _EXTRACTOR(url)
+@dataclass(frozen=True)
+class ExtractResult:
+    subdomain: str
+    domain: str
+    suffix: str
+
+    @property
+    def top_domain_under_public_suffix(self) -> str:
+        if not self.domain or not self.suffix:
+            return ""
+        return f"{self.domain}.{self.suffix}"
+
+
+def _hostname(url: str) -> str:
+    parsed = urlparse(url if "://" in url else f"https://{url}")
+    return (parsed.hostname or "").strip(".").lower()
+
+
+def extract_url_parts(url: str) -> ExtractResult:
+    host = _hostname(url)
+    if not host:
+        return ExtractResult(subdomain="", domain="", suffix="")
+
+    labels = [label for label in host.split(".") if label]
+    if len(labels) == 1:
+        return ExtractResult(subdomain="", domain=labels[0], suffix="")
+
+    suffix_len = 1
+    for size in range(min(3, len(labels) - 1), 1, -1):
+        candidate = ".".join(labels[-size:])
+        if candidate in _MULTI_LABEL_SUFFIXES:
+            suffix_len = size
+            break
+
+    suffix = ".".join(labels[-suffix_len:])
+    domain_index = len(labels) - suffix_len - 1
+    if domain_index < 0:
+        return ExtractResult(subdomain="", domain="", suffix=suffix)
+
+    domain = labels[domain_index]
+    subdomain = ".".join(labels[:domain_index])
+    return ExtractResult(subdomain=subdomain, domain=domain, suffix=suffix)
diff --git a/app/schemas/content_analysis.py b/app/schemas/content_analysis.py
index e44e407..a633b82 100644
--- a/app/schemas/content_analysis.py
+++ b/app/schemas/content_analysis.py
@@ -9,6 +9,13 @@ class ContentSignal(StrEnum):
     BRAND_IMPERSONATION_FORM = "BRAND_IMPERSONATION_FORM"
     LOGO_ALT_IMPERSONATION = "LOGO_ALT_IMPERSONATION"
     CREDENTIAL_FORM_EXTERNAL = "CREDENTIAL_FORM_EXTERNAL"
+    PII_COLLECTION_FORM = "PII_COLLECTION_FORM"
+    SENSITIVE_ID_FIELD = "SENSITIVE_ID_FIELD"
+    FINANCIAL_FIELD = "FINANCIAL_FIELD"
+    RISKY_DOWNLOAD_LINK = "RISKY_DOWNLOAD_LINK"
+    PUBLIC_AGENCY_LURE = "PUBLIC_AGENCY_LURE"
+    KOREAN_LURE_TEXT = "KOREAN_LURE_TEXT"
+    RENDERED_DOM_ANALYZED = "RENDERED_DOM_ANALYZED"
     META_REFRESH = "META_REFRESH"
     EXTERNAL_META_REFRESH = "EXTERNAL_META_REFRESH"
     EXTERNAL_LINK_OVERUSE = "EXTERNAL_LINK_OVERUSE"
@@ -60,6 +67,15 @@ class ContentAnalysisResult(BaseModel):
     # JS 마운트 셸로 판단돼 정적 추출이 결정적이지 않은 상태. 여기서 has_password_field=False 는
     # "폼 없음" 이 아니라 "판정 불가" 로 읽어야 한다.
     is_spa_shell: bool = False
+    body_text_snippets: list[str] = Field(default_factory=list)
+    form_field_summaries: list[str] = Field(default_factory=list)
+    cta_texts: list[str] = Field(default_factory=list)
+    download_links: list[str] = Field(default_factory=list)
+    sensitive_field_types: list[str] = Field(default_factory=list)
+    korean_lure_keywords: list[str] = Field(default_factory=list)
+    public_agency_keywords: list[str] = Field(default_factory=list)
+    precision_analysis_used: bool = False
+    precision_analysis_error: str | None = None
 
     ai_verdict: AIVerdict | None = None
     ai_reason: str | None = None
@@ -109,6 +125,13 @@ class ExtractedFeaturesView(BaseModel):
     external_link_ratio: float | None = None
     image_alts: list[str] = Field(default_factory=list)
     is_spa_shell: bool = False
+    body_text_snippets: list[str] = Field(default_factory=list)
+    form_field_summaries: list[str] = Field(default_factory=list)
+    cta_texts: list[str] = Field(default_factory=list)
+    download_links: list[str] = Field(default_factory=list)
+    sensitive_field_types: list[str] = Field(default_factory=list)
+    korean_lure_keywords: list[str] = Field(default_factory=list)
+    public_agency_keywords: list[str] = Field(default_factory=list)
 
 
 class FetchExtractResponse(BaseModel):
diff --git a/app/schemas/domain_heuristic.py b/app/schemas/domain_heuristic.py
index f1cf8da..f3a25da 100644
--- a/app/schemas/domain_heuristic.py
+++ b/app/schemas/domain_heuristic.py
@@ -27,6 +27,11 @@ class DomainHeuristicSignal(StrEnum):
     SUSPICIOUS_TLD = "SUSPICIOUS_TLD"
     DGA_LIKE = "DGA_LIKE"
     HOSTING_PLATFORM = "HOSTING_PLATFORM"
+    URL_USERINFO = "URL_USERINFO"
+    BRAND_IN_URL = "BRAND_IN_URL"
+    FREE_HOSTING_LURE = "FREE_HOSTING_LURE"
+    SENSITIVE_PATH = "SENSITIVE_PATH"
+    URL_SHORTENER = "URL_SHORTENER"
 
 
 class DomainHeuristicSkippedReason(StrEnum):
diff --git a/app/services/analysis_callback.py b/app/services/analysis_callback.py
index bfd1119..66148b5 100644
--- a/app/services/analysis_callback.py
+++ b/app/services/analysis_callback.py
@@ -3,17 +3,64 @@
 import asyncio
 from datetime import UTC, datetime
 from typing import Any
+from urllib.parse import urlparse
 
 import httpx
 
 from app.core.config import settings
 from app.core.logging import get_logger
-from app.schemas.pipeline import PipelineFailure, PipelineSuccess
+from app.schemas.pipeline import PipelineFailure, PipelineStage, PipelineSuccess, Verdict
 
 logger = get_logger(__name__)
 
 _CALLBACK_PATH = "/internal/analysis-result"
 _MAX_ATTEMPTS = 3
+_KNOWN_MALICIOUS_SUMMARY = "악성으로 알려진 페이지 입니다."
+
+_STAGE_NUMBERS: dict[PipelineStage, int] = {
+    PipelineStage.NORMALIZE: 1,
+    PipelineStage.UNCHAIN: 1,
+    PipelineStage.THREAT_DB: 2,
+    PipelineStage.DOMAIN_HEURISTIC: 3,
+    PipelineStage.CONTENT_ANALYSIS: 4,
+}
+
+_DOMAIN_REASON_MESSAGES: dict[str, str] = {
+    "IP_DIRECT": "IP 주소로 직접 접근하는 URL입니다.",
+    "TYPO_DOMAIN": "유명 서비스와 유사한 도메인입니다.",
+    "PUNYCODE_IDN": "문자 혼동을 유도할 수 있는 국제화 도메인입니다.",
+    "NEW_DOMAIN": "최근 등록된 도메인입니다.",
+    "SUBDOMAIN_OVERUSE": "하위 도메인을 과도하게 사용합니다.",
+    "NO_HTTPS": "HTTPS가 아닌 연결을 사용합니다.",
+    "OPEN_REDIRECT_PARAM": "오픈 리다이렉트에 악용될 수 있는 파라미터가 있습니다.",
+    "HYPHEN_OVERUSE": "하이픈을 과도하게 사용한 도메인입니다.",
+    "SUSPICIOUS_TLD": "피싱에 자주 악용되는 최상위 도메인입니다.",
+    "DGA_LIKE": "자동 생성된 것으로 의심되는 도메인입니다.",
+    "HOSTING_PLATFORM": "무료 호스팅 플랫폼을 사용합니다.",
+    "URL_USERINFO": "URL에 사용자 정보 형식이 포함되어 있습니다.",
+    "BRAND_IN_URL": "URL에 브랜드명을 포함해 신뢰를 유도합니다.",
+    "FREE_HOSTING_LURE": "무료 호스팅 주소에서 신뢰를 유도하는 문구를 사용합니다.",
+    "SENSITIVE_PATH": "로그인 또는 인증 관련 경로를 사용합니다.",
+    "URL_SHORTENER": "단축 URL 서비스를 사용합니다.",
+}
+
+_CONTENT_REASON_MESSAGES: dict[str, str] = {
+    "BRAND_IMPERSONATION_FORM": "브랜드를 사칭하는 로그인 폼이 확인되었습니다.",
+    "LOGO_ALT_IMPERSONATION": "브랜드 로고 또는 이미지 설명을 사칭합니다.",
+    "CREDENTIAL_FORM_EXTERNAL": "로그인 정보가 외부 주소로 전송될 수 있습니다.",
+    "PII_COLLECTION_FORM": "개인정보 입력을 유도하는 폼이 있습니다.",
+    "SENSITIVE_ID_FIELD": "민감 식별정보 입력을 요구합니다.",
+    "FINANCIAL_FIELD": "금융정보 입력을 요구합니다.",
+    "RISKY_DOWNLOAD_LINK": "위험한 파일 다운로드를 유도합니다.",
+    "PUBLIC_AGENCY_LURE": "공공기관을 사칭하는 문구가 있습니다.",
+    "KOREAN_LURE_TEXT": "국내 사용자를 노린 유인 문구가 있습니다.",
+    "META_REFRESH": "자동 이동 태그가 포함되어 있습니다.",
+    "EXTERNAL_META_REFRESH": "외부 사이트로 자동 이동할 수 있습니다.",
+    "EXTERNAL_LINK_OVERUSE": "외부 링크 비율이 높습니다.",
+    "SPA_SHELL": "정적 분석만으로는 페이지 내용을 확인하기 어렵습니다.",
+    "FETCH_FAILED": "페이지 내용을 가져오지 못했습니다.",
+    "SKIPPED_ALREADY_DANGER": "선행 단계에서 위험성이 확인되어 콘텐츠 분석을 생략했습니다.",
+}
 
 
 def _iso_z(dt: datetime) -> str:
@@ -26,6 +73,214 @@ def _callback_url() -> str:
     return f"{settings.spring_internal_url.rstrip('/')}{_CALLBACK_PATH}"
 
 
+def _enum_value(value: Any) -> str:
+    return value.value if hasattr(value, "value") else str(value)
+
+
+def _host(url: str) -> str | None:
+    return urlparse(url).hostname
+
+
+def _signal_weight(code: str) -> int:
+    domain_weights = {
+        "IP_DIRECT": settings.score_weight_ip_direct,
+        "TYPO_DOMAIN": settings.score_weight_typo_domain,
+        "PUNYCODE_IDN": settings.score_weight_punycode_idn,
+        "NEW_DOMAIN": settings.score_weight_new_domain,
+        "SUBDOMAIN_OVERUSE": settings.score_weight_subdomain_overuse,
+        "NO_HTTPS": settings.score_weight_no_https,
+        "OPEN_REDIRECT_PARAM": settings.score_weight_open_redirect_param,
+        "HYPHEN_OVERUSE": settings.score_weight_hyphen_overuse,
+        "SUSPICIOUS_TLD": settings.score_weight_suspicious_tld,
+        "DGA_LIKE": settings.score_weight_dga_like,
+        "HOSTING_PLATFORM": settings.score_weight_hosting_platform,
+        "URL_USERINFO": settings.score_weight_url_userinfo,
+        "BRAND_IN_URL": settings.score_weight_brand_in_url,
+        "FREE_HOSTING_LURE": settings.score_weight_free_hosting_lure,
+        "SENSITIVE_PATH": settings.score_weight_sensitive_path,
+        "URL_SHORTENER": settings.score_weight_url_shortener,
+    }
+    content_weights = {
+        "BRAND_IMPERSONATION_FORM": settings.score_weight_brand_impersonation,
+        "LOGO_ALT_IMPERSONATION": settings.score_weight_logo_alt_impersonation,
+        "CREDENTIAL_FORM_EXTERNAL": settings.score_weight_credential_form_external,
+        "PII_COLLECTION_FORM": settings.score_weight_pii_collection_form,
+        "SENSITIVE_ID_FIELD": settings.score_weight_sensitive_id_field,
+        "FINANCIAL_FIELD": settings.score_weight_financial_field,
+        "RISKY_DOWNLOAD_LINK": settings.score_weight_risky_download_link,
+        "PUBLIC_AGENCY_LURE": settings.score_weight_public_agency_lure,
+        "KOREAN_LURE_TEXT": settings.score_weight_korean_lure,
+        "META_REFRESH": settings.score_weight_meta_refresh,
+        "EXTERNAL_META_REFRESH": settings.score_weight_external_meta_refresh,
+        "EXTERNAL_LINK_OVERUSE": settings.score_weight_external_link_overuse,
+        "FETCH_FAILED": settings.score_weight_content_fetch_failed,
+    }
+    return domain_weights.get(code) or content_weights.get(code) or 0
+
+
+def _add_reason(
+    reasons: list[dict[str, Any]],
+    *,
+    code: str,
+    stage: int,
+    weight: int,
+    message: str,
+) -> None:
+    if weight <= 0 and code not in {"GSB_MATCH", "URLHAUS_MATCH"}:
+        return
+    reasons.append(
+        {
+            "code": code,
+            "stage": stage,
+            "weight": weight,
+            "message": message,
+        }
+    )
+
+
+def _callback_reasons(result: PipelineSuccess) -> list[dict[str, Any]]:
+    stages = result.stages
+    reasons: list[dict[str, Any]] = []
+
+    if stages.threat_db.gsb.is_threat:
+        _add_reason(
+            reasons,
+            code="GSB_MATCH",
+            stage=2,
+            weight=settings.score_weight_gsb,
+            message="Google Safe Browsing에 악성 URL로 등록되어 있습니다.",
+        )
+    if stages.threat_db.urlhaus.is_threat:
+        _add_reason(
+            reasons,
+            code="URLHAUS_MATCH",
+            stage=2,
+            weight=settings.score_weight_urlhaus,
+            message="URLhaus에 악성 URL로 등록되어 있습니다.",
+        )
+
+    for signal in stages.domain_heuristic.signals:
+        code = _enum_value(signal)
+        _add_reason(
+            reasons,
+            code=code,
+            stage=3,
+            weight=_signal_weight(code),
+            message=_DOMAIN_REASON_MESSAGES.get(code, "도메인 휴리스틱 위험 신호가 있습니다."),
+        )
+
+    for signal in stages.content_analysis.signals:
+        code = _enum_value(signal)
+        _add_reason(
+            reasons,
+            code=code,
+            stage=4,
+            weight=_signal_weight(code),
+            message=_CONTENT_REASON_MESSAGES.get(code, "콘텐츠 분석 위험 신호가 있습니다."),
+        )
+
+    ai_verdict = stages.content_analysis.ai_verdict
+    if ai_verdict is not None:
+        verdict = _enum_value(ai_verdict)
+        if verdict == "phishing":
+            _add_reason(
+                reasons,
+                code="AI_PHISHING",
+                stage=4,
+                weight=settings.score_weight_ai_phishing,
+                message=(
+                    stages.content_analysis.ai_reason
+                    or "AI가 피싱 가능성이 높다고 판단했습니다."
+                ),
+            )
+        elif verdict == "suspicious":
+            _add_reason(
+                reasons,
+                code="AI_SUSPICIOUS",
+                stage=4,
+                weight=settings.score_weight_ai_suspicious,
+                message=(
+                    stages.content_analysis.ai_reason
+                    or "AI가 의심스러운 페이지로 판단했습니다."
+                ),
+            )
+
+    return reasons
+
+
+def _callback_summary(result: PipelineSuccess) -> str:
+    if result.stages.threat_db.is_malicious:
+        return _KNOWN_MALICIOUS_SUMMARY
+    content = result.stages.content_analysis
+    if content.ai_reason:
+        return content.ai_reason
+    if content.reason:
+        return content.reason
+    if result.verdict == Verdict.DANGER:
+        return "위험성이 높은 페이지입니다. 접속과 정보 입력을 피하세요."
+    if result.verdict == Verdict.CAUTION:
+        return "주의가 필요한 페이지입니다. 링크와 입력 정보를 한 번 더 확인하세요."
+    return "현재 분석 기준에서 뚜렷한 위험 신호가 확인되지 않았습니다."
+
+
+def _spring_stages(result: PipelineSuccess) -> dict[str, Any]:
+    stages = result.stages
+    gsb_matches = [
+        match.threat_type
+        for match in stages.threat_db.gsb.matches
+        if match.threat_type
+    ]
+
+    chain = [hop.url for hop in stages.unchain.hops]
+    if not chain:
+        chain.append(stages.unchain.input_url)
+    if stages.unchain.final_url not in chain:
+        chain.append(stages.unchain.final_url)
+
+    rdap = stages.domain_heuristic.rdap
+    rdap_payload = None
+    if rdap is not None:
+        rdap_payload = {
+            "domain": rdap.domain,
+            "registrar": rdap.registrar,
+            "createdDate": _iso_z(rdap.created_date) if rdap.created_date else None,
+            "domainAgeDays": rdap.domain_age_days,
+            "isNewDomain": rdap.is_new_domain,
+        }
+
+    content = stages.content_analysis
+    return {
+        "externalDb": {
+            "gsb": {
+                "isThreat": stages.threat_db.gsb.is_threat,
+                "matchedTypes": gsb_matches,
+            },
+            "urlhaus": {
+                "isThreat": stages.threat_db.urlhaus.is_threat,
+                "host": _host(stages.threat_db.final_url),
+            },
+        },
+        "unchain": {
+            "hops": stages.unchain.hop_count,
+            "chain": chain,
+        },
+        "domainHeuristic": {
+            "rdap": rdap_payload,
+            "signals": [_enum_value(signal) for signal in stages.domain_heuristic.signals],
+        },
+        "contentAnalysis": {
+            "fetched": content.fetched,
+            "hasPasswordField": content.has_password_field,
+            "aiVerdict": _enum_value(content.ai_verdict) if content.ai_verdict else None,
+            "aiReason": content.ai_reason,
+        },
+    }
+
+
+def _error_stage(stage: PipelineStage) -> int:
+    return _STAGE_NUMBERS[stage]
+
+
 def _success_payload(
     result: PipelineSuccess,
     *,
@@ -41,13 +296,13 @@ def _success_payload(
         "finalUrl": result.final_url,
         "verdict": result.verdict.value,
         "score": result.score,
-        "stages": result.stages.model_dump(mode="json"),
+        "reasons": _callback_reasons(result),
+        "stages": _spring_stages(result),
+        "summary": _callback_summary(result),
         "engineVersion": settings.app_version,
         "analyzedAt": _iso_z(analyzed_at),
         "elapsedMs": elapsed_ms,
     }
-    if result.timings is not None:
-        payload["timings"] = result.timings.model_dump(mode="json")
     return payload
 
 
@@ -65,15 +320,13 @@ def _failure_payload(
         "originalUrl": result.original_url,
         "error": {
             "code": f"{result.failed_at_stage.value.upper()}_FAILED",
-            "stage": result.failed_at_stage.value,
+            "stage": _error_stage(result.failed_at_stage),
             "message": result.error,
         },
         "engineVersion": settings.app_version,
         "analyzedAt": _iso_z(analyzed_at),
         "elapsedMs": elapsed_ms,
     }
-    if result.timings is not None:
-        payload["timings"] = result.timings.model_dump(mode="json")
     return payload
 
 
diff --git a/app/services/content_analyzer/ai.py b/app/services/content_analyzer/ai.py
index 19229c7..33726c0 100644
--- a/app/services/content_analyzer/ai.py
+++ b/app/services/content_analyzer/ai.py
@@ -31,6 +31,14 @@ class AIPromptContext:
     # AI 가 페이지 피처 단독으로는 결론 내기 약한 케이스에 사전 정보로 활용한다.
     # 예: TYPO_DOMAIN + 브랜드 title + 비밀번호 폼이면 단독 페이지 분석보다 강한 phishing 신호.
     upstream_signals: tuple[str, ...] = ()
+    # HTML 원문 대신 비용이 낮은 구조화 요약만 전달한다.
+    body_text_snippets: tuple[str, ...] = ()
+    form_field_summaries: tuple[str, ...] = ()
+    cta_texts: tuple[str, ...] = ()
+    download_links: tuple[str, ...] = ()
+    sensitive_field_types: tuple[str, ...] = ()
+    korean_lure_keywords: tuple[str, ...] = ()
+    public_agency_keywords: tuple[str, ...] = ()
 
 
 @dataclass(frozen=True)
diff --git a/app/services/content_analyzer/ai_openai.py b/app/services/content_analyzer/ai_openai.py
index 750fd7b..e92fb49 100644
--- a/app/services/content_analyzer/ai_openai.py
+++ b/app/services/content_analyzer/ai_openai.py
@@ -49,7 +49,9 @@
 _SYSTEM_PROMPT = (
     "당신은 웹 페이지 피싱 여부를 판정하는 보안 엔진이다. "
     "입력은 이미 크롤러가 뽑아놓은 구조화된 피처(title, 비밀번호 필드 유무, "
-    "meta refresh 유무, 외부 링크 비율, 이미지 alt 텍스트, 최종 URL, spa_shell 플래그)와 "
+    "meta refresh 유무, 외부 링크 비율, 이미지 alt 텍스트, 본문 요약, 폼 필드 요약, "
+    "CTA 문구, 다운로드 링크, 민감 입력 필드 후보, 한국형 루어 키워드, "
+    "최종 URL, spa_shell 플래그)와 "
     "선행 단계(도메인 휴리스틱·위협 DB)에서 이미 잡힌 시그널 코드 배열 upstream_signals 다. "
     "규칙 기반 점수는 이미 상위 레이어에서 계산되고 있으므로, 당신은 "
     "'이 페이지가 특정 브랜드를 사칭하거나 자격증명을 탈취하려 하는지' 를 "
@@ -83,6 +85,13 @@ def _build_user_prompt(ctx: AIPromptContext) -> str:
         "spa_shell": ctx.is_spa_shell,
         # 과도하게 긴 alt 리스트는 비용만 늘리고 판정에 도움 안 됨
         "image_alts": list(ctx.image_alts[:10]),
+        "body_text": list(ctx.body_text_snippets[:10]),
+        "form_fields": list(ctx.form_field_summaries[:20]),
+        "cta_texts": list(ctx.cta_texts[:10]),
+        "download_links": list(ctx.download_links[:10]),
+        "sensitive_field_types": list(ctx.sensitive_field_types),
+        "korean_lure_keywords": list(ctx.korean_lure_keywords),
+        "public_agency_keywords": list(ctx.public_agency_keywords),
         # 선행 단계 시그널 — 빈 배열이면 단독 페이지 분석과 동일.
         "upstream_signals": list(ctx.upstream_signals),
     }
diff --git a/app/services/content_analyzer/analyze.py b/app/services/content_analyzer/analyze.py
index 6beb49f..f0a9478 100644
--- a/app/services/content_analyzer/analyze.py
+++ b/app/services/content_analyzer/analyze.py
@@ -22,6 +22,7 @@
 from app.services.content_analyzer.ai import AIPromptContext, AIProvider, get_ai_provider
 from app.services.content_analyzer.extract import ExtractedFeatures, extract_features_async
 from app.services.content_analyzer.fetch import fetch_page
+from app.services.content_analyzer.render import render_page
 from app.services.content_analyzer.signals import ContentScoring, score_content
 
 logger = get_logger(__name__)
@@ -43,6 +44,13 @@ def _build_prompt_context(
         has_password_form_external_action=features.has_password_form_external_action,
         has_external_meta_refresh=features.has_external_meta_refresh,
         upstream_signals=upstream_signals,
+        body_text_snippets=tuple(features.body_text_snippets),
+        form_field_summaries=tuple(features.form_field_summaries),
+        cta_texts=tuple(features.cta_texts),
+        download_links=tuple(features.download_links),
+        sensitive_field_types=tuple(features.sensitive_field_types),
+        korean_lure_keywords=tuple(features.korean_lure_keywords),
+        public_agency_keywords=tuple(features.public_agency_keywords),
     )
 
 
@@ -55,6 +63,78 @@ def _ai_score_weight(verdict: AIVerdict) -> int:
     return 0
 
 
+def _unique_merge(left: list[str], right: list[str]) -> list[str]:
+    merged = list(left)
+    for item in right:
+        if item not in merged:
+            merged.append(item)
+    return merged
+
+
+def _merge_features(static: ExtractedFeatures, rendered: ExtractedFeatures) -> ExtractedFeatures:
+    return ExtractedFeatures(
+        title=rendered.title or static.title,
+        has_password_field=static.has_password_field or rendered.has_password_field,
+        has_password_form_external_action=(
+            static.has_password_form_external_action
+            or rendered.has_password_form_external_action
+        ),
+        has_meta_refresh=static.has_meta_refresh or rendered.has_meta_refresh,
+        has_external_meta_refresh=(
+            static.has_external_meta_refresh or rendered.has_external_meta_refresh
+        ),
+        external_link_ratio=(
+            rendered.external_link_ratio
+            if rendered.external_link_ratio is not None
+            else static.external_link_ratio
+        ),
+        image_alts=_unique_merge(static.image_alts, rendered.image_alts),
+        is_spa_shell=rendered.is_spa_shell,
+        body_text_snippets=_unique_merge(
+            static.body_text_snippets,
+            rendered.body_text_snippets,
+        ),
+        form_field_summaries=_unique_merge(
+            static.form_field_summaries,
+            rendered.form_field_summaries,
+        ),
+        cta_texts=_unique_merge(static.cta_texts, rendered.cta_texts),
+        download_links=_unique_merge(static.download_links, rendered.download_links),
+        sensitive_field_types=_unique_merge(
+            static.sensitive_field_types,
+            rendered.sensitive_field_types,
+        ),
+        korean_lure_keywords=_unique_merge(
+            static.korean_lure_keywords,
+            rendered.korean_lure_keywords,
+        ),
+        public_agency_keywords=_unique_merge(
+            static.public_agency_keywords,
+            rendered.public_agency_keywords,
+        ),
+    )
+
+
+def _should_run_precision_analysis(
+    features: ExtractedFeatures,
+    scoring: ContentScoring,
+) -> bool:
+    if not settings.content_precision_enabled:
+        return False
+    if scoring.score >= settings.score_danger_threshold:
+        return False
+    if features.is_spa_shell:
+        return True
+    if scoring.score >= settings.content_precision_min_score:
+        return True
+    return bool(
+        features.download_links
+        or features.sensitive_field_types
+        or features.korean_lure_keywords
+        or features.public_agency_keywords
+    )
+
+
 # 정상 컨텐츠(이미지·PDF·대용량 정적 페이지) 또는 파이프라인 정합성 이슈(unchainer 가 놓친 3xx)는
 # 악성 신호로 보지 않는다 — FETCH_FAILED 시그널만 남기고 점수는 가산하지 않는다.
 # blocked_host 는 우리가 SSRF 1·2선에서 차단한 결과지 페이지의 악성 신호가 아니라 동일하게 0 가산.
@@ -150,6 +230,20 @@ async def analyze_content(
 
     features = await extract_features_async(fetch_result.html, base_url=final_url)
     scoring: ContentScoring = score_content(features, final_url)
+    precision_analysis_used = False
+    precision_analysis_error: str | None = None
+
+    if _should_run_precision_analysis(features, scoring):
+        precision_analysis_used = True
+        render_result = await render_page(final_url)
+        if render_result.ok:
+            rendered_features = await extract_features_async(render_result.html, base_url=final_url)
+            features = _merge_features(features, rendered_features)
+            scoring = score_content(features, final_url)
+            if ContentSignal.RENDERED_DOM_ANALYZED not in scoring.signals:
+                scoring.signals.append(ContentSignal.RENDERED_DOM_ANALYZED)
+        else:
+            precision_analysis_error = render_result.error
 
     ai_verdict: AIVerdict | None = None
     ai_reason: str | None = None
@@ -209,6 +303,15 @@ async def analyze_content(
         brand_impersonation=scoring.brand_impersonation,
         logo_alt_impersonation=scoring.logo_alt_impersonation,
         is_spa_shell=features.is_spa_shell,
+        body_text_snippets=features.body_text_snippets,
+        form_field_summaries=features.form_field_summaries,
+        cta_texts=features.cta_texts,
+        download_links=features.download_links,
+        sensitive_field_types=features.sensitive_field_types,
+        korean_lure_keywords=features.korean_lure_keywords,
+        public_agency_keywords=features.public_agency_keywords,
+        precision_analysis_used=precision_analysis_used,
+        precision_analysis_error=precision_analysis_error,
         ai_verdict=ai_verdict,
         ai_reason=ai_reason,
         ai_error=ai_error,
diff --git a/app/services/content_analyzer/extract.py b/app/services/content_analyzer/extract.py
index 8aa98a0..0dab603 100644
--- a/app/services/content_analyzer/extract.py
+++ b/app/services/content_analyzer/extract.py
@@ -16,6 +16,7 @@
 from urllib.parse import urljoin, urlparse
 
 from bs4 import BeautifulSoup
+from bs4.element import Tag
 
 from app.core.config import settings
 
@@ -46,6 +47,53 @@
 # 입력은 이미 캡 됐지만, 그 안에서도 노드를 압축적으로 쌓은 페이지(예: <a> 만 50만 개)에서
 # find_all 의 매칭 비용이 폭주하지 않도록 두 번째 방어선을 둔다.
 _MAX_NODES_PER_TAG = 1000
+_MAX_TEXT_SNIPPETS = 40
+_MAX_TEXT_CHARS = 180
+_MAX_FORM_FIELD_SUMMARIES = 80
+_MAX_CTA_TEXTS = 40
+_MAX_DOWNLOAD_LINKS = 40
+_RISKY_DOWNLOAD_EXTENSIONS: frozenset[str] = frozenset(
+    {".apk", ".ipa", ".exe", ".msi", ".dmg", ".scr", ".bat", ".cmd", ".js", ".vbs"}
+)
+
+_KOREAN_LURE_KEYWORDS: tuple[str, ...] = (
+    "지원금",
+    "환급금",
+    "택배",
+    "부고",
+    "청첩장",
+    "과태료",
+    "건강보험",
+    "정부기관",
+    "카카오톡",
+    "인증번호",
+    "otp",
+)
+
+_PUBLIC_AGENCY_KEYWORDS: tuple[str, ...] = (
+    "국민건강보험",
+    "건강보험공단",
+    "건강보험",
+    "정부24",
+    "정부",
+    "공단",
+    "국세청",
+    "관세청",
+    "경찰청",
+    "검찰청",
+    "고용노동부",
+    "보건복지부",
+    "행정안전부",
+)
+
+_SENSITIVE_FIELD_KEYWORDS: tuple[tuple[str, tuple[str, ...]], ...] = (
+    ("resident_registration_number", ("resident_registration", "rrn", "주민등록", "주민번호")),
+    ("phone", ("mobile", "phone", "tel", "휴대폰", "전화번호")),
+    ("card", ("card_number", "credit_card", "creditcard", "카드번호", "카드")),
+    ("cvc", ("cvc", "cvv")),
+    ("account", ("account", "bank", "계좌", "은행")),
+    ("otp", ("otp", "auth_code", "verification_code", "인증번호", "인증")),
+)
 
 
 @dataclass
@@ -58,6 +106,13 @@ class ExtractedFeatures:
     external_link_ratio: float | None = None
     image_alts: list[str] = field(default_factory=list)
     is_spa_shell: bool = False
+    body_text_snippets: list[str] = field(default_factory=list)
+    form_field_summaries: list[str] = field(default_factory=list)
+    cta_texts: list[str] = field(default_factory=list)
+    download_links: list[str] = field(default_factory=list)
+    sensitive_field_types: list[str] = field(default_factory=list)
+    korean_lure_keywords: list[str] = field(default_factory=list)
+    public_agency_keywords: list[str] = field(default_factory=list)
 
 
 def _normalize_host(host: str | None) -> str:
@@ -190,9 +245,164 @@ def _collect_image_alts(soup: BeautifulSoup) -> list[str]:
     return alts
 
 
+def _trim_text(text: str, *, max_chars: int = _MAX_TEXT_CHARS) -> str:
+    text = " ".join(text.split())
+    if len(text) <= max_chars:
+        return text
+    return text[:max_chars].rstrip()
+
+
+def _append_unique(items: list[str], value: str, *, limit: int) -> None:
+    value = _trim_text(value)
+    if value and value not in items and len(items) < limit:
+        items.append(value)
+
+
+def _collect_body_text_snippets(soup: BeautifulSoup) -> list[str]:
+    snippets: list[str] = []
+    body = soup.body or soup
+    for text in body.stripped_strings:
+        _append_unique(snippets, text, limit=_MAX_TEXT_SNIPPETS)
+        if len(snippets) >= _MAX_TEXT_SNIPPETS:
+            break
+    return snippets
+
+
+def _label_texts_by_for(soup: BeautifulSoup) -> dict[str, str]:
+    labels: dict[str, str] = {}
+    for label in soup.find_all("label", limit=_MAX_NODES_PER_TAG):
+        if not isinstance(label, Tag):
+            continue
+        target = label.get("for")
+        if isinstance(target, str) and target.strip():
+            text = label.get_text(" ", strip=True)
+            if text:
+                labels[target.strip()] = _trim_text(text)
+    return labels
+
+
+def _attr_text(tag: Tag, name: str) -> str | None:
+    value = tag.get(name)
+    if isinstance(value, str):
+        value = _trim_text(value)
+        return value or None
+    return None
+
+
+def _field_label(control: Tag, labels_by_for: dict[str, str]) -> str | None:
+    control_id = _attr_text(control, "id")
+    if control_id and control_id in labels_by_for:
+        return labels_by_for[control_id]
+    parent_label = control.find_parent("label")
+    if isinstance(parent_label, Tag):
+        text = parent_label.get_text(" ", strip=True)
+        return _trim_text(text) if text else None
+    return None
+
+
+def _collect_form_field_summaries(soup: BeautifulSoup) -> list[str]:
+    labels_by_for = _label_texts_by_for(soup)
+    summaries: list[str] = []
+    for control in soup.find_all(["input", "textarea", "select"], limit=_MAX_NODES_PER_TAG):
+        if not isinstance(control, Tag):
+            continue
+        parts: list[str] = []
+        label = _field_label(control, labels_by_for)
+        if label:
+            parts.append(f"label={label}")
+        for attr in ("name", "type", "placeholder", "id", "title", "aria-label"):
+            value = _attr_text(control, attr)
+            if value:
+                parts.append(f"{attr}={value}")
+        if parts:
+            _append_unique(summaries, " ".join(parts), limit=_MAX_FORM_FIELD_SUMMARIES)
+        if len(summaries) >= _MAX_FORM_FIELD_SUMMARIES:
+            break
+    return summaries
+
+
+def _collect_cta_texts(soup: BeautifulSoup) -> list[str]:
+    texts: list[str] = []
+    for tag in soup.find_all(["button", "a", "input"], limit=_MAX_NODES_PER_TAG):
+        if not isinstance(tag, Tag):
+            continue
+        text: str | None = None
+        if tag.name == "input":
+            input_type = (_attr_text(tag, "type") or "").lower()
+            if input_type not in {"submit", "button", "reset"}:
+                continue
+            text = _attr_text(tag, "value")
+        else:
+            text = tag.get_text(" ", strip=True)
+        if text:
+            _append_unique(texts, text, limit=_MAX_CTA_TEXTS)
+        if len(texts) >= _MAX_CTA_TEXTS:
+            break
+    return texts
+
+
+def _is_risky_download_url(raw_url: str, base_url: str) -> str | None:
+    joined = urljoin(base_url, raw_url.strip())
+    parsed = urlparse(joined)
+    if parsed.scheme not in _NAV_SCHEMES:
+        return None
+    path = parsed.path.lower()
+    if any(path.endswith(ext) for ext in _RISKY_DOWNLOAD_EXTENSIONS):
+        return joined
+    return None
+
+
+def _collect_download_links(soup: BeautifulSoup, base_url: str) -> list[str]:
+    links: list[str] = []
+    for anchor in soup.find_all("a", limit=_MAX_NODES_PER_TAG):
+        href = anchor.get("href")
+        if not isinstance(href, str):
+            continue
+        resolved = _is_risky_download_url(href, base_url)
+        if resolved is not None:
+            _append_unique(links, resolved, limit=_MAX_DOWNLOAD_LINKS)
+        if len(links) >= _MAX_DOWNLOAD_LINKS:
+            break
+    return links
+
+
+def _collect_sensitive_field_types(form_field_summaries: list[str]) -> list[str]:
+    found: list[str] = []
+    haystack = "\n".join(form_field_summaries).lower()
+    for field_type, keywords in _SENSITIVE_FIELD_KEYWORDS:
+        if any(keyword.lower() in haystack for keyword in keywords):
+            found.append(field_type)
+    return found
+
+
+def _collect_keywords(texts: list[str], keywords: tuple[str, ...]) -> list[str]:
+    found: list[str] = []
+    haystack = "\n".join(texts).lower()
+    for keyword in keywords:
+        if keyword.lower() in haystack and keyword not in found:
+            found.append(keyword)
+    return found
+
+
 def extract_features(html: str, base_url: str) -> ExtractedFeatures:
     soup = BeautifulSoup(html or "", "lxml")
     has_meta_refresh, has_external_meta_refresh = _meta_refresh_info(soup, base_url)
+    body_text_snippets = _collect_body_text_snippets(soup)
+    form_field_summaries = _collect_form_field_summaries(soup)
+    cta_texts = _collect_cta_texts(soup)
+    image_alts = _collect_image_alts(soup)
+    download_links = _collect_download_links(soup, base_url)
+    keyword_texts = [
+        item
+        for item in [
+            _extract_title(soup) or "",
+            *body_text_snippets,
+            *form_field_summaries,
+            *cta_texts,
+            *image_alts,
+        ]
+        if item
+    ]
     return ExtractedFeatures(
         title=_extract_title(soup),
         has_password_field=_has_password_field(soup),
@@ -200,8 +410,15 @@ def extract_features(html: str, base_url: str) -> ExtractedFeatures:
         has_meta_refresh=has_meta_refresh,
         has_external_meta_refresh=has_external_meta_refresh,
         external_link_ratio=_compute_external_link_ratio(soup, base_url),
-        image_alts=_collect_image_alts(soup),
+        image_alts=image_alts,
         is_spa_shell=_detect_spa_shell(soup),
+        body_text_snippets=body_text_snippets,
+        form_field_summaries=form_field_summaries,
+        cta_texts=cta_texts,
+        download_links=download_links,
+        sensitive_field_types=_collect_sensitive_field_types(form_field_summaries),
+        korean_lure_keywords=_collect_keywords(keyword_texts, _KOREAN_LURE_KEYWORDS),
+        public_agency_keywords=_collect_keywords(keyword_texts, _PUBLIC_AGENCY_KEYWORDS),
     )
 
 
diff --git a/app/services/content_analyzer/render.py b/app/services/content_analyzer/render.py
new file mode 100644
index 0000000..0350f3b
--- /dev/null
+++ b/app/services/content_analyzer/render.py
@@ -0,0 +1,112 @@
+"""선택적 Playwright 기반 렌더링 분석.
+
+정적 HTML 만으로 판정하기 어려운 SPA/동적 페이지에 한해 호출된다.
+Playwright 가 설치되지 않았거나 브라우저 실행에 실패하면 degraded 결과를 반환하고,
+기존 정적 분석 경로는 그대로 유지한다.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+from typing import Any
+from urllib.parse import urlparse
+
+from app.core.config import settings
+from app.core.logging import get_logger
+from app.services.content_analyzer.fetch import (
+    _is_blocked_host,
+    _is_ip_literal,
+    _pick_user_agent,
+    _resolved_addrs_blocked,
+)
+
+logger = get_logger(__name__)
+_BLOCKED_RESOURCE_TYPES: frozenset[str] = frozenset({"image", "media", "font"})
+
+
+@dataclass(frozen=True)
+class RenderResult:
+    ok: bool
+    url: str
+    html: str = ""
+    error: str | None = None
+
+
+_render_semaphore: asyncio.Semaphore | None = None
+
+
+def _get_render_semaphore() -> asyncio.Semaphore:
+    global _render_semaphore
+    if _render_semaphore is None:
+        _render_semaphore = asyncio.Semaphore(settings.content_render_concurrency)
+    return _render_semaphore
+
+
+async def _target_blocked(url: str) -> bool:
+    parsed = urlparse(url)
+    if _is_blocked_host(parsed.hostname):
+        return True
+    if parsed.hostname and not _is_ip_literal(parsed.hostname.strip("[]")):
+        try:
+            return await _resolved_addrs_blocked(parsed.hostname)
+        except OSError:
+            return False
+    return False
+
+
+async def render_page(url: str) -> RenderResult:
+    """브라우저 렌더링 후 DOM HTML 을 반환한다.
+
+    내부망 SSRF 방어는 fetch 와 같은 1·2선 검사 함수를 재사용한다. DNS rebind 잔여 위험은
+    fetch 와 동일하게 배포 단 egress 정책으로 닫아야 한다.
+    """
+    if not settings.content_precision_enabled:
+        return RenderResult(ok=False, url=url, error="precision_disabled")
+
+    if await _target_blocked(url):
+        logger.info("content_render.blocked_host", url=url)
+        return RenderResult(ok=False, url=url, error="blocked_host")
+
+    try:
+        from playwright.async_api import TimeoutError as PlaywrightTimeoutError
+        from playwright.async_api import async_playwright
+    except ImportError:
+        return RenderResult(ok=False, url=url, error="playwright_unavailable")
+
+    timeout_ms = int(settings.content_render_timeout_seconds * 1000)
+    sem = _get_render_semaphore()
+    async with sem:
+        try:
+            async with async_playwright() as p:
+                browser = await p.chromium.launch(headless=True)
+                try:
+                    page = await browser.new_page(
+                        user_agent=_pick_user_agent(),
+                        locale="ko-KR",
+                    )
+
+                    async def _route_handler(route: Any) -> None:
+                        request = route.request
+                        if request.resource_type in _BLOCKED_RESOURCE_TYPES:
+                            await route.abort()
+                        else:
+                            await route.continue_()
+
+                    await page.route("**/*", _route_handler)
+                    await page.goto(url, wait_until="domcontentloaded", timeout=timeout_ms)
+                    await page.wait_for_timeout(settings.content_render_settle_ms)
+                    html = await page.content()
+                    return RenderResult(ok=True, url=url, html=html)
+                finally:
+                    await browser.close()
+        except PlaywrightTimeoutError:
+            return RenderResult(ok=False, url=url, error="timeout")
+        except Exception as exc:
+            logger.warning(
+                "content_render.failed",
+                url=url,
+                error=str(exc),
+                error_type=type(exc).__name__,
+            )
+            return RenderResult(ok=False, url=url, error="render_failed")
diff --git a/app/services/content_analyzer/signals.py b/app/services/content_analyzer/signals.py
index 078a622..f51e7a8 100644
--- a/app/services/content_analyzer/signals.py
+++ b/app/services/content_analyzer/signals.py
@@ -82,6 +82,19 @@ class ContentScoring:
     logo_alt_impersonation: bool = False
 
 
+_SENSITIVE_ID_TYPES: frozenset[str] = frozenset({"resident_registration_number", "otp"})
+_PII_FIELD_TYPES: frozenset[str] = frozenset(
+    {"resident_registration_number", "phone", "otp"}
+)
+_FINANCIAL_FIELD_TYPES: frozenset[str] = frozenset({"card", "cvc", "account"})
+
+
+def _add_signal(result: ContentScoring, signal: ContentSignal, score: int = 0) -> None:
+    if signal not in result.signals:
+        result.signals.append(signal)
+    result.score += score
+
+
 def _url_brand_label(final_url: str) -> str:
     ext = extract_url_parts(final_url)
     return (ext.domain or "").lower()
@@ -123,8 +136,60 @@ def score_content(features: ExtractedFeatures, final_url: str) -> ContentScoring
             result.score += settings.score_weight_brand_impersonation
 
     if features.has_password_form_external_action:
-        result.signals.append(ContentSignal.CREDENTIAL_FORM_EXTERNAL)
-        result.score += settings.score_weight_credential_form_external
+        _add_signal(
+            result,
+            ContentSignal.CREDENTIAL_FORM_EXTERNAL,
+            settings.score_weight_credential_form_external,
+        )
+
+    sensitive_types = set(features.sensitive_field_types)
+    has_pii_fields = bool(sensitive_types & _PII_FIELD_TYPES)
+    has_sensitive_id_fields = bool(sensitive_types & _SENSITIVE_ID_TYPES)
+    has_financial_fields = bool(sensitive_types & _FINANCIAL_FIELD_TYPES)
+    has_lure_text = bool(features.korean_lure_keywords)
+    has_public_agency_lure = bool(features.public_agency_keywords)
+
+    if has_lure_text:
+        _add_signal(result, ContentSignal.KOREAN_LURE_TEXT)
+
+    if has_public_agency_lure:
+        _add_signal(result, ContentSignal.PUBLIC_AGENCY_LURE)
+
+    if has_pii_fields:
+        _add_signal(
+            result,
+            ContentSignal.PII_COLLECTION_FORM,
+            settings.score_weight_pii_collection_form,
+        )
+
+    if has_sensitive_id_fields:
+        _add_signal(
+            result,
+            ContentSignal.SENSITIVE_ID_FIELD,
+            settings.score_weight_sensitive_id_field,
+        )
+
+    if has_financial_fields:
+        _add_signal(
+            result,
+            ContentSignal.FINANCIAL_FIELD,
+            settings.score_weight_financial_field,
+        )
+
+    if features.download_links:
+        _add_signal(
+            result,
+            ContentSignal.RISKY_DOWNLOAD_LINK,
+            settings.score_weight_risky_download_link,
+        )
+
+    # 루어/기관명은 단독 문구만으로 점수를 만들면 정상 안내 페이지 오탐이 커진다.
+    # 실제 입력 폼이나 위험 다운로드와 결합될 때만 추가 가중한다.
+    if has_lure_text and (has_pii_fields or has_financial_fields or features.download_links):
+        result.score += settings.score_weight_korean_lure
+
+    if has_public_agency_lure and has_pii_fields:
+        result.score += settings.score_weight_public_agency_lure
 
     alt_brands: set[str] = set()
     for alt in features.image_alts:
diff --git a/app/services/db_independent_pipeline.py b/app/services/db_independent_pipeline.py
index b9773ca..46ce2b7 100644
--- a/app/services/db_independent_pipeline.py
+++ b/app/services/db_independent_pipeline.py
@@ -25,6 +25,13 @@
 from app.services.content_analyzer import analyze_content, skipped_already_danger
 from app.services.domain_heuristic import check_domain_heuristic
 from app.services.normalizer import normalize_url
+from app.services.pipeline_deadline import (
+    PipelineDeadline,
+    PipelineStageTimeoutError,
+    timed_out_content_result,
+    timed_out_domain_result,
+    timed_out_unchain_result,
+)
 from app.services.unchainer import unchain_url
 
 logger = structlog.get_logger(__name__)
@@ -100,6 +107,7 @@ async def run_db_independent_pipeline(
     log = logger.bind(analysis_id=analysis_id, pipeline="db_independent")
     log.info("db_independent_pipeline.start", url=original_url)
     total_started = time.perf_counter()
+    deadline = PipelineDeadline()
     stage_timings = PipelineStageTimings()
 
     stage_started = time.perf_counter()
@@ -118,14 +126,49 @@ async def run_db_independent_pipeline(
     _set_stage_timing(stage_timings, PipelineStage.NORMALIZE, stage_started)
 
     stage_started = time.perf_counter()
-    unchain = await unchain_url(
-        normalize.normalized_url,
-        prefer_https_when_schemeless=normalize.scheme_was_added,
-    )
+    try:
+        unchain = await deadline.run(
+            PipelineStage.UNCHAIN.value,
+            unchain_url(
+                normalize.normalized_url,
+                prefer_https_when_schemeless=normalize.scheme_was_added,
+            ),
+            settings.pipeline_unchain_timeout_seconds,
+        )
+    except PipelineStageTimeoutError:
+        log.warning("db_independent_pipeline.stage_timeout", stage=PipelineStage.UNCHAIN)
+        unchain = timed_out_unchain_result(normalize.normalized_url)
+    except Exception as exc:
+        log.warning(
+            "db_independent_pipeline.stage_error",
+            stage=PipelineStage.UNCHAIN,
+            error=str(exc),
+            error_type=type(exc).__name__,
+        )
+        unchain = timed_out_unchain_result(normalize.normalized_url, error="stage_error")
     _set_stage_timing(stage_timings, PipelineStage.UNCHAIN, stage_started)
 
     stage_started = time.perf_counter()
-    heuristic = await check_domain_heuristic(unchain.final_url)
+    try:
+        heuristic = await deadline.run(
+            PipelineStage.DOMAIN_HEURISTIC.value,
+            check_domain_heuristic(unchain.final_url),
+            settings.pipeline_domain_timeout_seconds,
+        )
+    except PipelineStageTimeoutError:
+        log.warning(
+            "db_independent_pipeline.stage_timeout",
+            stage=PipelineStage.DOMAIN_HEURISTIC,
+        )
+        heuristic = timed_out_domain_result(unchain.final_url)
+    except Exception as exc:
+        log.warning(
+            "db_independent_pipeline.stage_error",
+            stage=PipelineStage.DOMAIN_HEURISTIC,
+            error=str(exc),
+            error_type=type(exc).__name__,
+        )
+        heuristic = timed_out_domain_result(unchain.final_url)
     _set_stage_timing(stage_timings, PipelineStage.DOMAIN_HEURISTIC, stage_started)
 
     if heuristic.score >= settings.score_danger_threshold:
@@ -135,7 +178,26 @@ async def run_db_independent_pipeline(
     else:
         upstream = _collect_db_independent_signals(heuristic, unchain)
         stage_started = time.perf_counter()
-        content = await analyze_content(unchain.final_url, upstream_signals=upstream)
+        try:
+            content = await deadline.run(
+                PipelineStage.CONTENT_ANALYSIS.value,
+                analyze_content(unchain.final_url, upstream_signals=upstream),
+                settings.pipeline_content_timeout_seconds,
+            )
+        except PipelineStageTimeoutError:
+            log.warning(
+                "db_independent_pipeline.stage_timeout",
+                stage=PipelineStage.CONTENT_ANALYSIS,
+            )
+            content = timed_out_content_result(unchain.final_url)
+        except Exception as exc:
+            log.warning(
+                "db_independent_pipeline.stage_error",
+                stage=PipelineStage.CONTENT_ANALYSIS,
+                error=str(exc),
+                error_type=type(exc).__name__,
+            )
+            content = timed_out_content_result(unchain.final_url)
         _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started)
 
     score = _total_score(heuristic, content)
diff --git a/app/services/domain_heuristic/__init__.py b/app/services/domain_heuristic/__init__.py
index f77c8d8..e4821f9 100644
--- a/app/services/domain_heuristic/__init__.py
+++ b/app/services/domain_heuristic/__init__.py
@@ -1,3 +1,13 @@
-from app.services.domain_heuristic.check import check_domain_heuristic
+from __future__ import annotations
+
+from typing import Any
 
 __all__ = ["check_domain_heuristic"]
+
+
+def __getattr__(name: str) -> Any:
+    if name == "check_domain_heuristic":
+        from app.services.domain_heuristic.check import check_domain_heuristic
+
+        return check_domain_heuristic
+    raise AttributeError(name)
diff --git a/app/services/domain_heuristic/check.py b/app/services/domain_heuristic/check.py
index a824764..25aca97 100644
--- a/app/services/domain_heuristic/check.py
+++ b/app/services/domain_heuristic/check.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from functools import lru_cache
 from urllib.parse import urlparse
 
@@ -35,6 +36,11 @@ def _signal_scores() -> dict[DomainHeuristicSignal, int]:
         DomainHeuristicSignal.SUSPICIOUS_TLD: settings.score_weight_suspicious_tld,
         DomainHeuristicSignal.DGA_LIKE: settings.score_weight_dga_like,
         DomainHeuristicSignal.HOSTING_PLATFORM: settings.score_weight_hosting_platform,
+        DomainHeuristicSignal.URL_USERINFO: settings.score_weight_url_userinfo,
+        DomainHeuristicSignal.BRAND_IN_URL: settings.score_weight_brand_in_url,
+        DomainHeuristicSignal.FREE_HOSTING_LURE: settings.score_weight_free_hosting_lure,
+        DomainHeuristicSignal.SENSITIVE_PATH: settings.score_weight_sensitive_path,
+        DomainHeuristicSignal.URL_SHORTENER: settings.score_weight_url_shortener,
     }
 
 
@@ -71,8 +77,19 @@ async def check_domain_heuristic(url: str) -> DomainHeuristicResult:
             except Exception as exc:
                 logger.warning("domain_heuristic.typo_error", error=str(exc))
 
-        # lookup_rdap은 내부에서 모든 예외를 error_code로 변환하므로 재래핑 불필요
-        rdap_info, rdap_error = await lookup_rdap(url)
+        # RDAP은 보조 신호다. RDAP이 느려져도 앞에서 이미 잡은 URL 패턴/DGA/typo
+        # 신호를 잃지 않도록 내부 timeout으로 격리한다.
+        try:
+            timeout_seconds = min(
+                settings.rdap_timeout_seconds,
+                settings.pipeline_domain_timeout_seconds - 0.25,
+            )
+            rdap_info, rdap_error = await asyncio.wait_for(
+                lookup_rdap(url),
+                timeout=timeout_seconds,
+            )
+        except TimeoutError:
+            rdap_info, rdap_error = None, "timeout"
         if rdap_info and rdap_info.is_new_domain:
             signals.append(DomainHeuristicSignal.NEW_DOMAIN)
 
diff --git a/app/services/domain_heuristic/patterns.py b/app/services/domain_heuristic/patterns.py
index 15c4172..c3b3dd9 100644
--- a/app/services/domain_heuristic/patterns.py
+++ b/app/services/domain_heuristic/patterns.py
@@ -1,12 +1,17 @@
 from __future__ import annotations
 
 import ipaddress
+import re
+from functools import lru_cache
+from pathlib import Path
 from urllib.parse import parse_qs, urlparse
 
 from app.core.config import settings
 from app.core.tld import extract_url_parts
 from app.schemas.domain_heuristic import DomainHeuristicSignal
 
+_BRANDS_FILE = Path(__file__).parent / "brands.txt"
+
 _SUSPICIOUS_TLDS = frozenset(
     {
         "zip",
@@ -18,7 +23,9 @@
         "ml",
         "ga",
         "cf",
+        "cfd",
         "gq",
+        "help",
         "fit",
         "work",
         "surf",
@@ -61,6 +68,107 @@
         "fly.dev",
         "railway.app",
         "cyclic.app",
+        "workers.dev",
+        "wasmer.app",
+        "godaddysites.com",
+        "duckdns.org",
+        "free.nf",
+        "my.id",
+        "backblazeb2.com",
+    }
+)
+
+_URL_SHORTENERS = frozenset(
+    {
+        "bit.ly",
+        "cutt.ly",
+        "goo.gl",
+        "is.gd",
+        "ow.ly",
+        "qrco.de",
+        "rebrand.ly",
+        "shorturl.at",
+        "t.co",
+        "tinyurl.com",
+    }
+)
+
+_SENSITIVE_PATH_TOKENS = frozenset(
+    {
+        "account",
+        "admin",
+        "auth",
+        "billing",
+        "claim",
+        "confirm",
+        "credential",
+        "invoice",
+        "login",
+        "password",
+        "pay",
+        "portal",
+        "recover",
+        "secure",
+        "session",
+        "signin",
+        "support",
+        "suporte",
+        "validate",
+        "verification",
+        "verify",
+        "wallet",
+    }
+)
+
+_SEEDED_BRAND_LABELS = frozenset(
+    {
+        "adobe",
+        "amazon",
+        "apple",
+        "binance",
+        "coinbase",
+        "discord",
+        "dpd",
+        "facebook",
+        "google",
+        "instagram",
+        "iphone",
+        "metamask",
+        "microsoft",
+        "netflix",
+        "office",
+        "outlook",
+        "paypal",
+        "roblox",
+        "societegenerale",
+        "telegram",
+        "tmobile",
+        "t-mobile",
+        "trezor",
+        "whatsapp",
+        "wps",
+        "yahoo",
+    }
+)
+
+_GENERIC_BRAND_LABELS = frozenset(
+    {
+        "bank",
+        "blog",
+        "cafe",
+        "card",
+        "cloud",
+        "finance",
+        "gov",
+        "home",
+        "map",
+        "mvno",
+        "news",
+        "pay",
+        "place",
+        "shop",
+        "shopping",
+        "support",
     }
 )
 
@@ -96,6 +204,46 @@ def _is_ip_address(host: str) -> bool:
         return False
 
 
+@lru_cache(maxsize=1)
+def _brand_labels() -> frozenset[str]:
+    labels = set(_SEEDED_BRAND_LABELS)
+    if not _BRANDS_FILE.exists():
+        return frozenset(labels)
+    for line in _BRANDS_FILE.read_text(encoding="utf-8").splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        label = line.split(".", 1)[0].lower()
+        if len(label) >= 3 and label not in _GENERIC_BRAND_LABELS:
+            labels.add(label)
+    return frozenset(labels)
+
+
+@lru_cache(maxsize=1)
+def _trusted_brand_domains() -> frozenset[str]:
+    domains: set[str] = set()
+    if not _BRANDS_FILE.exists():
+        return frozenset(domains)
+    for line in _BRANDS_FILE.read_text(encoding="utf-8").splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        domains.add(line.lower())
+    return frozenset(domains)
+
+
+def _brand_labels_in_url_text(text: str) -> set[str]:
+    lowered = text.lower()
+    tokens = re.findall(r"[a-z0-9]+", lowered)
+    found: set[str] = set()
+    for label in _brand_labels():
+        if len(label) < 3:
+            continue
+        if label in lowered or label in tokens:
+            found.add(label)
+    return found
+
+
 def check_patterns(url: str) -> list[DomainHeuristicSignal]:
     signals: list[DomainHeuristicSignal] = []
     parsed = urlparse(url)
@@ -109,6 +257,12 @@ def check_patterns(url: str) -> list[DomainHeuristicSignal]:
         signals.append(DomainHeuristicSignal.NO_HTTPS)
 
     ext = extract_url_parts(url)
+    registered_domain = (ext.top_domain_under_public_suffix or host).lower()
+    registered_label = ext.domain.lower()
+    is_trusted_brand_domain = registered_domain in _trusted_brand_domains()
+
+    if parsed.username or parsed.password or "@" in parsed.netloc:
+        signals.append(DomainHeuristicSignal.URL_USERINFO)
 
     # 서브도메인 레이블 과다
     subdomain_labels = [s for s in ext.subdomain.split(".") if s]
@@ -137,10 +291,32 @@ def check_patterns(url: str) -> list[DomainHeuristicSignal]:
     if any(k.lower() in _OPEN_REDIRECT_PARAMS for k in qs):
         signals.append(DomainHeuristicSignal.OPEN_REDIRECT_PARAM)
 
-    # 합법 호스팅 플랫폼 (domain.suffix 조합으로 검사)
-    if ext.suffix and ext.subdomain:
+    # private suffix(github.io, vercel.app 등)는 tld parser에서 suffix로 잡히므로
+    # user.github.io처럼 subdomain이 없어도 플랫폼 사용자 공간으로 본다.
+    is_hosting_platform = False
+    if ext.suffix:
         domain_suffix = f"{ext.domain}.{ext.suffix}"
-        if domain_suffix in HOSTING_PLATFORMS or ext.suffix in HOSTING_PLATFORMS:
+        if ext.suffix in HOSTING_PLATFORMS or (
+            ext.subdomain and domain_suffix in HOSTING_PLATFORMS
+        ):
             signals.append(DomainHeuristicSignal.HOSTING_PLATFORM)
+            is_hosting_platform = True
+
+    if registered_domain in _URL_SHORTENERS:
+        signals.append(DomainHeuristicSignal.URL_SHORTENER)
+
+    url_text = " ".join(part for part in (host, parsed.path, parsed.query) if part)
+    url_tokens = set(re.findall(r"[a-z]+", url_text.lower()))
+    sensitive_path = bool(url_tokens & _SENSITIVE_PATH_TOKENS)
+    url_brands = _brand_labels_in_url_text(url_text)
+
+    if sensitive_path and not is_trusted_brand_domain:
+        signals.append(DomainHeuristicSignal.SENSITIVE_PATH)
+
+    if url_brands and not is_trusted_brand_domain and registered_label not in url_brands:
+        signals.append(DomainHeuristicSignal.BRAND_IN_URL)
+
+    if is_hosting_platform and (url_brands or sensitive_path):
+        signals.append(DomainHeuristicSignal.FREE_HOSTING_LURE)
 
     return signals
diff --git a/app/services/domain_heuristic/rdap.py b/app/services/domain_heuristic/rdap.py
index eb3172a..53ed54c 100644
--- a/app/services/domain_heuristic/rdap.py
+++ b/app/services/domain_heuristic/rdap.py
@@ -223,10 +223,15 @@ async def lookup_rdap(url: str) -> tuple[RdapInfo | None, str | None]:
     _inflight[domain] = fut
     try:
         result = await _fetch_rdap(domain)
-        fut.set_result(result)
+        if not fut.done():
+            fut.set_result(result)
         return result
     except BaseException as exc:
-        fut.set_exception(exc)
+        if not fut.done():
+            fut.set_exception(exc)
+            # If the owner task is cancelled before another waiter consumes this future,
+            # retrieve the exception to avoid noisy "Future exception was never retrieved".
+            fut.add_done_callback(lambda done: done.exception())
         raise
     finally:
         _inflight.pop(domain, None)
diff --git a/app/services/domain_heuristic/typosquatting.py b/app/services/domain_heuristic/typosquatting.py
index 33611eb..6475d46 100644
--- a/app/services/domain_heuristic/typosquatting.py
+++ b/app/services/domain_heuristic/typosquatting.py
@@ -35,7 +35,14 @@ def _load_brands() -> list[tuple[str, str]]:
     return brands
 
 
-_BRANDS: list[tuple[str, str]] = _load_brands()
+_BRANDS: list[tuple[str, str]] | None = None
+
+
+def _get_brands() -> list[tuple[str, str]]:
+    global _BRANDS
+    if _BRANDS is None:
+        _BRANDS = _load_brands()
+    return _BRANDS
 
 
 def _levenshtein(a: str, b: str) -> int:
@@ -56,6 +63,35 @@ def _levenshtein(a: str, b: str) -> int:
 
 
 _MIN_FUZZY_LEN = 5  # 4자 이하 브랜드는 편집거리 1~2 오탐이 심해 완전일치만 인정
+_SUSPICIOUS_SUFFIXES = frozenset(
+    {
+        "bid",
+        "cf",
+        "click",
+        "date",
+        "faith",
+        "ga",
+        "gq",
+        "link",
+        "loan",
+        "men",
+        "ml",
+        "mov",
+        "party",
+        "pw",
+        "racing",
+        "review",
+        "science",
+        "stream",
+        "surf",
+        "tk",
+        "top",
+        "trade",
+        "win",
+        "xyz",
+        "zip",
+    }
+)
 
 
 def check_typosquatting(url: str) -> DomainHeuristicSignal | None:
@@ -72,10 +108,13 @@ def check_typosquatting(url: str) -> DomainHeuristicSignal | None:
         return None
 
     is_typo = False
-    for brand_domain, brand_suffix in _BRANDS:
+    for brand_domain, brand_suffix in _get_brands():
         if target_domain == brand_domain and target_suffix == brand_suffix:
             return None  # 완전 일치 → 정상 도메인
 
+        if target_suffix != brand_suffix and target_suffix not in _SUSPICIOUS_SUFFIXES:
+            continue
+
         if abs(len(target_domain) - len(brand_domain)) > 2:
             continue  # 길이 차이가 2 초과면 거리가 반드시 2 초과 → 스킵
 
diff --git a/app/services/evaluation/__init__.py b/app/services/evaluation/__init__.py
new file mode 100644
index 0000000..926936f
--- /dev/null
+++ b/app/services/evaluation/__init__.py
@@ -0,0 +1 @@
+"""Evaluation helpers for detector quality measurement."""
diff --git a/app/services/evaluation/metrics.py b/app/services/evaluation/metrics.py
new file mode 100644
index 0000000..011be0d
--- /dev/null
+++ b/app/services/evaluation/metrics.py
@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from collections.abc import Iterable, Set
+from dataclasses import dataclass
+
+from app.schemas.pipeline import Verdict
+
+
+@dataclass(frozen=True)
+class VerdictCase:
+    expected: Verdict
+    predicted: Verdict
+
+
+@dataclass(frozen=True)
+class VerdictMetrics:
+    total: int
+    correct: int
+    true_positive: int
+    false_positive: int
+    true_negative: int
+    false_negative: int
+    accuracy: float
+    positive_precision: float
+    positive_recall: float
+    positive_f1: float
+    false_positive_rate: float
+
+
+def _ratio(numerator: int, denominator: int) -> float:
+    if denominator == 0:
+        return 0.0
+    return round(numerator / denominator, 6)
+
+
+def compute_verdict_metrics(
+    cases: Iterable[VerdictCase],
+    *,
+    positive_expected: Set[Verdict],
+    positive_predicted: Set[Verdict],
+) -> VerdictMetrics:
+    rows = list(cases)
+    total = len(rows)
+    correct = sum(1 for row in rows if row.expected == row.predicted)
+
+    true_positive = 0
+    false_positive = 0
+    true_negative = 0
+    false_negative = 0
+    for row in rows:
+        expected_positive = row.expected in positive_expected
+        predicted_positive = row.predicted in positive_predicted
+        if expected_positive and predicted_positive:
+            true_positive += 1
+        elif not expected_positive and predicted_positive:
+            false_positive += 1
+        elif expected_positive and not predicted_positive:
+            false_negative += 1
+        else:
+            true_negative += 1
+
+    precision = _ratio(true_positive, true_positive + false_positive)
+    recall = _ratio(true_positive, true_positive + false_negative)
+    f1 = (
+        0.0
+        if precision + recall == 0
+        else round((2 * precision * recall) / (precision + recall), 6)
+    )
+
+    return VerdictMetrics(
+        total=total,
+        correct=correct,
+        true_positive=true_positive,
+        false_positive=false_positive,
+        true_negative=true_negative,
+        false_negative=false_negative,
+        accuracy=_ratio(correct, total),
+        positive_precision=precision,
+        positive_recall=recall,
+        positive_f1=f1,
+        false_positive_rate=_ratio(false_positive, false_positive + true_negative),
+    )
diff --git a/app/services/pipeline.py b/app/services/pipeline.py
index f92b421..0b0eb01 100644
--- a/app/services/pipeline.py
+++ b/app/services/pipeline.py
@@ -31,6 +31,14 @@
 from app.services.content_analyzer import analyze_content, skipped_already_danger
 from app.services.domain_heuristic import check_domain_heuristic
 from app.services.normalizer import normalize_url
+from app.services.pipeline_deadline import (
+    PipelineDeadline,
+    PipelineStageTimeoutError,
+    timed_out_content_result,
+    timed_out_domain_result,
+    timed_out_threat_db_result,
+    timed_out_unchain_result,
+)
 from app.services.threat_db import check_threat_db
 from app.services.unchainer import unchain_url
 
@@ -279,6 +287,7 @@ async def run_pipeline(
     log = logger.bind(analysis_id=analysis_id)
     log.info("pipeline.start", url=original_url)
     total_started = time.perf_counter()
+    deadline = PipelineDeadline()
     stage_timings = PipelineStageTimings()
 
     stage_started = time.perf_counter()
@@ -296,17 +305,60 @@ async def run_pipeline(
         )
     _set_stage_timing(stage_timings, PipelineStage.NORMALIZE, stage_started)
 
-    unchain: UnchainResult = await _timed_async_stage(
-        stage_timings,
-        PipelineStage.UNCHAIN,
-        unchain_url(norm.normalized_url, prefer_https_when_schemeless=norm.scheme_was_added),
-    )
+    stage_started = time.perf_counter()
+    try:
+        unchain: UnchainResult = await deadline.run(
+            PipelineStage.UNCHAIN.value,
+            _timed_async_stage(
+                stage_timings,
+                PipelineStage.UNCHAIN,
+                unchain_url(
+                    norm.normalized_url,
+                    prefer_https_when_schemeless=norm.scheme_was_added,
+                ),
+            ),
+            settings.pipeline_unchain_timeout_seconds,
+        )
+    except PipelineStageTimeoutError:
+        log.warning("pipeline.stage_timeout", stage=PipelineStage.UNCHAIN)
+        unchain = timed_out_unchain_result(norm.normalized_url)
+        if stage_timings.unchain is None:
+            _set_stage_timing(stage_timings, PipelineStage.UNCHAIN, stage_started)
+    except Exception as exc:
+        log.warning(
+            "pipeline.stage_error",
+            stage=PipelineStage.UNCHAIN,
+            error=str(exc),
+            error_type=type(exc).__name__,
+        )
+        unchain = timed_out_unchain_result(norm.normalized_url, error="stage_error")
+        if stage_timings.unchain is None:
+            _set_stage_timing(stage_timings, PipelineStage.UNCHAIN, stage_started)
+
     # 2·3단계는 둘 다 unchain.final_url 만 필요하고 서로 독립이라 병렬로 돈다.
     # threat_db 가 먼저 malicious 로 끝나면 verdict 가 이미 danger 로 확정이므로
     # heuristic 을 cancel 하고 4단계까지 skip — 여기서 조기 종료가 일어난다.
-    threat, heuristic, short_circuited = await _run_stage_2_and_3(
-        log, unchain.final_url, session, stage_timings
-    )
+    try:
+        threat, heuristic, short_circuited = await deadline.run(
+            "reputation",
+            _run_stage_2_and_3(log, unchain.final_url, session, stage_timings),
+            settings.pipeline_reputation_timeout_seconds,
+        )
+    except PipelineStageTimeoutError:
+        log.warning("pipeline.stage_timeout", stage="reputation")
+        threat = timed_out_threat_db_result(unchain.final_url)
+        heuristic = timed_out_domain_result(unchain.final_url)
+        short_circuited = False
+    except Exception as exc:
+        log.warning(
+            "pipeline.stage_error",
+            stage="reputation",
+            error=str(exc),
+            error_type=type(exc).__name__,
+        )
+        threat = timed_out_threat_db_result(unchain.final_url)
+        heuristic = timed_out_domain_result(unchain.final_url)
+        short_circuited = False
 
     if short_circuited:
         log.info(
@@ -333,11 +385,27 @@ async def run_pipeline(
             _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started)
         else:
             upstream = _collect_upstream_signals(threat, heuristic)
-            content = await _timed_async_stage(
-                stage_timings,
-                PipelineStage.CONTENT_ANALYSIS,
-                _stage_content_analysis(log, unchain.final_url, upstream),
-            )
+            try:
+                content = await deadline.run(
+                    PipelineStage.CONTENT_ANALYSIS.value,
+                    _timed_async_stage(
+                        stage_timings,
+                        PipelineStage.CONTENT_ANALYSIS,
+                        _stage_content_analysis(log, unchain.final_url, upstream),
+                    ),
+                    settings.pipeline_content_timeout_seconds,
+                )
+            except PipelineStageTimeoutError:
+                log.warning("pipeline.stage_timeout", stage=PipelineStage.CONTENT_ANALYSIS)
+                content = timed_out_content_result(unchain.final_url)
+            except Exception as exc:
+                log.warning(
+                    "pipeline.stage_error",
+                    stage=PipelineStage.CONTENT_ANALYSIS,
+                    error=str(exc),
+                    error_type=type(exc).__name__,
+                )
+                content = timed_out_content_result(unchain.final_url)
 
     score = _total_score(threat, heuristic, content)
     verdict = _decide_verdict(score, threat)
diff --git a/app/services/pipeline_deadline.py b/app/services/pipeline_deadline.py
new file mode 100644
index 0000000..73846cc
--- /dev/null
+++ b/app/services/pipeline_deadline.py
@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from collections.abc import Awaitable
+from typing import TypeVar
+from urllib.parse import urlparse
+
+from app.core.config import settings
+from app.core.tld import extract_url_parts
+from app.schemas.content_analysis import ContentAnalysisResult, ContentSignal
+from app.schemas.domain_heuristic import DomainHeuristicResult
+from app.schemas.threat_db import GSBResult, ThreatDbResult, URLhausResult
+from app.schemas.unchain import UnchainResult
+
+T = TypeVar("T")
+
+
+class PipelineStageTimeoutError(TimeoutError):
+    def __init__(self, stage: str) -> None:
+        super().__init__(f"{stage} timed out")
+        self.stage = stage
+
+
+class PipelineDeadline:
+    def __init__(self, total_seconds: float | None = None) -> None:
+        self._started = time.perf_counter()
+        self._total_seconds = total_seconds or settings.pipeline_total_timeout_seconds
+
+    def remaining(self) -> float:
+        return max(0.0, self._total_seconds - (time.perf_counter() - self._started))
+
+    def budget(self, stage_seconds: float) -> float:
+        # 작은 양수로 wait_for까지 진입시켜 stage timeout 경로를 일관되게 탄다.
+        return max(0.001, min(stage_seconds, self.remaining()))
+
+    async def run(self, stage: str, coro: Awaitable[T], stage_seconds: float) -> T:
+        try:
+            return await asyncio.wait_for(coro, timeout=self.budget(stage_seconds))
+        except TimeoutError as exc:
+            raise PipelineStageTimeoutError(stage) from exc
+
+
+def timed_out_unchain_result(url: str, *, error: str = "pipeline_timeout") -> UnchainResult:
+    return UnchainResult(
+        input_url=url,
+        final_url=url,
+        hops=[],
+        hop_count=0,
+        timed_out=True,
+        error=error,
+        signals=[error],
+    )
+
+
+def timed_out_threat_db_result(final_url: str) -> ThreatDbResult:
+    return ThreatDbResult(
+        final_url=final_url,
+        is_malicious=False,
+        sources_checked=0,
+        gsb=GSBResult(checked=False, is_threat=False, error="pipeline_timeout"),
+        urlhaus=URLhausResult(checked=False, is_threat=False, error="pipeline_timeout"),
+        threat_types=[],
+    )
+
+
+def timed_out_domain_result(final_url: str) -> DomainHeuristicResult:
+    ext = extract_url_parts(final_url)
+    domain = ext.top_domain_under_public_suffix or (urlparse(final_url).hostname or "")
+    return DomainHeuristicResult(
+        domain=domain,
+        score=0,
+        signals=[],
+        rdap=None,
+        rdap_error="pipeline_timeout",
+    )
+
+
+def timed_out_content_result(final_url: str) -> ContentAnalysisResult:
+    return ContentAnalysisResult(
+        final_url=final_url,
+        fetched=False,
+        status_code=None,
+        score=settings.score_weight_content_fetch_failed,
+        signals=[ContentSignal.FETCH_FAILED],
+        reason="콘텐츠 분석 시간이 초과되었습니다.",
+        error="pipeline_timeout",
+    )
diff --git a/pyproject.toml b/pyproject.toml
index f9d102d..9acba7b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -20,6 +20,7 @@ dependencies = [
     "lxml>=5.3.0",
     "cachetools>=5.5.0",
     "openai>=1.60.0",
+    "playwright>=1.45.0",
 ]
 
 [project.optional-dependencies]
diff --git a/tests/api/test_stages.py b/tests/api/test_stages.py
index 830d383..756510c 100644
--- a/tests/api/test_stages.py
+++ b/tests/api/test_stages.py
@@ -63,6 +63,7 @@ async def test_returns_fetch_and_features_on_success(self, client: httpx.AsyncCl
         assert body["features"]["title"] == "NAVER"
         assert body["features"]["has_password_field"] is True
         assert body["features"]["image_alts"] == ["NAVER logo"]
+        assert body["features"]["form_field_summaries"]
         assert body["html_preview"] is not None
         assert "<title>NAVER</title>" in body["html_preview"]
 
diff --git a/tests/core/test_tld.py b/tests/core/test_tld.py
index 59e083b..965f6d8 100644
--- a/tests/core/test_tld.py
+++ b/tests/core/test_tld.py
@@ -1,20 +1,21 @@
 from __future__ import annotations
 
-import pytest
-import tldextract
 
-
-def test_extract_url_parts_does_not_use_default_disk_cached_extractor(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
+def test_extract_url_parts_handles_multi_label_suffix() -> None:
     from app.core.tld import extract_url_parts
 
-    def fail_default_extractor(*_args: object, **_kwargs: object) -> None:
-        raise AssertionError("default tldextract extractor should not be used")
-
-    monkeypatch.setattr(tldextract, "extract", fail_default_extractor)
-
     result = extract_url_parts("https://login.example.co.kr/path")
 
     assert result.domain == "example"
     assert result.suffix == "co.kr"
+    assert result.subdomain == "login"
+
+
+def test_extract_url_parts_handles_private_hosting_suffix() -> None:
+    from app.core.tld import extract_url_parts
+
+    result = extract_url_parts("https://shop-login.vercel.app/path")
+
+    assert result.domain == "shop-login"
+    assert result.suffix == "vercel.app"
+    assert result.top_domain_under_public_suffix == "shop-login.vercel.app"
diff --git a/tests/services/content_analyzer/test_ai_openai.py b/tests/services/content_analyzer/test_ai_openai.py
index 28f6ddc..2336294 100644
--- a/tests/services/content_analyzer/test_ai_openai.py
+++ b/tests/services/content_analyzer/test_ai_openai.py
@@ -201,6 +201,42 @@ async def test_openai_user_prompt_includes_upstream_signals() -> None:
     assert body["upstream_signals"] == ["TYPO_DOMAIN", "NEW_DOMAIN"]
 
 
+async def test_openai_user_prompt_includes_high_signal_structured_context() -> None:
+    client = _mock_http_client(
+        json.dumps({"verdict": "suspicious", "reason": "민감정보 입력 유도"})
+    )
+    ctx = AIPromptContext(
+        final_url="https://nhis-support.test/",
+        title="고유가 피해지원금 대상 조회",
+        has_password_field=False,
+        has_meta_refresh=False,
+        image_alts=(),
+        external_link_ratio=None,
+        body_text_snippets=("국민건강보험 고유가 피해지원금 지급대상 여부 조회",),
+        form_field_summaries=(
+            "label=주민등록번호 name=resident_registration_number placeholder=주민등록번호",
+        ),
+        cta_texts=("지원금 대상 조회하기",),
+        download_links=(),
+        sensitive_field_types=("resident_registration_number",),
+        korean_lure_keywords=("지원금",),
+        public_agency_keywords=("국민건강보험",),
+    )
+    with patch("app.services.content_analyzer.ai_openai.httpx.AsyncClient", return_value=client):
+        await OpenAIProvider().infer(ctx)
+
+    user_msg = client.post.await_args.kwargs["json"]["messages"][1]["content"]
+    body = json.loads(user_msg)
+    assert body["body_text"] == ["국민건강보험 고유가 피해지원금 지급대상 여부 조회"]
+    assert body["form_fields"] == [
+        "label=주민등록번호 name=resident_registration_number placeholder=주민등록번호"
+    ]
+    assert body["cta_texts"] == ["지원금 대상 조회하기"]
+    assert body["sensitive_field_types"] == ["resident_registration_number"]
+    assert body["korean_lure_keywords"] == ["지원금"]
+    assert body["public_agency_keywords"] == ["국민건강보험"]
+
+
 async def test_openai_user_prompt_omits_empty_upstream_signals() -> None:
     """기본값(빈 튜플)이면 빈 배열로 직렬화돼 단독 페이지 분석 모드와 동일하게 동작."""
     client = _mock_http_client(json.dumps({"verdict": "benign", "reason": "ok"}))
diff --git a/tests/services/content_analyzer/test_analyze.py b/tests/services/content_analyzer/test_analyze.py
index 8e165aa..5601963 100644
--- a/tests/services/content_analyzer/test_analyze.py
+++ b/tests/services/content_analyzer/test_analyze.py
@@ -11,6 +11,7 @@
 from app.services.content_analyzer.ai import AIInference, AIPromptContext, NullAIProvider
 from app.services.content_analyzer.analyze import analyze_content, skipped_already_danger
 from app.services.content_analyzer.fetch import FetchResult
+from app.services.content_analyzer.render import RenderResult
 
 
 def _mock_fetch(ok: bool, html: str = "", error: str | None = None, status: int = 200):
@@ -126,6 +127,71 @@ async def test_brand_impersonation_form_scored(self) -> None:
         assert result.has_password_field is True
 
 
+class TestHighSignalContextEndToEnd:
+    async def test_high_signal_features_surface_and_ai_sees_structured_summary(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        seen: list[AIPromptContext] = []
+
+        class CaptureAI:
+            async def infer(self, ctx: AIPromptContext) -> AIInference:
+                seen.append(ctx)
+                return AIInference(verdict=AIVerdict.BENIGN, reason="local signals enough")
+
+        monkeypatch.setattr(
+            "app.services.content_analyzer.analyze.get_ai_provider",
+            lambda: CaptureAI(),
+        )
+        html = """
+        <html>
+          <head><title>고유가 피해지원금 대상 조회</title></head>
+          <body>
+            <h1>국민건강보험 고유가 피해지원금 지급대상 여부 조회</h1>
+            <form>
+              <label for="rrn">주민등록번호</label>
+              <input id="rrn" name="resident_registration_number" placeholder="주민등록번호">
+              <button type="button">지원금 대상 조회하기</button>
+            </form>
+          </body>
+        </html>
+        """
+        with _mock_fetch(ok=True, html=html):
+            result = await analyze_content("https://nhis-support.test/")
+
+        assert result.body_text_snippets
+        assert "지원금 대상 조회하기" in result.cta_texts
+        assert "resident_registration_number" in result.sensitive_field_types
+        assert ContentSignal.PII_COLLECTION_FORM in result.signals
+        assert len(seen) == 1
+        assert "지원금 대상 조회하기" in seen[0].cta_texts
+        assert "resident_registration_number" in seen[0].sensitive_field_types
+        assert "국민건강보험" in seen[0].public_agency_keywords
+
+    async def test_ai_provider_still_called_once_with_extended_context(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        calls = 0
+
+        class CountingAI:
+            async def infer(self, ctx: AIPromptContext) -> AIInference:
+                nonlocal calls
+                calls += 1
+                return AIInference(verdict=AIVerdict.SUSPICIOUS, reason="structured context")
+
+        monkeypatch.setattr(
+            "app.services.content_analyzer.analyze.get_ai_provider",
+            lambda: CountingAI(),
+        )
+        html = (
+            '<html><body><a href="/downloads/kakaotalk.apk">'
+            "카카오톡 최신버전 다운로드</a></body></html>"
+        )
+        with _mock_fetch(ok=True, html=html):
+            await analyze_content("https://obituary.test/")
+
+        assert calls == 1
+
+
 class TestAI:
     async def test_ai_phishing_verdict_adds_score(self, monkeypatch: pytest.MonkeyPatch) -> None:
         class StubAI:
@@ -267,6 +333,79 @@ async def infer(self, ctx: AIPromptContext) -> AIInference:
         assert seen[0].is_spa_shell is True
 
 
+class TestPrecisionSlowPath:
+    async def test_spa_shell_uses_rendered_dom_for_scoring(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """정적 HTML 이 SPA 셸뿐이어도 렌더링 DOM 에 민감 폼이 있으면 공통 룰로 재점수화한다."""
+
+        class BenignAI:
+            async def infer(self, ctx: AIPromptContext) -> AIInference:
+                return AIInference(verdict=AIVerdict.BENIGN, reason="local evidence")
+
+        static_shell = (
+            '<!doctype html><html><head><title>조회</title>'
+            '<script type="module" src="/app.js"></script></head>'
+            '<body><div id="root"></div></body></html>'
+        )
+        rendered_html = """
+        <html>
+          <head><title>고유가 피해지원금 대상 조회</title></head>
+          <body>
+            <h1>국민건강보험 고유가 피해지원금 지급대상 여부 조회</h1>
+            <form>
+              <label for="rrn">주민등록번호</label>
+              <input id="rrn" name="resident_registration_number" placeholder="주민등록번호">
+              <button type="button">지원금 대상 조회하기</button>
+            </form>
+          </body>
+        </html>
+        """
+
+        monkeypatch.setattr(
+            "app.services.content_analyzer.analyze.get_ai_provider",
+            lambda: BenignAI(),
+        )
+        with (
+            _mock_fetch(ok=True, html=static_shell),
+            patch(
+                "app.services.content_analyzer.analyze.render_page",
+                AsyncMock(
+                    return_value=RenderResult(
+                        ok=True,
+                        url="https://spa.test/",
+                        html=rendered_html,
+                    )
+                ),
+            ) as mock_render,
+        ):
+            result = await analyze_content("https://spa.test/")
+
+        mock_render.assert_awaited_once_with("https://spa.test/")
+        assert result.precision_analysis_used is True
+        assert result.precision_analysis_error is None
+        assert result.is_spa_shell is False
+        assert ContentSignal.RENDERED_DOM_ANALYZED in result.signals
+        assert ContentSignal.PII_COLLECTION_FORM in result.signals
+        assert "resident_registration_number" in result.sensitive_field_types
+        assert result.score >= settings.score_danger_threshold
+
+    async def test_clean_static_page_skips_precision_slow_path(self) -> None:
+        html = "<html><head><title>Hello</title></head><body><p>ok</p></body></html>"
+        with (
+            _mock_fetch(ok=True, html=html),
+            patch(
+                "app.services.content_analyzer.analyze.render_page",
+                new_callable=AsyncMock,
+            ) as mock_render,
+        ):
+            result = await analyze_content("https://benign.test/")
+
+        mock_render.assert_not_awaited()
+        assert result.precision_analysis_used is False
+        assert result.precision_analysis_error is None
+
+
 class TestCancelledError:
     async def test_fetch_cancelled_propagates(self) -> None:
         """shutdown/timeout 신호는 degraded 결과로 흡수하지 않고 re-raise."""
diff --git a/tests/services/content_analyzer/test_extract.py b/tests/services/content_analyzer/test_extract.py
index 2c4dea5..5655a22 100644
--- a/tests/services/content_analyzer/test_extract.py
+++ b/tests/services/content_analyzer/test_extract.py
@@ -160,6 +160,74 @@ def test_relative_resolved_against_base(self) -> None:
         assert features.external_link_ratio == 0.0
 
 
+class TestHighSignalStaticFeatures:
+    def test_extracts_body_text_form_fields_and_ctas(self) -> None:
+        html = """
+        <html>
+          <head><title>고유가 피해지원금 대상 조회</title></head>
+          <body>
+            <main>
+              <h1>국민건강보험 고유가 피해지원금 지급대상 여부 조회</h1>
+              <form action="https://example.invalid/pii">
+                <label for="rrn">주민등록번호</label>
+                <input id="rrn" name="resident_registration_number" placeholder="주민등록번호">
+                <label for="phone">휴대폰 번호</label>
+                <input id="phone" name="mobile_phone" placeholder="010-0000-0000">
+                <button type="button">지원금 대상 조회하기</button>
+              </form>
+            </main>
+          </body>
+        </html>
+        """
+
+        features = extract_features(html, base_url="https://nhis-support.test/")
+
+        assert any("고유가 피해지원금" in text for text in features.body_text_snippets)
+        assert any(
+            "resident_registration_number" in field
+            for field in features.form_field_summaries
+        )
+        assert any("주민등록번호" in field for field in features.form_field_summaries)
+        assert "지원금 대상 조회하기" in features.cta_texts
+        assert "resident_registration_number" in features.sensitive_field_types
+        assert "phone" in features.sensitive_field_types
+        assert "지원금" in features.korean_lure_keywords
+        assert "국민건강보험" in features.public_agency_keywords
+
+    def test_extracts_risky_download_links_and_lure_text(self) -> None:
+        html = """
+        <html>
+          <body>
+            <p>故 홍길동님의 모바일 부고장을 카카오톡으로 확인하세요.</p>
+            <a href="/downloads/kakaotalk.apk" download>카카오톡 최신버전 다운로드</a>
+            <a href="/notice.pdf">일반 안내문</a>
+          </body>
+        </html>
+        """
+
+        features = extract_features(html, base_url="https://obituary.test/")
+
+        assert features.download_links == ["https://obituary.test/downloads/kakaotalk.apk"]
+        assert "부고" in features.korean_lure_keywords
+        assert "카카오톡" in features.korean_lure_keywords
+        assert "카카오톡 최신버전 다운로드" in features.cta_texts
+
+    def test_extraction_caps_high_signal_lists(self) -> None:
+        inputs = "".join(
+            f'<label for="i{i}">휴대폰 번호 {i}</label>'
+            f'<input id="i{i}" name="mobile_phone_{i}" placeholder="휴대폰">'
+            for i in range(200)
+        )
+        links = "".join(f'<a href="/app-{i}.apk">download {i}</a>' for i in range(200))
+        html = f"<html><body><form>{inputs}</form>{links}</body></html>"
+
+        features = extract_features(html, base_url="https://x.test/")
+
+        assert len(features.form_field_summaries) <= 80
+        assert len(features.download_links) <= 40
+        assert len(features.body_text_snippets) <= 40
+
+
 class TestSpaShellDetection:
     def test_react_vite_shell_detected(self) -> None:
         html = (
diff --git a/tests/services/content_analyzer/test_signals.py b/tests/services/content_analyzer/test_signals.py
index 85d865a..d199547 100644
--- a/tests/services/content_analyzer/test_signals.py
+++ b/tests/services/content_analyzer/test_signals.py
@@ -157,6 +157,89 @@ def test_none_ratio_no_signal(self) -> None:
         assert ContentSignal.EXTERNAL_LINK_OVERUSE not in result.signals
 
 
+class TestHighSignalKoreanPhishingRules:
+    def test_public_agency_subsidy_resident_number_form_scores_strongly(self) -> None:
+        features = _features(
+            title="고유가 피해지원금 대상 여부 조회",
+            body_text_snippets=["국민건강보험 고유가 피해지원금 지급대상 여부를 조회하세요."],
+            form_field_summaries=[
+                "label=주민등록번호 name=resident_registration_number placeholder=주민등록번호",
+                "label=휴대폰 번호 name=mobile_phone placeholder=010-0000-0000",
+            ],
+            cta_texts=["지원금 대상 조회하기"],
+            sensitive_field_types=["resident_registration_number", "phone"],
+            korean_lure_keywords=["지원금"],
+            public_agency_keywords=["국민건강보험"],
+        )
+
+        result = score_content(features, final_url="https://nhis-support.test/")
+
+        assert ContentSignal.PII_COLLECTION_FORM in result.signals
+        assert ContentSignal.SENSITIVE_ID_FIELD in result.signals
+        assert ContentSignal.PUBLIC_AGENCY_LURE in result.signals
+        assert ContentSignal.KOREAN_LURE_TEXT in result.signals
+        assert result.score >= settings.score_danger_threshold
+
+    def test_kakao_obituary_apk_download_scores_without_credentials(self) -> None:
+        features = _features(
+            body_text_snippets=["故 홍길동님의 모바일 부고장을 카카오톡으로 확인하세요."],
+            cta_texts=["카카오톡 최신버전 다운로드"],
+            download_links=["https://obituary.test/downloads/kakaotalk.apk"],
+            korean_lure_keywords=["부고", "카카오톡"],
+        )
+
+        result = score_content(features, final_url="https://obituary.test/")
+
+        assert ContentSignal.RISKY_DOWNLOAD_LINK in result.signals
+        assert ContentSignal.KOREAN_LURE_TEXT in result.signals
+        assert ContentSignal.PII_COLLECTION_FORM not in result.signals
+        assert result.score >= settings.score_caution_threshold
+
+    def test_financial_card_fields_score_but_readonly_free_text_alone_does_not(self) -> None:
+        readonly_like_text_only = _features(
+            body_text_snippets=["카드번호, CVC, 만료일 안내"],
+            korean_lure_keywords=[],
+        )
+        text_result = score_content(readonly_like_text_only, final_url="https://safe.test/")
+        assert ContentSignal.FINANCIAL_FIELD not in text_result.signals
+        assert text_result.score == 0
+
+        actual_form = _features(
+            form_field_summaries=[
+                "label=카드번호 name=card_number placeholder=카드번호 16자리",
+                "label=CVC name=card_cvc placeholder=CVC",
+            ],
+            sensitive_field_types=["card", "cvc"],
+        )
+        form_result = score_content(actual_form, final_url="https://pay-check.test/")
+        assert ContentSignal.FINANCIAL_FIELD in form_result.signals
+        assert form_result.score >= settings.score_weight_financial_field
+
+    def test_legitimate_login_with_birthdate_guidance_does_not_gain_new_scores(self) -> None:
+        features = _features(
+            title="명지대학교 통합로그인",
+            has_password_field=True,
+            body_text_snippets=[
+                "최초 사용자는 주민등록 기재된 생년월일 6자리 입력 후 비밀번호를 변경해야 함"
+            ],
+            form_field_summaries=[
+                "label=아이디 name=id placeholder=아이디",
+                "label=비밀번호 name=pw type=password placeholder=비밀번호",
+            ],
+            cta_texts=["로그인"],
+            sensitive_field_types=[],
+            korean_lure_keywords=[],
+            public_agency_keywords=[],
+        )
+
+        result = score_content(features, final_url="https://sso.mju.ac.kr/")
+
+        assert ContentSignal.PII_COLLECTION_FORM not in result.signals
+        assert ContentSignal.SENSITIVE_ID_FIELD not in result.signals
+        assert ContentSignal.KOREAN_LURE_TEXT not in result.signals
+        assert result.score == 0
+
+
 class TestSpaShellSignal:
     def test_spa_shell_emits_signal_without_score(self) -> None:
         """판정 불가 플래그라 점수는 0, 시그널 코드만 올라간다."""
diff --git a/tests/services/domain_heuristic/test_check.py b/tests/services/domain_heuristic/test_check.py
index db9b9a6..0ff1da7 100644
--- a/tests/services/domain_heuristic/test_check.py
+++ b/tests/services/domain_heuristic/test_check.py
@@ -21,14 +21,14 @@ async def test_ip_direct_url():
 
 @pytest.mark.asyncio
 async def test_score_accumulates():
-    # http + .xyz TLD → NO_HTTPS(30) + SUSPICIOUS_TLD(20) = 50
+    # http + .xyz TLD → NO_HTTPS(20) + SUSPICIOUS_TLD(25) = 45
     with patch(_RDAP_PATH, new_callable=AsyncMock) as mock_rdap:
         mock_rdap.return_value = (None, "not_found")
         result = await check_domain_heuristic("http://example.xyz/")
 
     assert DomainHeuristicSignal.NO_HTTPS in result.signals
     assert DomainHeuristicSignal.SUSPICIOUS_TLD in result.signals
-    assert result.score == 50
+    assert result.score == 45
 
 
 @pytest.mark.asyncio
@@ -43,6 +43,28 @@ async def test_rdap_failure_does_not_break_pipeline():
     assert DomainHeuristicSignal.NEW_DOMAIN not in result.signals
 
 
+@pytest.mark.asyncio
+async def test_rdap_timeout_preserves_pattern_signals(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def slow_rdap(_: str):
+        import asyncio
+
+        await asyncio.sleep(1)
+        return None, None
+
+    monkeypatch.setattr("app.services.domain_heuristic.check.lookup_rdap", slow_rdap)
+    monkeypatch.setattr("app.services.domain_heuristic.check.settings.rdap_timeout_seconds", 0.001)
+
+    result = await check_domain_heuristic("https://t-mobile.htufgk.top/pay/")
+
+    assert result.rdap_error == "timeout"
+    assert DomainHeuristicSignal.SUSPICIOUS_TLD in result.signals
+    assert DomainHeuristicSignal.SENSITIVE_PATH in result.signals
+    assert DomainHeuristicSignal.BRAND_IN_URL in result.signals
+    assert result.score >= 61
+
+
 @pytest.mark.asyncio
 async def test_new_domain_signal_from_rdap():
     from datetime import datetime
diff --git a/tests/services/domain_heuristic/test_patterns.py b/tests/services/domain_heuristic/test_patterns.py
index d09b41e..5e70e15 100644
--- a/tests/services/domain_heuristic/test_patterns.py
+++ b/tests/services/domain_heuristic/test_patterns.py
@@ -101,6 +101,11 @@ def test_hosting_platform_github_io() -> None:
     assert DomainHeuristicSignal.HOSTING_PLATFORM in signals
 
 
+def test_private_suffix_user_space_without_subdomain_is_hosting_platform() -> None:
+    signals = check_patterns("https://datrucsmain.wasmer.app/")
+    assert DomainHeuristicSignal.HOSTING_PLATFORM in signals
+
+
 def test_normal_domain_no_hosting_flag() -> None:
     signals = check_patterns("https://naver.com/")
     assert DomainHeuristicSignal.HOSTING_PLATFORM not in signals
@@ -110,3 +115,56 @@ def test_hosting_platform_root_not_flagged() -> None:
     # 플랫폼 자체 홈페이지는 HOSTING_PLATFORM 미발동
     signals = check_patterns("https://netlify.com/")
     assert DomainHeuristicSignal.HOSTING_PLATFORM not in signals
+
+
+def test_userinfo_is_high_risk_url_signal() -> None:
+    signals = check_patterns("https://google.com@evil-login.example.com/session")
+    assert DomainHeuristicSignal.URL_USERINFO in signals
+
+
+def test_brand_in_untrusted_url_is_signal() -> None:
+    signals = check_patterns("https://amazon-clone-kappa-ashen.vercel.app/login")
+    assert DomainHeuristicSignal.BRAND_IN_URL in signals
+    assert DomainHeuristicSignal.FREE_HOSTING_LURE in signals
+
+
+def test_shortener_is_signal() -> None:
+    signals = check_patterns("https://qrco.de/bgoz9a")
+    assert DomainHeuristicSignal.URL_SHORTENER in signals
+
+
+def test_trusted_brand_login_path_not_sensitive_path() -> None:
+    signals = check_patterns("https://naver.com/login")
+    assert DomainHeuristicSignal.SENSITIVE_PATH not in signals
+
+
+def test_generic_brand_labels_do_not_flag_public_normal_sites() -> None:
+    assert DomainHeuristicSignal.BRAND_IN_URL not in check_patterns("https://www.nasa.gov/")
+    assert DomainHeuristicSignal.BRAND_IN_URL not in check_patterns("https://apnews.com/")
+
+
+def test_openphish_like_brand_on_private_hosting_is_lure() -> None:
+    signals = check_patterns("https://www.findmyiphone.vercel.app/")
+    assert DomainHeuristicSignal.HOSTING_PLATFORM in signals
+    assert DomainHeuristicSignal.BRAND_IN_URL in signals
+    assert DomainHeuristicSignal.FREE_HOSTING_LURE in signals
+
+
+def test_sensitive_login_path_on_untrusted_domain_is_signal() -> None:
+    signals = check_patterns("https://www.driegang.nl/new-fr-societegenerale/sg/login/fr/login.php")
+    assert DomainHeuristicSignal.SENSITIVE_PATH in signals
+    assert DomainHeuristicSignal.BRAND_IN_URL in signals
+
+
+def test_tmobile_pay_on_suspicious_tld_is_high_signal() -> None:
+    signals = check_patterns("https://t-mobile.htufgk.top/pay/")
+    assert DomainHeuristicSignal.SUSPICIOUS_TLD in signals
+    assert DomainHeuristicSignal.SENSITIVE_PATH in signals
+    assert DomainHeuristicSignal.BRAND_IN_URL in signals
+
+
+def test_cfd_and_help_are_suspicious_tlds() -> None:
+    assert DomainHeuristicSignal.SUSPICIOUS_TLD in check_patterns("https://dpd.yzqpkmr.cfd/com/")
+    assert DomainHeuristicSignal.SUSPICIOUS_TLD in check_patterns(
+        "https://t-mobile.converselidom.help/pay/"
+    )
diff --git a/tests/services/domain_heuristic/test_typosquatting.py b/tests/services/domain_heuristic/test_typosquatting.py
index 7bebf8f..7499466 100644
--- a/tests/services/domain_heuristic/test_typosquatting.py
+++ b/tests/services/domain_heuristic/test_typosquatting.py
@@ -50,6 +50,12 @@ def test_distance_3_not_flagged(monkeypatch: pytest.MonkeyPatch) -> None:
 def test_same_label_different_tld(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(typosquatting, "_BRANDS", [("naver", "com")])
     signal = check_typosquatting("https://naver.net/")
+    assert signal is None
+
+
+def test_same_label_suspicious_tld_still_flagged(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(typosquatting, "_BRANDS", [("naver", "com")])
+    signal = check_typosquatting("https://naver.xyz/")
     assert signal == DomainHeuristicSignal.TYPO_DOMAIN
 
 
diff --git a/tests/services/test_analysis_callback.py b/tests/services/test_analysis_callback.py
index 0a435f8..acbbe85 100644
--- a/tests/services/test_analysis_callback.py
+++ b/tests/services/test_analysis_callback.py
@@ -6,8 +6,8 @@
 import httpx
 import pytest
 from app.core.config import settings
-from app.schemas.content_analysis import ContentAnalysisResult
-from app.schemas.domain_heuristic import DomainHeuristicResult
+from app.schemas.content_analysis import ContentAnalysisResult, ContentSignal
+from app.schemas.domain_heuristic import DomainHeuristicResult, DomainHeuristicSignal, RdapInfo
 from app.schemas.normalize import NormalizeResult
 from app.schemas.pipeline import (
     PipelineFailure,
@@ -18,8 +18,8 @@
     PipelineTimings,
     Verdict,
 )
-from app.schemas.threat_db import GSBResult, ThreatDbResult, URLhausResult
-from app.schemas.unchain import UnchainResult
+from app.schemas.threat_db import GSBMatch, GSBResult, ThreatDbResult, URLhausResult
+from app.schemas.unchain import HopRecord, UnchainResult
 from app.services.analysis_callback import post_analysis_callback
 
 
@@ -49,8 +49,15 @@ def _success_result() -> PipelineSuccess:
             unchain=UnchainResult(
                 input_url="https://bit.ly/a",
                 final_url=final_url,
-                hops=[],
-                hop_count=0,
+                hops=[
+                    HopRecord(
+                        url="https://bit.ly/a",
+                        status_code=302,
+                        location=final_url,
+                    ),
+                    HopRecord(url=final_url, status_code=200),
+                ],
+                hop_count=2,
             ),
             threat_db=ThreatDbResult(
                 final_url=final_url,
@@ -61,19 +68,46 @@ def _success_result() -> PipelineSuccess:
             ),
             domain_heuristic=DomainHeuristicResult(
                 domain="example.test",
-                score=0,
-                signals=[],
+                score=25,
+                signals=[DomainHeuristicSignal.NEW_DOMAIN],
+                rdap=RdapInfo(
+                    domain="example.test",
+                    registrar="Example Registrar",
+                    created_date=datetime(2026, 4, 1, tzinfo=UTC),
+                    expiry_date=None,
+                    domain_age_days=6,
+                    is_new_domain=True,
+                ),
             ),
             content_analysis=ContentAnalysisResult(
                 final_url=final_url,
                 fetched=True,
-                score=0,
-                signals=[],
+                score=45,
+                signals=[ContentSignal.BRAND_IMPERSONATION_FORM],
+                has_password_field=True,
+                ai_verdict="phishing",
+                ai_reason="브랜드 사칭 로그인 폼이 확인되었습니다.",
             ),
         ),
     )
 
 
+def _malicious_result() -> PipelineSuccess:
+    result = _success_result()
+    result.stages.threat_db.is_malicious = True
+    result.stages.threat_db.gsb = GSBResult(
+        checked=True,
+        is_threat=True,
+        matches=[GSBMatch(threat_type="SOCIAL_ENGINEERING")],
+    )
+    result.stages.threat_db.urlhaus = URLhausResult(
+        checked=True,
+        is_threat=True,
+        matched_key="example.test",
+    )
+    return result
+
+
 @pytest.mark.asyncio
 async def test_posts_success_callback_payload(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(settings, "spring_internal_url", "http://spring.internal")
@@ -110,17 +144,93 @@ async def test_posts_success_callback_payload(monkeypatch: pytest.MonkeyPatch) -
     assert payload["engineVersion"] == settings.app_version
     assert payload["analyzedAt"] == "2026-04-07T05:42:11Z"
     assert payload["elapsedMs"] == 123
-    assert payload["timings"] == {
-        "total_seconds": 0.123,
-        "stages": {
-            "normalize": 0.001,
-            "unchain": 0.002,
-            "threat_db": 0.003,
-            "domain_heuristic": 0.004,
-            "content_analysis": 0.005,
+    assert payload["summary"] == "브랜드 사칭 로그인 폼이 확인되었습니다."
+    assert payload["reasons"] == [
+        {
+            "code": "NEW_DOMAIN",
+            "stage": 3,
+            "weight": settings.score_weight_new_domain,
+            "message": "최근 등록된 도메인입니다.",
+        },
+        {
+            "code": "BRAND_IMPERSONATION_FORM",
+            "stage": 4,
+            "weight": settings.score_weight_brand_impersonation,
+            "message": "브랜드를 사칭하는 로그인 폼이 확인되었습니다.",
+        },
+        {
+            "code": "AI_PHISHING",
+            "stage": 4,
+            "weight": settings.score_weight_ai_phishing,
+            "message": "브랜드 사칭 로그인 폼이 확인되었습니다.",
+        },
+    ]
+    assert payload["stages"] == {
+        "externalDb": {
+            "gsb": {"isThreat": False, "matchedTypes": []},
+            "urlhaus": {"isThreat": False, "host": "example.test"},
+        },
+        "unchain": {
+            "hops": 2,
+            "chain": ["https://bit.ly/a", "https://example.test/"],
+        },
+        "domainHeuristic": {
+            "rdap": {
+                "domain": "example.test",
+                "registrar": "Example Registrar",
+                "createdDate": "2026-04-01T00:00:00Z",
+                "domainAgeDays": 6,
+                "isNewDomain": True,
+            },
+            "signals": ["NEW_DOMAIN"],
+        },
+        "contentAnalysis": {
+            "fetched": True,
+            "hasPasswordField": True,
+            "aiVerdict": "phishing",
+            "aiReason": "브랜드 사칭 로그인 폼이 확인되었습니다.",
+        },
+    }
+
+
+@pytest.mark.asyncio
+async def test_posts_known_malicious_callback_with_fixed_summary(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(settings, "spring_internal_url", "http://spring.internal")
+    mock_client = AsyncMock()
+    mock_client.__aenter__.return_value = mock_client
+    mock_client.post.return_value = httpx.Response(200, json={"received": True})
+
+    with patch("app.services.analysis_callback.httpx.AsyncClient", return_value=mock_client):
+        delivered = await post_analysis_callback(
+            _malicious_result(),
+            request_id="rid-1",
+            elapsed_ms=123,
+            analyzed_at=datetime(2026, 4, 7, 5, 42, 11, tzinfo=UTC),
+        )
+
+    assert delivered is True
+    payload = mock_client.post.await_args.kwargs["json"]
+    assert payload["summary"] == "악성으로 알려진 페이지 입니다."
+    assert payload["reasons"][:2] == [
+        {
+            "code": "GSB_MATCH",
+            "stage": 2,
+            "weight": settings.score_weight_gsb,
+            "message": "Google Safe Browsing에 악성 URL로 등록되어 있습니다.",
         },
+        {
+            "code": "URLHAUS_MATCH",
+            "stage": 2,
+            "weight": settings.score_weight_urlhaus,
+            "message": "URLhaus에 악성 URL로 등록되어 있습니다.",
+        },
+    ]
+    assert payload["stages"]["externalDb"] == {
+        "gsb": {"isThreat": True, "matchedTypes": ["SOCIAL_ENGINEERING"]},
+        "urlhaus": {"isThreat": True, "host": "example.test"},
     }
-    assert payload["stages"]["threat_db"]["is_malicious"] is False
 
 
 @pytest.mark.asyncio
@@ -182,9 +292,8 @@ async def test_posts_failure_callback_payload(monkeypatch: pytest.MonkeyPatch) -
     assert payload["originalUrl"] == "not a url"
     assert payload["error"] == {
         "code": "NORMALIZE_FAILED",
-        "stage": "normalize",
+        "stage": 1,
         "message": "invalid url",
     }
-    assert payload["timings"]["total_seconds"] == 0.001
-    assert payload["timings"]["stages"]["normalize"] == 0.001
+    assert "timings" not in payload
     assert "stages" not in payload
diff --git a/tests/services/test_db_independent_pipeline.py b/tests/services/test_db_independent_pipeline.py
index 6f2028b..9d34db3 100644
--- a/tests/services/test_db_independent_pipeline.py
+++ b/tests/services/test_db_independent_pipeline.py
@@ -7,6 +7,7 @@
 from unittest.mock import AsyncMock, patch
 
 import pytest
+from app.core.config import settings
 from app.schemas.content_analysis import ContentAnalysisResult
 from app.schemas.db_independent_pipeline import (
     DbIndependentPipelineFailure,
@@ -163,7 +164,7 @@ async def _slow_content(_: str, **__: object) -> ContentAnalysisResult:
 
 
 @pytest.mark.asyncio
-async def test_db_independent_pipeline_does_not_start_content_when_heuristic_fails() -> None:
+async def test_db_independent_pipeline_degrades_when_heuristic_fails() -> None:
     final_url = "https://error.example.com/login"
 
     async def _failing_heuristic(_: str) -> DomainHeuristicResult:
@@ -186,11 +187,50 @@ async def _failing_heuristic(_: str) -> DomainHeuristicResult:
     ):
         mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url)
         mock_unchain.return_value = _make_unchain(final_url)
+        mock_content.return_value = _make_content(final_url)
 
-        with pytest.raises(RuntimeError, match="heuristic failed"):
-            await run_db_independent_pipeline("aid-error-cleanup", final_url)
+        result = await run_db_independent_pipeline("aid-error-cleanup", final_url)
 
-    mock_content.assert_not_awaited()
+    assert isinstance(result, DbIndependentPipelineSuccess)
+    assert result.stages.domain_heuristic.rdap_error == "pipeline_timeout"
+    mock_content.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_db_independent_pipeline_caps_content_stage_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    final_url = "https://slow-content.example.com/login"
+    monkeypatch.setattr(settings, "pipeline_content_timeout_seconds", 0.001)
+
+    async def _slow_content(_: str, **__: object) -> ContentAnalysisResult:
+        await asyncio.sleep(1)
+        return _make_content(final_url, score=90)
+
+    with (
+        patch("app.services.db_independent_pipeline.normalize_url") as mock_norm,
+        patch(
+            "app.services.db_independent_pipeline.unchain_url", new_callable=AsyncMock
+        ) as mock_unchain,
+        patch(
+            "app.services.db_independent_pipeline.check_domain_heuristic",
+            new_callable=AsyncMock,
+        ) as mock_heuristic,
+        patch(
+            "app.services.db_independent_pipeline.analyze_content",
+            new=AsyncMock(side_effect=_slow_content),
+        ),
+    ):
+        mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url)
+        mock_unchain.return_value = _make_unchain(final_url)
+        mock_heuristic.return_value = _make_heuristic(0)
+
+        result = await run_db_independent_pipeline("aid-content-timeout", final_url)
+
+    assert isinstance(result, DbIndependentPipelineSuccess)
+    assert result.stages.content_analysis.error == "pipeline_timeout"
+    assert result.timings is not None
+    assert result.timings.total_seconds < 0.5
 
 
 @pytest.mark.asyncio
diff --git a/tests/services/test_evaluation_metrics.py b/tests/services/test_evaluation_metrics.py
new file mode 100644
index 0000000..eca60f3
--- /dev/null
+++ b/tests/services/test_evaluation_metrics.py
@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from app.schemas.pipeline import Verdict
+from app.services.evaluation.metrics import VerdictCase, compute_verdict_metrics
+
+
+def test_compute_verdict_metrics_for_detection_goal() -> None:
+    cases = [
+        VerdictCase(expected=Verdict.DANGER, predicted=Verdict.DANGER),
+        VerdictCase(expected=Verdict.DANGER, predicted=Verdict.DANGER),
+        VerdictCase(expected=Verdict.DANGER, predicted=Verdict.CAUTION),
+        VerdictCase(expected=Verdict.DANGER, predicted=Verdict.SAFE),
+        VerdictCase(expected=Verdict.SAFE, predicted=Verdict.SAFE),
+        VerdictCase(expected=Verdict.SAFE, predicted=Verdict.SAFE),
+        VerdictCase(expected=Verdict.SAFE, predicted=Verdict.CAUTION),
+        VerdictCase(expected=Verdict.SAFE, predicted=Verdict.DANGER),
+    ]
+
+    metrics = compute_verdict_metrics(
+        cases,
+        positive_expected={Verdict.DANGER},
+        positive_predicted={Verdict.DANGER, Verdict.CAUTION},
+    )
+
+    assert metrics.total == 8
+    assert metrics.accuracy == 0.5
+    assert metrics.positive_recall == 0.75
+    assert metrics.positive_precision == 0.6
+    assert metrics.positive_f1 == 0.666667
+    assert metrics.false_positive_rate == 0.5
diff --git a/tests/services/test_pipeline.py b/tests/services/test_pipeline.py
index 225ca19..c8abe95 100644
--- a/tests/services/test_pipeline.py
+++ b/tests/services/test_pipeline.py
@@ -623,7 +623,7 @@ async def test_run_pipeline_skips_content_when_heuristic_alone_exceeds_threshold
 
 
 @pytest.mark.asyncio
-async def test_run_pipeline_does_not_start_content_when_stage_2_3_fails(
+async def test_run_pipeline_degrades_reputation_error_and_continues_to_content(
     async_session: AsyncSession,
 ) -> None:
     final_url = "https://error.test/"
@@ -644,8 +644,40 @@ async def failing_threat(_session: AsyncSession, _url: str) -> ThreatDbResult:
     ):
         mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url)
         mock_unchain.return_value = _make_unchain(final_url)
+        mock_content.return_value = _make_content(final_url)
 
-        with pytest.raises(RuntimeError, match="threat db failed"):
-            await run_pipeline("aid-error-cleanup", final_url, async_session)
+        result = await run_pipeline("aid-error-cleanup", final_url, async_session)
 
-    mock_content.assert_not_awaited()
+    assert isinstance(result, PipelineSuccess)
+    assert result.stages.threat_db.gsb.error == "pipeline_timeout"
+    assert result.stages.domain_heuristic.rdap_error == "pipeline_timeout"
+    mock_content.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_run_pipeline_degrades_content_error(
+    async_session: AsyncSession,
+) -> None:
+    final_url = "https://content-error.test/"
+
+    async def failing_content(_url: str, **_kwargs: object) -> ContentAnalysisResult:
+        raise RuntimeError("content failed")
+
+    with (
+        patch("app.services.pipeline.normalize_url") as mock_norm,
+        patch("app.services.pipeline.unchain_url", new_callable=AsyncMock) as mock_unchain,
+        patch("app.services.pipeline.check_threat_db", new_callable=AsyncMock) as mock_threat,
+        patch(
+            "app.services.pipeline.check_domain_heuristic", new_callable=AsyncMock
+        ) as mock_heuristic,
+        patch("app.services.pipeline.analyze_content", side_effect=failing_content),
+    ):
+        mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url)
+        mock_unchain.return_value = _make_unchain(final_url)
+        mock_threat.return_value = _make_threat(final_url)
+        mock_heuristic.return_value = _heuristic_with_score(0)
+
+        result = await run_pipeline("aid-content-error", final_url, async_session)
+
+    assert isinstance(result, PipelineSuccess)
+    assert result.stages.content_analysis.error == "pipeline_timeout"
diff --git a/tests/test_makefile.py b/tests/test_makefile.py
index 24147f5..3d24ac6 100644
--- a/tests/test_makefile.py
+++ b/tests/test_makefile.py
@@ -1,11 +1,18 @@
 from pathlib import Path
 
 
-def test_run_target_reloads_app_only() -> None:
+def test_run_target_does_not_use_reload_watcher() -> None:
     makefile = Path("Makefile").read_text()
-    run_block = makefile.split("run:  ## Run the API locally with hot reload", 1)[1].split(
+    run_block = makefile.split("run:  ## Run the API locally", 1)[1].split("\n\n", 1)[0]
+
+    assert "--reload" not in run_block
+
+
+def test_dev_target_reloads_app_only() -> None:
+    makefile = Path("Makefile").read_text()
+    dev_block = makefile.split("dev:  ## Run the API locally with hot reload", 1)[1].split(
         "\n\n", 1
     )[0]
 
-    assert "--reload-dir app" in run_block
-    assert "--reload-exclude .venv" in run_block
+    assert "--reload-dir app" in dev_block
+    assert "--reload-exclude .venv" in dev_block