"""Minimal MCP discovery UI for PoC4.

Run:
    streamlit run apps/poc4/mcp_discovery_ui.py --server.port 8622

This screen intentionally does not use the existing LangGraph/actor-context
pipeline.  It discovers MCP tools, selects a query-capable tool, and calls it
with the Bearer token entered on the screen.
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import html
import json
import logging
import os
import sqlite3
import sys
from datetime import datetime, timezone
from dataclasses import dataclass, field
from pathlib import Path
import re
from time import perf_counter
from typing import Any, Mapping
from uuid import uuid4
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener

ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import streamlit as st
import streamlit.components.v1 as components
import oracledb

from src.mcp_tool_router import (
    McpTool,
    McpToolRouterError,
    RoutedMcpTool,
    build_mcp_tool_arguments,
    route_mcp_tool_across_servers_with_llm,
)
from src.oci_genai_sdk import (
    build_oci_genai_completion_client,
    temperature_for_model_profile,
)
from src.poc3.model_registry import load_model_registry, resolve_model_profile
from src.agent_console.presentation import (
    apply_console_theme,
    render_console_header,
    render_login_brand,
)
from src.agent_console.profile import AppProfile, AppProfileError, load_app_profile
from src.poc4.qa_history import QaJudgment, QaQuestion, evaluate_sql, load_benchmark_questions
from src.poc4.qa_history_store import QaHistoryStore, QaHistoryStoreError


LOG = logging.getLogger(__name__)
MCP_PROTOCOL_VERSION = "2025-11-25"
PREFERRED_TOOL = "oracle.select_ai.smilegate_fewshot_nl2sql"
DEFAULT_QUESTION = ""
MAX_RESPONSE_BYTES = 1_000_000
MAX_CONVERSATION_MESSAGES = 8
CHAT_TURNS_PER_PAGE = 3
CHAT_CONTEXT_TURNS = 4
MAX_AGENT_TOOL_STEPS = 4
AGENT_FINAL_ROUTE = "__final__"
COMPLEX_REASONING_MODEL_PROFILE_ENV = "POC4_COMPLEX_REASONING_MODEL_PROFILE"
DEFAULT_COMPLEX_REASONING_MODEL_PROFILE = "grok43"
DEFAULT_SYNTHESIS_FALLBACK_MODEL_PROFILE = "gpt54_mini_oci"
DEFAULT_QUERY_MODEL_PROFILE = "gpt54_mini_oci"
ENV_FILE = ROOT / ".env"
MCP_SERVERS_FILE = ROOT / "config" / "mcp_servers.json"
VPD_TOKEN_PRESETS_FILE = ROOT / "config" / "vpd_token_presets.json"
QA_BENCHMARK_FILE = ROOT / "config" / "smilegate_qa_benchmark.json"
APP_PROFILE_FILE = ROOT / "config" / "app_profile.json"
CHAT_DB_FILE = ROOT / "data" / "poc4_mcp_chat.sqlite3"
DEFAULT_VPD_USER_ID = "1001"
VPD_OPERATIONS_URL = "https://smilegate-backoffice.cloud-handson.com/"
PORTAL_AUTHENTICATED_KEY = "poc4_portal_authenticated"
PORTAL_AUTH_USER_KEY = "poc4_portal_auth_user"
PORTAL_LOGIN_FAILURE_KEY = "poc4_portal_login_failed"
PORTAL_AUTH_COOKIE_NAME = "poc4_portal_auth"
AUDIT_SCHEMA = "SGMP_POC"
REFERENCE_EVIDENCE_ENABLED = (
    os.environ.get("POC4_REFERENCE_EVIDENCE_ENABLED", "").strip().lower()
    in {"1", "true", "yes"}
)
AUDIT_DB_ENV_FILE = Path(
    os.environ.get("POC4_AUDIT_DB_ENV_FILE", "/home/opc/kbmcp/.env")
).expanduser()
_OPAQUE_BEARER = re.compile(r"^[\x21-\x7e]{1,4096}$")
KB_THEME_CSS = """
<style>
:root {
  --kb-yellow: #004b87;
  --kb-yellow-soft: #dceef8;
  --kb-yellow-pale: #f2f8fc;
  --kb-ink: #0c2e4b;
  --kb-muted: #526776;
  --kb-line: #cbdbe6;
  --kb-card: #ffffff;
  --kb-bg: #f3f7fa;
  --kb-blue: #0077b6;
  --kb-green: #008a74;
  --kb-red: #d73a49;
}

html,
body,
[data-testid="stAppViewContainer"],
.stApp {
  color-scheme: light !important;
}

header[data-testid="stHeader"],
[data-testid="stHeader"],
.stAppHeader,
[data-testid="stToolbar"] {
  background: #ffffff !important;
  background-color: #ffffff !important;
  color: #111111 !important;
}

header[data-testid="stHeader"] button,
header[data-testid="stHeader"] button span,
header[data-testid="stHeader"] svg,
[data-testid="stToolbar"] button,
[data-testid="stToolbar"] button span,
[data-testid="stToolbar"] svg {
  color: #111111 !important;
  fill: currentColor !important;
  stroke: currentColor !important;
}

#MainMenu,
[data-testid="stMainMenu"],
header[data-testid="stHeader"] button[aria-label="Main menu"] {
  display: none !important;
  visibility: hidden !important;
}

.stApp {
  background: linear-gradient(180deg, #ffffff 0%, var(--kb-bg) 100%);
  color: var(--kb-ink);
}

section[data-testid="stSidebar"] {
  background: #ffffff;
  border-right: 1px solid #ececec;
}

section[data-testid="stSidebar"] [data-testid="stSidebarContent"] {
  padding-top: 1rem !important;
}

@media (min-width: 761px) {
  section[data-testid="stSidebar"],
  section[data-testid="stSidebar"] > div {
    width: 280px !important;
    min-width: 280px !important;
  }

  section[data-testid="stSidebar"] [data-testid="stSidebarUserContent"] {
    margin-top: -1.75rem !important;
  }
}

section[data-testid="stSidebar"] [data-testid="stMarkdownContainer"] p,
section[data-testid="stSidebar"] label {
  color: var(--kb-muted);
}

section[data-testid="stSidebar"] label,
section[data-testid="stSidebar"] [data-testid="stWidgetLabel"] p,
section[data-testid="stSidebar"] .kb-sidebar-section-title,
section[data-testid="stSidebar"] summary {
  color: var(--kb-ink) !important;
  font-size: 1rem !important;
  font-weight: 850 !important;
  line-height: 1.35 !important;
}

section[data-testid="stSidebar"] [data-testid="stPopover"] button,
section[data-testid="stSidebar"] [data-testid="stPopover"] button p,
section[data-testid="stSidebar"] [data-testid="stPopover"] button strong {
  color: var(--kb-ink) !important;
  font-weight: 900 !important;
}

[data-baseweb="popover"]:has([role="radiogroup"]) {
  max-height: min(280px, calc(100vh - 120px)) !important;
  overflow-y: auto !important;
  overscroll-behavior: contain;
  transform-origin: top left !important;
}

[data-baseweb="popover"] [role="radiogroup"] {
  flex-direction: column !important;
  flex-wrap: nowrap !important;
  max-height: 232px !important;
  overflow-y: auto !important;
  overscroll-behavior: contain;
  padding-right: 6px;
  scrollbar-gutter: stable;
}

section[data-testid="stSidebar"] [data-testid="stMarkdownContainer"] p strong {
  color: var(--kb-ink);
  font-weight: 850;
}

section[data-testid="stSidebar"] input:disabled {
  background: #ffffff !important;
  color: #111111 !important;
  -webkit-text-fill-color: #111111 !important;
  opacity: 1 !important;
}

.block-container {
  padding-top: 1.45rem;
  max-width: 1120px;
}

.st-key-kb_login_container {
  width: min(100%, 520px);
  margin: clamp(2.5rem, 9vh, 6.5rem) auto 2.5rem;
}

.kb-login-brand {
  margin-bottom: 1.5rem;
  text-align: center;
}

.kb-login-logo {
  display: block;
  width: min(74%, 290px);
  height: auto;
  margin: 0 auto 1.35rem;
  object-fit: contain;
}

.kb-login-kicker {
  margin-bottom: 0.5rem;
  color: #6d675e;
  font-size: 0.86rem;
  font-weight: 800;
}

.kb-login-brand h1 {
  margin: 0;
  color: var(--kb-ink);
  font-family: "Noto Sans KR", "Malgun Gothic", sans-serif;
  font-size: 2.25rem;
  font-weight: 600;
  line-height: 1.2;
  letter-spacing: 0;
  word-break: keep-all;
}

.kb-login-brand p {
  margin: 0.85rem auto 0;
  color: var(--kb-muted);
  font-size: 0.96rem;
  line-height: 1.6;
  word-break: keep-all;
}

.st-key-kb_login_container [data-testid="stForm"] {
  padding: 1.65rem 1.7rem 1.5rem;
  border: 1px solid var(--kb-line);
  border-top: 6px solid var(--kb-yellow);
  border-radius: 8px;
  background: #ffffff;
  box-shadow: 0 16px 34px rgba(17, 17, 17, 0.1);
}

.st-key-kb_login_container [data-testid="stForm"] [data-testid="stWidgetLabel"] {
  width: 100%;
  margin-bottom: 0.45rem;
  padding: 0.55rem 0.7rem;
  border-left: 5px solid var(--kb-yellow);
  border-radius: 4px;
  background: #fff1a8;
}

.st-key-kb_login_container [data-testid="stForm"] [data-testid="stWidgetLabel"] p {
  color: var(--kb-ink) !important;
  font-weight: 850 !important;
}

.st-key-kb_login_container [data-testid="stTextInput"] input {
  min-height: 48px;
  background: #ffffff !important;
  color: #111111 !important;
  -webkit-text-fill-color: #111111 !important;
}

.st-key-kb_login_container [data-testid="stFormSubmitButton"] button {
  min-height: 50px;
  border: 1px solid #e5b700 !important;
  border-radius: 6px !important;
  background: var(--kb-yellow) !important;
  color: var(--kb-ink) !important;
  font-family: "Noto Sans KR", "Malgun Gothic", sans-serif;
  font-size: 1.08rem !important;
  font-weight: 600 !important;
  box-shadow: 0 3px 0 #111111;
}

.st-key-kb_login_container [data-testid="stFormSubmitButton"] button:hover {
  background: #ffd633 !important;
}

.kb-login-note {
  margin-top: 1.15rem;
  color: #777169;
  font-size: 0.78rem;
  line-height: 1.55;
  text-align: center;
}

.kb-hero {
  display: grid;
  grid-template-columns: minmax(0, 1fr) 236px;
  grid-template-areas: "main logo";
  gap: 24px;
  align-items: start;
  border: 1px solid var(--kb-line);
  border-radius: 8px;
  padding: 28px;
  margin-bottom: 24px;
  background: #ffffff;
  box-shadow: none;
}

.kb-hero-main {
  position: relative;
  z-index: 1;
  grid-area: main;
  min-width: 0;
}

.kb-brand-row {
  display: flex;
  align-items: center;
  gap: 12px;
  margin-bottom: 18px;
}

.kb-mini-mark {
  display: inline-flex;
  align-items: center;
  justify-content: center;
  width: 42px;
  height: 42px;
  border-radius: 13px;
  background: var(--kb-yellow);
  color: #ffffff;
  font-weight: 900;
  letter-spacing: 0;
  box-shadow: inset 0 -2px 0 rgba(255, 255, 255, 0.12);
}

.kb-brand-name {
  font-size: 0.92rem;
  color: var(--kb-ink);
  font-weight: 800;
}

.kb-brand-sub {
  font-size: 0.78rem;
  color: var(--kb-muted);
}

.kb-hero h1 {
  margin: 0;
  font-family: "Noto Sans KR", "Malgun Gothic", sans-serif;
  font-size: 2.35rem;
  font-weight: 700;
  -webkit-text-stroke: 0.2px currentColor;
  line-height: 1.16;
  letter-spacing: 0;
  color: var(--kb-ink);
  word-break: keep-all;
}

.kb-hero p {
  margin: 12px 0 0;
  max-width: 680px;
  color: var(--kb-muted);
  font-size: 0.98rem;
  line-height: 1.58;
}

.kb-ci-panel {
  position: relative;
  z-index: 1;
  grid-area: logo;
  display: flex;
  align-items: flex-start;
  justify-content: flex-end;
  width: 100%;
  align-self: start;
  justify-self: end;
}

.kb-ci-panel img {
  display: block;
  width: min(100%, 228px);
  height: auto;
  object-fit: contain;
}

.kb-panel-title {
  margin: 0 0 0.85rem;
  padding: 0.72rem 0.8rem;
  border-radius: 4px;
  background: var(--kb-yellow);
  border-bottom: 3px solid var(--kb-ink);
  color: var(--kb-ink);
  font-size: 1.18rem;
  font-weight: 900;
  line-height: 1.2;
  letter-spacing: 0;
  text-align: center;
}

section[data-testid="stSidebar"] .kb-panel-title {
  font-weight: 900 !important;
}

.kb-section-title {
  margin: 0.75rem 0 0.85rem;
  padding: 0.65rem 0.85rem;
  border-left: 6px solid;
  border-radius: 4px;
  font-family: "Noto Sans KR", "Malgun Gothic", sans-serif;
  font-size: 1.75rem;
  font-weight: 600;
  line-height: 1.2;
  letter-spacing: 0;
}

.kb-section-title strong {
  font-family: inherit;
  font-weight: inherit;
}

.kb-section-title.input {
  border-left-color: var(--kb-yellow);
  background: var(--kb-yellow-soft);
  color: var(--kb-ink);
}

.kb-section-title.result {
  border-left-color: #4473c9;
  background: #eef3ff;
  color: #173b79;
}

.kb-sidebar-card {
  border: 1px solid var(--kb-line);
  border-radius: 20px;
  padding: 14px;
  margin: 8px 0 14px;
  background: #ffffff;
}

.kb-vpd-card {
  border: 1px solid rgba(17, 17, 17, 0.08);
  border-radius: 20px;
  padding: 14px;
  margin: 10px 0 12px;
  background: #111111;
  color: #fff8cf;
}

.kb-vpd-card .kb-vpd-user {
  font-weight: 900;
  color: #ffcc00;
  margin-bottom: 4px;
}

.kb-vpd-card .kb-vpd-meta {
  font-size: 0.82rem;
  color: #f2e6a5;
}

.kb-vpd-card .kb-vpd-scope {
  margin-top: 8px;
  font-size: 0.82rem;
  color: #ffffff;
}

div[data-testid="stButton"] > button {
  border: 0;
  border-radius: 999px;
  background: #111111;
  color: var(--kb-yellow);
  font-weight: 850;
  box-shadow: 0 8px 18px rgba(17, 17, 17, 0.14);
}

div[data-testid="stButton"] > button:hover {
  background: #333333;
  color: #ffe36b;
  border: 0;
}

div[data-testid="stButton"] > button[kind="primary"] {
  background: #111111;
  color: var(--kb-yellow);
}

div[data-testid="stForm"] {
  margin: 0.75rem 0 1rem;
  padding: 1rem 1rem 0.9rem;
  border: 2px solid var(--kb-yellow);
  border-radius: 8px;
  background: #fff8d6;
  box-shadow: 0 8px 18px rgba(17, 17, 17, 0.08);
}

div[data-testid="stForm"] div[data-testid="stWidgetLabel"] p {
  color: var(--kb-ink);
  font-weight: 850;
}

div[data-testid="stFormSubmitButton"] {
  margin-top: 0.35rem;
}

div[data-testid="stFormSubmitButton"]
  > button[data-testid="stBaseButton-primaryFormSubmit"] {
  width: 100%;
  min-height: 58px;
  border: 2px solid #111111;
  border-radius: 8px;
  background: #111111;
  color: var(--kb-yellow);
  font-size: 1.06rem;
  font-weight: 900;
  letter-spacing: 0;
  box-shadow: 0 10px 22px rgba(17, 17, 17, 0.18);
  transition: background 160ms ease, color 160ms ease,
    box-shadow 160ms ease, transform 160ms ease;
}

div[data-testid="stFormSubmitButton"]
  > button[data-testid="stBaseButton-primaryFormSubmit"] p {
  color: inherit !important;
  font-size: inherit !important;
  font-weight: inherit !important;
}

div[data-testid="stFormSubmitButton"]
  > button[data-testid="stBaseButton-primaryFormSubmit"]:hover {
  border-color: #111111;
  background: var(--kb-yellow);
  color: #111111;
  box-shadow: 0 12px 26px rgba(17, 17, 17, 0.22);
  transform: translateY(-1px);
}

div[data-testid="stFormSubmitButton"]
  > button[data-testid="stBaseButton-primaryFormSubmit"]:active {
  box-shadow: 0 6px 14px rgba(17, 17, 17, 0.18);
  transform: translateY(0);
}

div[data-testid="stFormSubmitButton"]
  > button[data-testid="stBaseButton-primaryFormSubmit"]:focus-visible {
  outline: 3px solid rgba(68, 115, 201, 0.38);
  outline-offset: 3px;
}

div[data-testid="stTextArea"] textarea,
div[data-testid="stTextInput"] input,
div[data-baseweb="select"] > div,
div[data-testid="stNumberInput"] input {
  border-radius: 14px;
  border-color: var(--kb-line);
  background: #fff;
}

.st-key-poc4_mcp_discovery_question_text textarea {
  font-size: 1.12rem !important;
  line-height: 1.65 !important;
}

div[data-testid="stChatMessage"] {
  border-radius: 20px;
  border: 1px solid var(--kb-line);
  background: #ffffff;
  box-shadow: 0 10px 22px rgba(17, 17, 17, 0.05);
}

details {
  border-radius: 16px !important;
  border: 1px solid var(--kb-line) !important;
  background: #ffffff !important;
}

[data-testid="stAlert"] {
  border-radius: 18px;
}

div[data-baseweb="tab-list"] {
  gap: 10px;
  border-bottom: 2px solid var(--kb-line);
}

button[data-baseweb="tab"] {
  min-width: 128px;
  min-height: 54px;
  padding: 12px 20px;
  color: #2452a3 !important;
  background: #eef3ff;
  border-radius: 6px 6px 0 0;
  font-family: "Noto Sans KR", "Malgun Gothic", sans-serif !important;
  font-size: 1.2rem !important;
  font-weight: 700 !important;
  letter-spacing: 0;
}

button[data-baseweb="tab"] * {
  color: inherit !important;
  font-family: inherit !important;
  font-size: inherit !important;
  font-weight: inherit !important;
}

button[data-baseweb="tab"]:hover {
  color: #173b79 !important;
  background: #dfe9fb;
}

button[data-baseweb="tab"][aria-selected="true"] {
  color: var(--kb-ink) !important;
  background: var(--kb-yellow);
  border-bottom: 4px solid var(--kb-ink);
}

div[data-testid="stMetric"] {
  min-height: 96px;
  padding: 12px 14px;
  border: 1px solid var(--kb-line);
  border-radius: 8px;
  background: #ffffff;
}

.kb-audit-lead {
  margin: -0.1rem 0 1rem;
  color: #4f5663;
  font-size: 0.98rem;
  line-height: 1.6;
}

.kb-audit-heading {
  display: flex;
  align-items: center;
  gap: 10px;
  margin: 1.05rem 0 0.65rem;
  color: #111111;
  font-family: "Noto Sans KR", "Malgun Gothic", sans-serif;
  font-size: 1.18rem;
  font-weight: 700;
  line-height: 1.3;
}

.kb-audit-heading::before {
  content: "";
  width: 5px;
  height: 22px;
  border-radius: 2px;
  background: var(--kb-yellow);
}

.kb-audit-caption {
  margin: -0.25rem 0 0.75rem;
  color: #667085;
  font-size: 0.84rem;
  line-height: 1.55;
}

.st-key-poc4_fga_filters {
  padding: 14px 16px 8px;
  border: 1px solid #d9d9d9;
  border-top: 4px solid var(--kb-yellow);
  border-radius: 8px;
  background: #fffdf3;
  box-shadow: 0 6px 16px rgba(17, 17, 17, 0.05);
}

.st-key-poc4_fga_filters label p,
.st-key-poc4_fga_filters [data-testid="stWidgetLabel"] p {
  color: #111111 !important;
  font-size: 0.9rem !important;
  font-weight: 850 !important;
}

.st-key-poc4_fga_filters div[data-baseweb="select"] > div,
.st-key-poc4_fga_filters div[data-testid="stNumberInput"] input {
  min-height: 46px;
  border-color: #c7cbd1;
  border-radius: 6px;
  color: #111111;
  background: #ffffff;
}

.st-key-poc4_fga_filters div[data-testid="stButton"] > button {
  min-height: 46px;
  border-radius: 6px;
}

.st-key-poc4_fga_metrics {
  margin: 0.9rem 0 1rem;
}

.st-key-poc4_fga_metrics div[data-testid="stMetric"] {
  min-height: 108px;
  border: 1px solid #cfd4dc;
  border-top: 5px solid #4473c9;
  background: #ffffff;
  box-shadow: 0 7px 18px rgba(17, 17, 17, 0.06);
}

.st-key-poc4_fga_metrics div[data-testid="column"]:nth-child(3)
  div[data-testid="stMetric"] {
  border-top-color: #3f845b;
  background: #f4fbf6;
}

.st-key-poc4_fga_metrics div[data-testid="column"]:nth-child(4)
  div[data-testid="stMetric"] {
  border-top-color: #c62828;
  background: #fff6f6;
}

.st-key-poc4_fga_metrics [data-testid="stMetricLabel"] p {
  color: #4f5663 !important;
  font-size: 0.92rem !important;
  font-weight: 800 !important;
}

.st-key-poc4_fga_metrics [data-testid="stMetricValue"] {
  color: #111111 !important;
  font-size: 1.8rem !important;
  font-weight: 900 !important;
}

.st-key-poc4_fga_policy_panel details,
.st-key-poc4_fga_event_panel details {
  border-radius: 8px !important;
  border-color: #cfd4dc !important;
  box-shadow: 0 5px 14px rgba(17, 17, 17, 0.04);
}

.st-key-poc4_fga_policy_panel summary,
.st-key-poc4_fga_event_panel summary {
  color: #111111 !important;
  font-size: 1rem !important;
  font-weight: 850 !important;
}

.st-key-poc4_fga_policy_panel [data-testid="stDataFrame"],
.st-key-poc4_fga_event_panel [data-testid="stDataFrame"] {
  overflow: hidden;
  border: 1px solid #cfd4dc;
  border-radius: 6px;
  background: #ffffff;
  box-shadow: 0 8px 20px rgba(17, 17, 17, 0.05);
}

.st-key-poc4_fga_event_toolbar {
  padding: 9px 14px 3px;
  border: 1px solid #d9d9d9;
  border-radius: 8px;
  background: #f4f7fb;
}

.st-key-poc4_fga_event_toolbar label p {
  color: #111111 !important;
  font-size: 0.92rem !important;
  font-weight: 800 !important;
}

.kb-arch-lead {
  max-width: 860px;
  margin: 10px 0 20px;
  color: var(--kb-muted);
  font-size: 0.96rem;
  line-height: 1.6;
}

.kb-arch-map {
  display: grid;
  grid-template-columns:
    minmax(0, 2.1fr) 32px minmax(0, 1.2fr) 32px minmax(0, 1.2fr);
  align-items: stretch;
  gap: 0;
}

.kb-arch-stage {
  min-width: 0;
  min-height: 178px;
  padding: 16px;
  border: 1px solid var(--kb-line);
  border-radius: 8px;
  background: #ffffff;
}

.kb-arch-stage.yellow {
  border-color: #e6cf72;
  background: #fff8d7;
}

.kb-arch-stage.experience {
  border-color: #e2c959;
  border-left: 6px solid var(--kb-yellow);
  background: #fffbea;
}

.kb-arch-experience-grid {
  display: grid;
  grid-template-columns: repeat(3, minmax(0, 1fr));
  gap: 10px;
  margin-top: 12px;
}

.kb-arch-experience-item {
  min-width: 0;
  padding: 10px;
  border: 1px solid rgba(17, 17, 17, 0.12);
  border-radius: 6px;
  background: #ffffff;
}

.kb-arch-experience-item strong {
  font-size: 0.82rem;
}

.kb-arch-experience-item span {
  display: block;
  margin-top: 6px;
  color: var(--kb-muted);
  font-size: 0.7rem;
  line-height: 1.45;
  overflow-wrap: anywhere;
}

.kb-arch-stage.blue {
  border-color: #aec3e9;
  background: #edf3ff;
}

.kb-arch-stage.green {
  border-color: #add0b9;
  background: #edf7f1;
}

.kb-arch-stage.security {
  border-color: #d6b6aa;
  background: #fff2ed;
}

.kb-arch-stage-label {
  display: block;
  margin-bottom: 10px;
  color: #6b625b;
  font-size: 0.72rem;
  font-weight: 900;
}

.kb-arch-stage strong {
  display: block;
  color: var(--kb-ink);
  font-size: 0.96rem;
  font-weight: 900;
  line-height: 1.35;
  word-break: keep-all;
}

.kb-arch-stage p {
  margin: 8px 0 0;
  color: var(--kb-muted);
  font-size: 0.79rem;
  line-height: 1.5;
}

.kb-arch-detail {
  margin-top: 10px;
  padding-top: 8px;
  border-top: 1px solid rgba(17, 17, 17, 0.1);
  color: #514a45;
  font-size: 0.74rem;
  font-weight: 750;
  line-height: 1.45;
  overflow-wrap: anywhere;
}

.kb-arch-arrow {
  display: flex;
  align-items: center;
  justify-content: center;
  color: #8b3f2b;
  font-size: 1.25rem;
  font-weight: 900;
}

.kb-arch-lower {
  display: grid;
  grid-template-columns: minmax(0, 1.1fr) minmax(0, 0.9fr);
  gap: 30px;
  margin-top: 24px;
  padding-top: 22px;
  border-top: 1px solid var(--kb-line);
}

.kb-arch-section-title {
  margin: 0 0 12px;
  color: var(--kb-ink);
  font-size: 1.02rem;
  font-weight: 900;
}

.kb-arch-step {
  display: grid;
  grid-template-columns: 28px minmax(0, 1fr);
  gap: 10px;
  align-items: start;
  padding: 8px 0;
  border-bottom: 1px solid #ececec;
}

.kb-arch-step:last-child {
  border-bottom: 0;
}

.kb-arch-step-no {
  display: inline-flex;
  align-items: center;
  justify-content: center;
  width: 26px;
  height: 26px;
  border-radius: 50%;
  background: var(--kb-ink);
  color: var(--kb-yellow);
  font-size: 0.82rem;
  font-weight: 900;
  line-height: 1;
  margin-top: 0;
}

.kb-arch-step strong,
.kb-arch-control strong {
  display: block;
  color: var(--kb-ink);
  font-size: 0.84rem;
  font-weight: 900;
}

.kb-arch-step > div > span,
.kb-arch-control span {
  display: block;
  margin-top: 3px;
  color: var(--kb-muted);
  font-size: 0.77rem;
  line-height: 1.48;
}

.kb-arch-controls {
  display: grid;
  grid-template-columns: repeat(2, minmax(0, 1fr));
  gap: 16px 18px;
}

.kb-arch-control {
  min-width: 0;
  padding-left: 12px;
  border-left: 4px solid var(--kb-yellow);
}

.kb-arch-control.blue {
  border-left-color: var(--kb-blue);
}

.kb-arch-control.green {
  border-left-color: var(--kb-green);
}

.kb-arch-control.red {
  border-left-color: #a34d36;
}

.kb-ops-lead {
  max-width: 760px;
  margin-top: 10px;
  color: var(--kb-muted);
  font-size: 0.98rem;
  line-height: 1.6;
}

.kb-ops-section-heading {
  margin: 26px 0 16px;
  color: var(--kb-ink);
  font-size: 1.2rem;
  font-weight: 900;
  letter-spacing: 0;
}

.kb-ops-guide-copy {
  margin: -6px 0 18px;
  color: var(--kb-muted);
  font-size: 0.9rem;
  line-height: 1.6;
}

.kb-ops-flow {
  display: grid;
  grid-template-columns: repeat(5, minmax(0, 1fr));
  gap: 24px;
}

.kb-ops-step {
  position: relative;
  min-height: 138px;
  padding: 18px;
  border: 1px solid var(--kb-line);
  border-radius: 8px;
  background: #ffffff;
}

.kb-ops-step:not(:last-child)::after {
  content: "→";
  position: absolute;
  top: 50%;
  right: -20px;
  width: 16px;
  color: #8b3f2b;
  font-size: 1.4rem;
  font-weight: 900;
  line-height: 1;
  transform: translateY(-50%);
}

.kb-ops-step-no {
  display: inline-flex;
  align-items: center;
  justify-content: center;
  width: 34px;
  height: 34px;
  margin-bottom: 16px;
  border-radius: 50%;
  background: #111111;
  color: #ffffff;
  font-size: 0.82rem;
  font-weight: 900;
}

.kb-ops-step.accent {
  border-color: #b56f58;
  background: #fff0ea;
}

.kb-ops-step.accent .kb-ops-step-no {
  background: #8b3f2b;
}

.kb-ops-step.success {
  border-color: #4c876d;
  background: #edf7f1;
}

.kb-ops-step.success .kb-ops-step-no {
  background: #3f7b60;
}

.kb-ops-step strong {
  display: block;
  margin-bottom: 8px;
  color: var(--kb-ink);
  font-size: 1rem;
  font-weight: 900;
}

.kb-ops-step span {
  color: var(--kb-muted);
  font-size: 0.86rem;
  line-height: 1.5;
}

.kb-ops-portal {
  margin-top: 24px;
  padding: 20px;
  border-left: 6px solid var(--kb-yellow);
  background: #fff8d7;
}

.kb-ops-portal strong {
  display: block;
  color: var(--kb-ink);
  font-size: 1rem;
  font-weight: 900;
}

.kb-ops-portal p {
  margin: 6px 0 0;
  color: var(--kb-muted);
  font-size: 0.88rem;
  line-height: 1.55;
}

.kb-ops-link {
  display: inline-flex;
  align-items: center;
  justify-content: center;
  min-height: 46px;
  margin-top: 14px;
  padding: 10px 18px;
  border-radius: 8px;
  background: #111111;
  color: var(--kb-yellow) !important;
  font-size: 0.9rem;
  font-weight: 900;
  text-decoration: none !important;
  box-shadow: 0 8px 18px rgba(17, 17, 17, 0.14);
}

.kb-ops-link:hover {
  background: #006aab;
  color: #ffffff !important;
}

/* HMM minimal presentation: one ocean-blue accent, white surfaces, no heavy shading. */
.stApp,
section[data-testid="stSidebar"] {
  background: #f7fafc !important;
}

.kb-hero,
.kb-sidebar-card,
.kb-arch-stage,
.kb-ops-step,
.kb-ops-portal,
div[data-testid="stForm"],
.st-key-kb_login_container [data-testid="stForm"] {
  background: #ffffff !important;
  border-color: #d6e3ec !important;
  box-shadow: none !important;
}

.kb-section-title.input,
.kb-panel-title,
.kb-arch-stage.experience,
.kb-ops-portal {
  border-color: #004b87 !important;
}

.kb-vpd-card,
div[data-testid="stButton"] > button,
div[data-testid="stButton"] > button[kind="primary"],
div[data-testid="stFormSubmitButton"] > button[data-testid="stBaseButton-primaryFormSubmit"],
.kb-ops-link,
.st-key-kb_login_container [data-testid="stFormSubmitButton"] button {
  background: #004b87 !important;
  border-color: #004b87 !important;
  color: #ffffff !important;
  box-shadow: none !important;
}

div[data-testid="stButton"] > button:hover,
div[data-testid="stFormSubmitButton"] > button[data-testid="stBaseButton-primaryFormSubmit"]:hover,
.st-key-kb_login_container [data-testid="stFormSubmitButton"] button:hover,
.kb-ops-link:hover {
  background: #006aab !important;
  color: #ffffff !important;
}

.kb-vpd-card .kb-vpd-user,
.kb-vpd-card .kb-vpd-meta,
.kb-vpd-card .kb-vpd-scope {
  color: #ffffff !important;
}

h2, h3 {
  letter-spacing: 0;
}

@media (max-width: 1024px) {
  .kb-arch-map {
    grid-template-columns: repeat(2, minmax(0, 1fr));
    gap: 12px;
  }

  .kb-arch-arrow {
    display: none;
  }

  .kb-arch-stage {
    min-height: 190px;
  }

  .kb-arch-experience-grid {
    grid-template-columns: 1fr;
  }

  .kb-ops-flow {
    grid-template-columns: repeat(2, minmax(0, 1fr));
  }

  .kb-ops-step:not(:last-child)::after {
    display: none;
  }
}

@media (max-width: 760px) {
  .st-key-kb_login_container {
    margin-top: 1.5rem;
  }

  .kb-login-brand h1 {
    font-size: 1.78rem;
  }

  .st-key-kb_login_container [data-testid="stForm"] {
    padding: 1.35rem 1.15rem 1.25rem;
  }

  div[data-baseweb="tab-list"] {
    gap: 4px;
  }

  button[data-baseweb="tab"] {
    flex: 1 1 0;
    min-width: 0;
    padding: 10px 6px;
    font-size: 1rem !important;
  }

  .kb-hero {
    grid-template-columns: 1fr;
    grid-template-areas:
      "logo"
      "main";
    gap: 18px;
    padding: 22px;
  }

  .kb-ci-panel {
    justify-content: flex-end;
  }

  .kb-ci-panel img {
    width: min(74%, 230px);
  }

  .kb-hero h1 {
    font-size: 1.68rem;
  }

  .kb-arch-map,
  .kb-arch-lower,
  .kb-arch-controls {
    grid-template-columns: 1fr;
  }

  .kb-arch-map {
    gap: 10px;
  }

  .kb-arch-stage {
    min-height: 0;
  }

  .kb-arch-lower {
    gap: 22px;
  }

  .kb-ops-flow {
    grid-template-columns: 1fr;
  }

  .kb-ops-step:not(:last-child)::after {
    display: none;
  }
}
</style>
"""

HMM_MINIMAL_THEME_CSS = """
<style>
html, body, [data-testid="stAppViewContainer"], .stApp {
  color-scheme: light !important;
  background: #ffffff !important;
  color: #12324a !important;
  font-family: "Noto Sans KR", "Malgun Gothic", sans-serif;
}

#MainMenu { display: none; }
.block-container { max-width: 1120px; padding-top: 1.5rem; }

.kb-hero {
  display: flex;
  align-items: flex-start;
  justify-content: space-between;
  gap: 24px;
  margin: 0 0 28px;
  padding: 0 0 22px;
  border: 0;
  border-bottom: 1px solid #d9e4ec;
  border-radius: 0;
  background: transparent;
  box-shadow: none;
}
.kb-hero::before { display: none; }
.kb-hero-main { min-width: 0; }
.kb-brand-row { display: flex; align-items: center; gap: 10px; margin-bottom: 12px; }
.kb-mini-mark {
  display: inline-flex; align-items: center; justify-content: center;
  min-width: 46px; height: 28px; padding: 0 8px;
  border-radius: 4px; background: #004b87; color: #ffffff;
  font-size: 0.72rem; font-weight: 800;
}
.kb-brand-name { color: #004b87; font-size: 0.9rem; font-weight: 700; }
.kb-brand-sub, .kb-hero p { color: #5f7280; }
.kb-hero h1 { margin: 0; color: #12324a; font-size: 2rem; font-weight: 700; }
.kb-hero p { margin: 10px 0 0; max-width: 680px; line-height: 1.6; }
.kb-ci-panel { flex: 0 0 auto; }
.kb-ci-panel img { display: block; width: 112px; height: auto; }

.st-key-kb_login_container { max-width: 560px; margin: 56px auto 0; }
.kb-login-brand { text-align: center; }
.hmm-login-wordmark {
  margin: 0 auto 12px; color: #004b87; font-size: 2rem;
  font-weight: 800; letter-spacing: 0.08em; line-height: 1;
}
.kb-login-kicker { color: #5f7280; font-size: 0.75rem; font-weight: 700; }
.kb-login-brand h1 { margin: 12px 0 8px; color: #12324a; font-size: 1.8rem; }
.kb-login-brand p { margin: 0 0 22px; color: #5f7280; }
.st-key-kb_login_container [data-testid="stForm"] {
  padding: 20px; border: 1px solid #d9e4ec; border-radius: 8px;
  background: #ffffff; box-shadow: none;
}
.st-key-kb_login_container [data-testid="stFormSubmitButton"] button {
  background: #004b87; color: #ffffff; border: 1px solid #004b87; box-shadow: none;
}
.st-key-kb_login_container [data-testid="stFormSubmitButton"] button:hover {
  background: #006aab; border-color: #006aab; color: #ffffff;
}

.kb-vpd-card {
  margin: 10px 0; padding: 14px; border: 1px solid #d9e4ec;
  border-radius: 6px; background: #ffffff; color: #12324a; box-shadow: none;
}
.kb-vpd-card .kb-vpd-user { color: #004b87; font-weight: 700; }
.kb-vpd-card .kb-vpd-meta, .kb-vpd-card .kb-vpd-scope { color: #5f7280; }

@media (max-width: 760px) {
  .kb-hero { display: block; }
  .kb-ci-panel { margin-top: 18px; }
  .kb-hero h1 { font-size: 1.65rem; }
}
</style>
"""

HMM_CLEAN_THEME_CSS = """
<style>
:root {
  --hmm-navy: #003b70;
  --hmm-ink: #172b3a;
  --hmm-muted: #667785;
  --hmm-line: #dfe7ed;
  --hmm-surface: #ffffff;
}

html, body, [data-testid="stAppViewContainer"], .stApp {
  background: var(--hmm-surface) !important;
  color: var(--hmm-ink) !important;
  font-family: "Noto Sans KR", "Malgun Gothic", sans-serif;
}

#MainMenu, header [data-testid="stToolbar"] { display: none !important; }
.block-container { max-width: 1180px; padding: 2rem 3rem 4rem; }

section[data-testid="stSidebar"] {
  background: #ffffff !important;
  border-right: 1px solid var(--hmm-line);
}
section[data-testid="stSidebar"] > div { background: #ffffff !important; }
section[data-testid="stSidebar"] [data-testid="stMarkdownContainer"] p { color: var(--hmm-ink); }

input, textarea, [data-baseweb="select"] > div {
  border-radius: 4px !important;
  border-color: #cfdbe4 !important;
  box-shadow: none !important;
}
div[data-testid="stButton"] > button,
div[data-testid="stFormSubmitButton"] > button {
  border-radius: 4px !important;
  box-shadow: none !important;
}
div[data-testid="stButton"] > button[kind="primary"],
div[data-testid="stFormSubmitButton"] > button[data-testid="stBaseButton-primaryFormSubmit"] {
  background: var(--hmm-navy) !important;
  border-color: var(--hmm-navy) !important;
  color: #ffffff !important;
}

.hmm-app-header {
  display: flex;
  align-items: baseline;
  gap: 18px;
  margin: 0 0 28px;
  padding: 0 0 22px;
  border-bottom: 1px solid var(--hmm-line);
}
.hmm-app-wordmark, .hmm-login-wordmark {
  color: var(--hmm-navy);
  font-size: 1.35rem;
  font-weight: 800;
  letter-spacing: 0.08em;
}
.hmm-app-header h1 { margin: 0; color: var(--hmm-ink); font-size: 1.7rem; }
.hmm-app-header p { margin: 0; color: var(--hmm-muted); font-size: 0.92rem; }

.st-key-kb_login_container { max-width: 440px; margin: 12vh auto 0; }
.kb-login-brand { text-align: left; }
.kb-login-kicker { margin-top: 12px; color: var(--hmm-muted); font-size: 0.76rem; font-weight: 700; }
.kb-login-brand h1 { margin: 12px 0 8px; color: var(--hmm-ink); font-size: 1.7rem; }
.kb-login-brand p { margin: 0 0 22px; color: var(--hmm-muted); }
.st-key-kb_login_container [data-testid="stForm"] {
  padding: 0; border: 0; background: transparent; box-shadow: none;
}
.st-key-kb_login_container [data-testid="stFormSubmitButton"] button {
  min-height: 44px; background: var(--hmm-navy); border-color: var(--hmm-navy);
  color: #ffffff; box-shadow: none;
}
.kb-login-note { color: var(--hmm-muted); }

@media (max-width: 760px) {
  .block-container { padding: 1.25rem 1.25rem 3rem; }
  .hmm-app-header { display: block; }
  .hmm-app-header h1 { margin-top: 10px; }
  .st-key-kb_login_container { margin-top: 8vh; }
}
</style>
"""


@dataclass(frozen=True)
class McpServer:
    server_id: str
    endpoint_url: str
    auth_token_env: str
    default_tool: str
    tool_allowlist: tuple[str, ...]
    router_model_profile: str
    description: str


@dataclass(frozen=True)
class VpdTokenPreset:
    user_id: str
    name: str
    role: str
    channel: str
    scope: str
    token: str = field(repr=False, compare=False)
    is_default: bool = False

    @property
    def display_label(self) -> str:
        return " · ".join(
            item
            for item in (
                self.user_id,
                self.name,
                self.role,
                self.channel,
                self.scope,
            )
            if item
        )

    @property
    def select_label(self) -> str:
        return " · ".join(
            item
            for item in (
                self.user_id,
                self.name,
            )
            if item
        )


class PublicMcpError(RuntimeError):
    """Safe UI error. Never include request headers, tokens, or raw traces."""


class AnswerSynthesisError(RuntimeError):
    """Safe final-answer synthesis error."""

    def __init__(
        self,
        message: str,
        diagnostics: Mapping[str, Any] | None = None,
    ) -> None:
        super().__init__(message)
        self.diagnostics = dict(diagnostics or {})


@dataclass(frozen=True)
class McpDiscoveryResult:
    server: McpServer
    tools: tuple[McpTool, ...]


@dataclass(frozen=True)
class _JsonRpcExchange:
    result: Mapping[str, Any]
    session_id: str = ""


class _McpHttpStatusError(PublicMcpError):
    def __init__(self, code: int) -> None:
        self.code = code
        super().__init__(f"MCP 호출이 실패했습니다. HTTP {code}")


def _utc_now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def _chat_db_path() -> Path:
    configured = _runtime_env_value("POC4_CHAT_DB_PATH")
    if not configured:
        return CHAT_DB_FILE
    path = Path(configured).expanduser()
    return path if path.is_absolute() else ROOT / path


def _chat_db_connect() -> sqlite3.Connection:
    path = _chat_db_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    connection = sqlite3.connect(path)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA journal_mode=WAL")
    connection.execute("PRAGMA busy_timeout=5000")
    return connection


def init_chat_store() -> None:
    with _chat_db_connect() as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS poc4_mcp_chat_conversations (
                conversation_id TEXT PRIMARY KEY,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                title TEXT NOT NULL DEFAULT ''
            )
            """
        )
        columns = {
            str(row["name"])
            for row in connection.execute(
                "PRAGMA table_info(poc4_mcp_chat_conversations)"
            ).fetchall()
        }
        if "title" not in columns:
            connection.execute(
                "ALTER TABLE poc4_mcp_chat_conversations "
                "ADD COLUMN title TEXT NOT NULL DEFAULT ''"
            )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS poc4_mcp_chat_turns (
                turn_id INTEGER PRIMARY KEY AUTOINCREMENT,
                conversation_id TEXT NOT NULL,
                created_at TEXT NOT NULL,
                selected_user_id TEXT,
                selected_user_label TEXT,
                question TEXT NOT NULL,
                standalone_question TEXT NOT NULL,
                answer TEXT NOT NULL,
                basis_json TEXT NOT NULL,
                limitations TEXT NOT NULL,
                details_json TEXT NOT NULL,
                FOREIGN KEY(conversation_id)
                    REFERENCES poc4_mcp_chat_conversations(conversation_id)
            )
            """
        )
        connection.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_poc4_chat_turns_conversation_latest
            ON poc4_mcp_chat_turns(conversation_id, turn_id DESC)
            """
        )


def ensure_conversation(conversation_id: str) -> None:
    now = _utc_now()
    with _chat_db_connect() as connection:
        connection.execute(
            """
            INSERT OR IGNORE INTO poc4_mcp_chat_conversations
                (conversation_id, created_at, updated_at, title)
            VALUES (?, ?, ?, '')
            """,
            (conversation_id, now, now),
        )


def count_chat_turns(conversation_id: str) -> int:
    with _chat_db_connect() as connection:
        row = connection.execute(
            """
            SELECT COUNT(*) AS count
            FROM poc4_mcp_chat_turns
            WHERE conversation_id = ?
            """,
            (conversation_id,),
        ).fetchone()
    return int(row["count"] if row else 0)


def list_conversations(limit: int = 50) -> list[dict[str, Any]]:
    with _chat_db_connect() as connection:
        rows = connection.execute(
            """
            SELECT
                c.conversation_id,
                c.created_at,
                c.updated_at,
                c.title,
                COUNT(t.turn_id) AS turn_count,
                GROUP_CONCAT(
                    COALESCE(t.question, '') || ' ' ||
                    COALESCE(t.standalone_question, '') || ' ' ||
                    COALESCE(t.answer, '') || ' ' ||
                    COALESCE(t.selected_user_id, '') || ' ' ||
                    COALESCE(t.selected_user_label, ''),
                    ' '
                ) AS search_text,
                (
                    SELECT latest.question
                    FROM poc4_mcp_chat_turns latest
                    WHERE latest.conversation_id = c.conversation_id
                    ORDER BY latest.turn_id DESC
                    LIMIT 1
                ) AS latest_question
            FROM poc4_mcp_chat_conversations c
            LEFT JOIN poc4_mcp_chat_turns t
                ON t.conversation_id = c.conversation_id
            GROUP BY c.conversation_id, c.created_at, c.updated_at, c.title
            HAVING COUNT(t.turn_id) > 0
            ORDER BY c.updated_at DESC
            LIMIT ?
            """,
            (limit,),
        ).fetchall()
    return [dict(row) for row in rows]


def rename_conversation(conversation_id: str, title: str) -> None:
    with _chat_db_connect() as connection:
        connection.execute(
            """
            UPDATE poc4_mcp_chat_conversations
            SET title = ?, updated_at = ?
            WHERE conversation_id = ?
            """,
            (title.strip()[:80], _utc_now(), conversation_id),
        )


def delete_conversation(conversation_id: str) -> None:
    with _chat_db_connect() as connection:
        connection.execute(
            "DELETE FROM poc4_mcp_chat_turns WHERE conversation_id = ?",
            (conversation_id,),
        )
        connection.execute(
            "DELETE FROM poc4_mcp_chat_conversations WHERE conversation_id = ?",
            (conversation_id,),
        )


def load_all_chat_turns(conversation_id: str) -> list[dict[str, Any]]:
    with _chat_db_connect() as connection:
        rows = connection.execute(
            """
            SELECT turn_id, created_at, selected_user_id, selected_user_label,
                   question, standalone_question, answer, basis_json,
                   limitations, details_json
            FROM poc4_mcp_chat_turns
            WHERE conversation_id = ?
            ORDER BY turn_id ASC
            """,
            (conversation_id,),
        ).fetchall()
    return [_chat_turn_from_row(row) for row in rows]


def filter_conversations(
    rows: list[dict[str, Any]],
    search_text: str,
) -> list[dict[str, Any]]:
    query = search_text.strip().casefold()
    if not query:
        return rows
    filtered: list[dict[str, Any]] = []
    for row in rows:
        haystack = " ".join(
            str(row.get(key) or "")
            for key in (
                "conversation_id",
                "title",
                "latest_question",
                "search_text",
                "updated_at",
            )
        ).casefold()
        if query in haystack:
            filtered.append(row)
    return filtered


def conversation_label(item: Mapping[str, Any]) -> str:
    turn_count = int(item.get("turn_count") or 0)
    saved_title = str(item.get("title") or "").strip()
    latest_question = str(item.get("latest_question") or "").strip()
    if saved_title:
        title = saved_title[:34] + ("..." if len(saved_title) > 34 else "")
    elif latest_question:
        title = latest_question[:34] + ("..." if len(latest_question) > 34 else "")
    else:
        title = "새 대화"
    updated_at = str(item.get("updated_at") or "")
    short_id = str(item.get("conversation_id") or "")[-8:]
    return f"{title} · {turn_count}건 · {updated_at} · {short_id}"


def load_chat_turns(conversation_id: str, page: int) -> list[dict[str, Any]]:
    offset = max(page, 0) * CHAT_TURNS_PER_PAGE
    with _chat_db_connect() as connection:
        rows = connection.execute(
            """
            SELECT turn_id, created_at, selected_user_id, selected_user_label,
                   question, standalone_question, answer, basis_json,
                   limitations, details_json
            FROM poc4_mcp_chat_turns
            WHERE conversation_id = ?
            ORDER BY turn_id DESC
            LIMIT ? OFFSET ?
            """,
            (conversation_id, CHAT_TURNS_PER_PAGE, offset),
        ).fetchall()
    return [_chat_turn_from_row(row) for row in rows]


def _is_failed_synthesis_answer(value: object) -> bool:
    text = str(value or "").strip()
    if not text:
        return False
    return (
        text.startswith("MCP 조회는 완료됐지만 최종 답변 합성에 실패했습니다.")
        or "MCP 결과 기반 최종 답변 생성에 실패했습니다" in text
    )


def load_chat_context(conversation_id: str) -> list[dict[str, str]]:
    with _chat_db_connect() as connection:
        rows = connection.execute(
            """
            SELECT question, answer
            FROM poc4_mcp_chat_turns
            WHERE conversation_id = ?
            ORDER BY turn_id DESC
            LIMIT ?
            """,
            (conversation_id, CHAT_CONTEXT_TURNS),
        ).fetchall()
    messages: list[dict[str, str]] = []
    for row in reversed(rows):
        question = str(row["question"] or "").strip()
        answer = str(row["answer"] or "").strip()
        if question:
            messages.append({"role": "user", "content": question[:1200]})
        if answer and not _is_failed_synthesis_answer(answer):
            messages.append({"role": "assistant", "content": answer[:1200]})
    return messages[-MAX_CONVERSATION_MESSAGES:]


def save_chat_turn(
    *,
    conversation_id: str,
    selected_user_id: str,
    selected_user_label: str,
    question: str,
    standalone_question: str,
    answer: str,
    basis: Any,
    limitations: str,
    details: Mapping[str, Any],
) -> None:
    now = _utc_now()
    with _chat_db_connect() as connection:
        connection.execute(
            """
            INSERT OR IGNORE INTO poc4_mcp_chat_conversations
                (conversation_id, created_at, updated_at, title)
            VALUES (?, ?, ?, '')
            """,
            (conversation_id, now, now),
        )
        connection.execute(
            """
            INSERT INTO poc4_mcp_chat_turns (
                conversation_id, created_at, selected_user_id,
                selected_user_label, question, standalone_question, answer,
                basis_json, limitations, details_json
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                conversation_id,
                now,
                selected_user_id,
                selected_user_label,
                question,
                standalone_question,
                answer,
                json.dumps(basis if isinstance(basis, list) else [], ensure_ascii=False),
                limitations,
                json.dumps(dict(details), ensure_ascii=False, default=str),
            ),
        )
        connection.execute(
            """
            UPDATE poc4_mcp_chat_conversations
            SET updated_at = ?,
                title = CASE
                    WHEN title IS NULL OR title = '' THEN ?
                    ELSE title
                END
            WHERE conversation_id = ?
            """,
            (now, question[:60], conversation_id),
        )


def _json_loads_or(value: object, fallback: Any) -> Any:
    if not isinstance(value, str):
        return fallback
    try:
        return json.loads(value)
    except ValueError:
        return fallback


def _chat_turn_from_row(row: sqlite3.Row) -> dict[str, Any]:
    return {
        "turn_id": row["turn_id"],
        "created_at": row["created_at"],
        "selected_user_id": row["selected_user_id"],
        "selected_user_label": row["selected_user_label"],
        "question": row["question"],
        "standalone_question": row["standalone_question"],
        "answer": row["answer"],
        "basis": _json_loads_or(row["basis_json"], []),
        "limitations": row["limitations"],
        "details": _json_loads_or(row["details_json"], {}),
    }


def new_conversation_id() -> str:
    return "poc4-" + uuid4().hex


def _question_label(question: object) -> str:
    question_id = str(
        getattr(question, "question_code", getattr(question, "question_id", ""))
    )
    category = str(getattr(question, "category", ""))
    title = str(
        getattr(question, "title", getattr(question, "question_text", ""))
    )
    return f"{question_id} · {category} · {title}"


@st.cache_resource(show_spinner=False)
def _qa_history_store() -> QaHistoryStore:
    return QaHistoryStore()


def _load_qa_questions() -> tuple[list[QaQuestion], QaHistoryStore | None, str]:
    try:
        return list(load_benchmark_questions(QA_BENCHMARK_FILE)), _qa_history_store(), ""
    except Exception as exc:
        return [], None, f"질답 기준 파일을 읽지 못했습니다: {exc}"


def _judgment_label(status: str) -> str:
    labels = {
        "PASS": "통과",
        "WARN": "검토",
        "FAIL": "실패",
        "REVIEW": "수동 검토",
    }
    return labels.get(str(status or "").upper(), "미실행")


def _qa_answer_preview(value: object, limit: int = 180) -> str:
    text = str(value or "").strip().replace("\n", " ")
    return text if len(text) <= limit else text[:limit] + "..."


def _render_qa_benchmark_panel(
    questions: list[QaQuestion],
    store: QaHistoryStore | None,
    store_error: str,
) -> QaQuestion | None:
    st.markdown(
        '<div class="kb-section-title input" role="heading" aria-level="3">'
        "고객 질답 검증 시나리오"
        "</div>",
        unsafe_allow_html=True,
    )
    st.caption(
        "고객 제공 Excel에서 정리한 47개 후보입니다. 후보를 선택하면 기준 답변과 과거 검증 이력을 확인하고 바로 실행할 수 있습니다."
    )
    if store_error:
        st.warning(
            "현재는 파일 기준 후보만 표시합니다. ADB 이력 저장은 연결 복구 후 사용할 수 있습니다. "
            + store_error
        )
    if not questions:
        st.error("표시할 고객 질답 기준이 없습니다.")
        return None

    category_options = tuple(sorted({question.category for question in questions}))
    category = st.selectbox(
        "시나리오 구분",
        options=("ALL", *category_options),
        format_func=lambda item: "전체 47개" if item == "ALL" else item,
        key="poc4_qa_category_filter",
    )
    visible_questions = [
        question for question in questions if category == "ALL" or question.category == category
    ]
    st.dataframe(
        [
            {
                "케이스": question.question_code,
                "구분": question.category,
                "제목": question.title,
                "질문": question.question_text,
                "기대 기준": _qa_answer_preview(question.expected_focus, 150),
                "기준 상태": _judgment_label(question.last_judgment_status)
                if question.last_judgment_status
                else "과거 기준",
                "최근 실행": question.last_evaluated_at or "후보 선택 시 조회",
            }
            for question in visible_questions
        ],
        hide_index=True,
        width="stretch",
        height=min(520, 110 + 36 * len(visible_questions)),
        column_config={
            "질문": st.column_config.TextColumn(width="large"),
            "기대 기준": st.column_config.TextColumn(width="large"),
            "최근 실행": st.column_config.TextColumn(width="medium"),
        },
    )
    by_code = {question.question_code: question for question in visible_questions}
    selected_code = st.selectbox(
        "실행할 검증 후보 선택",
        options=("", *by_code),
        format_func=lambda code: "선택 안 함 · 자유 질의" if not code else _question_label(by_code[code]),
        key="poc4_qa_question_code",
    )
    selected = by_code.get(selected_code)
    if selected is None:
        st.info("자유 텍스트 질문도 실행하고 이력으로 남길 수 있습니다. 자유 질의는 정답 기준이 없어 수동 검토로 표시됩니다.")
        return None

    if store is not None:
        try:
            stored_question = store.get_question_by_code(selected.question_code)
            if stored_question is not None:
                selected = stored_question
            else:
                st.warning("선택한 후보가 ADB 이력 테이블에 아직 적재되지 않았습니다.")
        except QaHistoryStoreError as exc:
            st.warning("선택한 후보의 ADB 이력을 불러오지 못했습니다. " + str(exc))

    with st.expander("선택한 질문의 정답 기준·원본·과거 답변", expanded=True):
        left, right = st.columns(2)
        with left:
            st.markdown("**정답 기준**")
            st.write(selected.expected_focus or "정답 기준 설명이 없습니다.")
            st.markdown("**기준 답변**")
            st.code(selected.baseline_answer or "기준 실행 결과가 없습니다.", language=None)
        with right:
            st.markdown("**고객 원본 출처**")
            st.write(
                f"{selected.source_document} / {selected.source_sheet} / 행 {selected.source_row or '-'}"
            )
            if selected.source_scenario:
                st.caption(selected.source_scenario)
            st.markdown("**기준 SQL**")
            st.code(selected.baseline_sql or selected.sample_sql or "기준 SQL이 없습니다.", language="sql")

        if store is not None and selected.question_id is not None:
            try:
                answers = store.list_answers(selected.question_id, limit=30)
            except QaHistoryStoreError as exc:
                st.warning(str(exc))
                answers = []
            if answers:
                st.markdown("**과거 답변 이력**")
                st.dataframe(
                    [
                        {
                            "순번": item["answer_seq"],
                            "구분": "과거 기준" if item["answer_kind"] == "HISTORICAL" else "이번 실행",
                            "판정": _judgment_label(item["judgment_status"]),
                            "실행 상태": item["execution_status"],
                            "실행 시각": item["requested_at"],
                            "답변": _qa_answer_preview(item["answer_text"]),
                        }
                        for item in answers
                    ],
                    hide_index=True,
                    width="stretch",
                )
                answer_by_seq = {str(item["answer_seq"]): item for item in answers}
                detail_seq = st.selectbox(
                    "답변 이력 상세",
                    options=tuple(answer_by_seq),
                    format_func=lambda value: (
                        f"#{value} · {_judgment_label(answer_by_seq[value]['judgment_status'])} · "
                        f"{answer_by_seq[value]['requested_at']}"
                    ),
                    key=f"poc4_qa_answer_detail_{selected.question_id}",
                )
                detail = answer_by_seq[detail_seq]
                st.markdown("**판정 근거**")
                st.write(detail["judgment_reason"] or "판정 근거가 없습니다.")
                st.markdown("**생성 SQL**")
                st.code(detail["generated_sql"] or "생성 SQL이 없습니다.", language="sql")
                st.markdown("**실행 답변**")
                st.write(detail["answer_text"] or "답변이 없습니다.")
            else:
                st.info("아직 저장된 답변 이력이 없습니다.")
    return selected


def _apply_console_theme(profile: AppProfile) -> None:
    apply_console_theme(st, profile)


def _portal_auth_value(name: str) -> str:
    return (os.environ.get(name) or _dotenv_value(name)).strip()


def _portal_password_matches(password: str, encoded_password: str) -> bool:
    try:
        scheme, iterations_text, salt_hex, expected_hex = encoded_password.split("$", 3)
        iterations = int(iterations_text)
        salt = bytes.fromhex(salt_hex)
        expected = bytes.fromhex(expected_hex)
    except (TypeError, ValueError):
        return False
    if scheme != "pbkdf2_sha256" or not 100_000 <= iterations <= 2_000_000:
        return False
    candidate = hashlib.pbkdf2_hmac(
        "sha256",
        password.encode("utf-8"),
        salt,
        iterations,
    )
    return hmac.compare_digest(candidate, expected)


def _portal_auth_configured() -> bool:
    return bool(
        _portal_auth_value("POC4_LOGIN_USER")
        and _portal_auth_value("POC4_LOGIN_PASSWORD_PBKDF2")
    )


def _portal_credentials_are_valid(username: str, password: str) -> bool:
    expected_username = _portal_auth_value("POC4_LOGIN_USER")
    encoded_password = _portal_auth_value("POC4_LOGIN_PASSWORD_PBKDF2")
    username_matches = hmac.compare_digest(username.strip(), expected_username)
    password_matches = _portal_password_matches(password, encoded_password)
    return username_matches and password_matches


def _portal_cookie_user() -> str:
    """Return the verified portal user stored in the HttpOnly auth cookie."""
    try:
        token = str(st.context.cookies.get(PORTAL_AUTH_COOKIE_NAME, "")).strip()
        encoded_claims, supplied_signature = token.rsplit(".", 1)
        secret = _portal_auth_value("POC4_LOGIN_REMEMBER_SECRET")
        expected_signature = hmac.new(
            secret.encode("utf-8"), encoded_claims.encode("ascii"), hashlib.sha256
        ).hexdigest()
        if not secret or not hmac.compare_digest(supplied_signature, expected_signature):
            return ""
        padding = "=" * (-len(encoded_claims) % 4)
        claims = json.loads(
            base64.urlsafe_b64decode((encoded_claims + padding).encode("ascii")).decode("utf-8")
        )
        username = str(claims.get("u", "")).strip()
        expires_at = int(claims.get("e", 0))
        expected_username = _portal_auth_value("POC4_LOGIN_USER")
        if claims.get("v") != 1 or expires_at <= int(datetime.now(timezone.utc).timestamp()):
            return ""
        if not username or not hmac.compare_digest(username, expected_username):
            return ""
        return username
    except (AttributeError, TypeError, ValueError, UnicodeDecodeError, json.JSONDecodeError):
        return ""


def _restore_portal_session_from_cookie() -> None:
    if st.session_state.get(PORTAL_AUTHENTICATED_KEY, False):
        return
    username = _portal_cookie_user()
    if username:
        st.session_state[PORTAL_AUTHENTICATED_KEY] = True
        st.session_state[PORTAL_AUTH_USER_KEY] = username
        st.session_state[PORTAL_LOGIN_FAILURE_KEY] = False


def _render_portal_login(profile: AppProfile) -> None:
    with st.container(key="console_login_container"):
        render_login_brand(st, profile)
        if not _portal_auth_configured():
            st.info("데모 계정 설정 중입니다. 운영 담당자에게 계정 발급을 요청해 주세요.")
            return
        if st.query_params.get("login") == "failed":
            st.error("사용자 ID 또는 비밀번호를 확인해 주세요.")
        components.html(
            """
            <style>
              * { box-sizing: border-box; }
              body { margin: 0; font-family: sans-serif; color: #15293a; }
              form { display: grid; gap: 10px; }
              label { font-size: 13px; font-weight: 700; color: #365064; }
              input { width: 100%; border: 1px solid #afc4d3; border-radius: 7px;
                      padding: 11px 12px; font-size: 15px; color: #15293a; }
              button { margin-top: 6px; width: 100%; border: 0; border-radius: 7px;
                       padding: 12px; background: #005c97; color: white; cursor: pointer;
                       font-size: 15px; font-weight: 700; }
              button:hover { background: #004b7c; }
            </style>
            <form action="/_poc4/auth/login" method="post" target="_parent">
              <label>사용자 ID
                <input name="username" autocomplete="username" maxlength="80" required>
              </label>
              <label>비밀번호
                <input name="password" type="password" autocomplete="current-password"
                       maxlength="200" required>
              </label>
              <button type="submit">로그인</button>
            </form>
            """,
            height=220,
        )
        st.markdown(
            f'<p class="console-muted">{html.escape(profile.login_footer)}</p>',
            unsafe_allow_html=True,
        )


def _logout_portal() -> None:
    st.session_state.pop(PORTAL_AUTHENTICATED_KEY, None)
    st.session_state.pop(PORTAL_AUTH_USER_KEY, None)
    st.session_state.pop(PORTAL_LOGIN_FAILURE_KEY, None)
    components.html(
        '<script>window.parent.location.assign("/_poc4/auth/logout");</script>',
        height=0,
    )


def _render_app_header(profile: AppProfile) -> None:
    render_console_header(st, profile)


def _render_vpd_user_card(preset: VpdTokenPreset) -> None:
    st.markdown(
        f"""
        <div class="kb-vpd-card">
          <div class="kb-vpd-user">
            {html.escape(preset.user_id)} · {html.escape(preset.name)}
          </div>
          <div class="kb-vpd-meta">
            {html.escape(preset.role)} · {html.escape(preset.channel)}
          </div>
          <div class="kb-vpd-scope">
            권한 범위: {html.escape(preset.scope)}
          </div>
        </div>
        """,
        unsafe_allow_html=True,
    )


def _normalized_bearer(value: object) -> str:
    if not isinstance(value, str):
        return ""
    token = value.strip()
    if token.lower().startswith("bearer "):
        token = token[7:].strip()
    return token if _OPAQUE_BEARER.fullmatch(token) else ""


def _token_fingerprint(token: str) -> str:
    value = str(token or "")
    if not value:
        return "empty"
    digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:12]
    return f"len={len(value)}, sha256={digest}"


def _bounded_json(value: Any, max_chars: int = 24000) -> str:
    text = json.dumps(value, ensure_ascii=False, default=str)
    if len(text) <= max_chars:
        return text
    return text[:max_chars] + "\n...<truncated>"


def _default_vpd_user_id(
    presets: tuple[VpdTokenPreset, ...],
) -> str | None:
    for preset in presets:
        if preset.user_id == DEFAULT_VPD_USER_ID:
            return preset.user_id
    for preset in presets:
        if preset.is_default:
            return preset.user_id
    return presets[0].user_id if presets else None


def _dotenv_value(name: str, path: Path = ENV_FILE) -> str:
    if not name or not path.exists():
        return ""
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except (OSError, UnicodeError):
        return ""
    for line in lines:
        stripped = line.strip()
        if not stripped or stripped.startswith("#") or "=" not in stripped:
            continue
        if stripped.startswith("export "):
            stripped = stripped[7:].lstrip()
        key, raw_value = stripped.split("=", 1)
        if key.strip() != name:
            continue
        value = raw_value.strip()
        if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
            value = value[1:-1]
        return value.strip()
    return ""


class AuditLogError(RuntimeError):
    """Safe audit-log error that never includes DB credentials or provider details."""


def _audit_db_env_value(name: str, default: str = "") -> str:
    return (
        os.environ.get(name)
        or _dotenv_value(name, AUDIT_DB_ENV_FILE)
        or default
    ).strip()


@st.cache_resource(show_spinner=False)
def _audit_db_pool() -> Any:
    password = _audit_db_env_value("ORACLE_DB_PASSWORD")
    if not password:
        raise AuditLogError("감사로그 DB 접속 설정을 확인해 주세요.")
    wallet_dir = Path(
        _audit_db_env_value(
            "ORACLE_WALLET_DIR",
            "/home/opc/wallet/kbaipoc",
        )
    ).expanduser().resolve()
    if not wallet_dir.is_dir():
        raise AuditLogError("감사로그 DB Wallet 경로를 확인해 주세요.")
    try:
        return oracledb.create_pool(
            user=_audit_db_env_value("ORACLE_DB_USER", "ADMIN"),
            password=password,
            dsn=_audit_db_env_value("ORACLE_DSN", "kbaipoc_high"),
            config_dir=str(wallet_dir),
            wallet_location=str(wallet_dir),
            wallet_password=_audit_db_env_value("ORACLE_WALLET_PASSWORD") or None,
            min=1,
            max=2,
            increment=1,
            getmode=oracledb.POOL_GETMODE_WAIT,
        )
    except (oracledb.Error, OSError, ValueError):
        raise AuditLogError("감사로그 DB에 연결하지 못했습니다.") from None


def _audit_rows(
    sql: str,
    binds: Mapping[str, Any] | None = None,
) -> list[dict[str, Any]]:
    try:
        with _audit_db_pool().acquire() as connection:
            with connection.cursor() as cursor:
                cursor.execute(sql, dict(binds or {}))
                columns = [item[0].lower() for item in cursor.description]
                records: list[dict[str, Any]] = []
                for row in cursor:
                    record: dict[str, Any] = {}
                    for column, value in zip(columns, row):
                        if hasattr(value, "read"):
                            value = value.read()
                        record[column] = value
                    records.append(record)
                return records
    except AuditLogError:
        raise
    except (oracledb.Error, OSError, ValueError):
        raise AuditLogError("감사로그를 조회하지 못했습니다.") from None


@st.cache_data(ttl=60, show_spinner=False)
def _load_fga_inventory() -> dict[str, list[dict[str, Any]]]:
    policies = _audit_rows(
        """
        SELECT policy.object_name,
               policy.policy_name,
               policy.policy_text,
               LISTAGG(policy_columns.policy_column, ',') WITHIN GROUP (
                   ORDER BY policy_columns.policy_column
               ) AS policy_column,
               policy.enabled,
               policy.sel,
               policy.ins,
               policy.upd,
               policy.del
        FROM dba_audit_policies policy
        LEFT JOIN dba_audit_policy_columns policy_columns
          ON policy_columns.object_schema = policy.object_schema
         AND policy_columns.object_name = policy.object_name
         AND policy_columns.policy_name = policy.policy_name
        WHERE policy.object_schema = :schema
        GROUP BY policy.object_name,
                 policy.policy_name,
                 policy.policy_text,
                 policy.enabled,
                 policy.sel,
                 policy.ins,
                 policy.upd,
                 policy.del
        ORDER BY policy.object_name, policy.policy_name
        """,
        {"schema": AUDIT_SCHEMA},
    )
    catalog = _audit_rows(
        """
        SELECT object_name,
               column_name,
               policy_name,
               policy_expression,
               enabled_yn,
               description,
               updated_at
        FROM POC_2.KB_SECURITY_POLICY_CATALOG
        WHERE control_type = 'DBMS_FGA'
        ORDER BY object_name, policy_name, column_name
        """
    )
    return {"policies": policies, "catalog": catalog}


@st.cache_data(ttl=30, show_spinner=False)
def _load_fga_audit_events(
    days: int,
    row_limit: int,
    policy_name: str,
    object_name: str,
) -> list[dict[str, Any]]:
    return _audit_rows(
        """
        SELECT *
        FROM (
            SELECT TO_CHAR(
                       audit_event.event_timestamp AT TIME ZONE 'Asia/Seoul',
                       'YYYY-MM-DD HH24:MI:SS'
                   ) AS event_time,
                   audit_event.dbusername,
                   audit_event.client_identifier,
                   audit_event.userhost,
                   audit_event.object_schema,
                   audit_event.object_name,
                   audit_event.action_name,
                   audit_event.fga_policy_name,
                   policy_columns.audit_column,
                   audit_event.return_code,
                   DBMS_LOB.SUBSTR(audit_event.sql_text, 1000, 1) AS sql_text
            FROM unified_audit_trail audit_event
            LEFT JOIN (
                SELECT object_schema,
                       object_name,
                       policy_name,
                       LISTAGG(policy_column, ',') WITHIN GROUP (
                           ORDER BY policy_column
                       ) AS audit_column
                FROM dba_audit_policy_columns
                WHERE object_schema = :schema
                GROUP BY object_schema, object_name, policy_name
            ) policy_columns
              ON policy_columns.object_schema = audit_event.object_schema
             AND policy_columns.object_name = audit_event.object_name
             AND policy_columns.policy_name = audit_event.fga_policy_name
            WHERE audit_event.object_schema = :schema
              AND audit_event.fga_policy_name IS NOT NULL
              AND audit_event.event_timestamp >= (
                  SYSTIMESTAMP - NUMTODSINTERVAL(:days, 'DAY')
              )
              AND (:policy_name IS NULL OR audit_event.fga_policy_name = :policy_name)
              AND (:object_name IS NULL OR audit_event.object_name = :object_name)
            ORDER BY audit_event.event_timestamp DESC
        )
        WHERE ROWNUM <= :row_limit
        """,
        {
            "schema": AUDIT_SCHEMA,
            "days": int(days),
            "policy_name": policy_name or None,
            "object_name": object_name or None,
            "row_limit": int(row_limit),
        },
    )


def _security_evidence_kind(question: str) -> str:
    text = " ".join(str(question or "").casefold().split())
    if "보험료" in text and any(
        term in text for term in ("합계", "개별", "상세", "마스킹")
    ):
        return "PREMIUM_MASK_AGGREGATE"
    if any(term in text for term in ("주민번호", "rrn_masked")):
        return "RRN_DISPLAY_MASK"
    if (
        any(term in text for term in ("공통계정", "공통 계정"))
        and "채널" in text
    ):
        return "CHANNEL_SCOPE"
    if re.search(r"\bCT[0-9]+\b", str(question or ""), re.IGNORECASE) and any(
        term in text for term in ("조회되는지", "담당 고객이 아님", "권한")
    ):
        return "CONTRACT_SCOPE"
    if any(term in text for term in ("로그인하지 않은", "미인증", "인증 실패")):
        return "AUTH_DENIAL"
    return ""


def _safe_audit_log(
    cursor: Any,
    *,
    user_id: str,
    action_name: str,
    policy_result: str,
    request_text: str,
    response_summary: str,
) -> Mapping[str, Any]:
    cursor.callproc(
        "POC_2.LOG_TOOL_CALL",
        [
            user_id,
            "KB_AI_PORTAL",
            "KB_SECURITY_EVIDENCE",
            action_name,
            policy_result,
            request_text[:1000],
            response_summary[:2000],
        ],
    )
    cursor.execute(
        """
        SELECT log_id, TO_CHAR(event_ts, 'YYYY-MM-DD HH24:MI:SS')
        FROM POC_2.KB_SECURITY_AUDIT_LOG
        WHERE log_id = (
            SELECT MAX(log_id)
            FROM POC_2.KB_SECURITY_AUDIT_LOG
            WHERE end_user_id = :user_id
              AND tool_name = 'KB_SECURITY_EVIDENCE'
              AND action_name = :action_name
        )
        """,
        {"user_id": user_id, "action_name": action_name},
    )
    row = cursor.fetchone()
    return {
        "audit_log_id": int(row[0]) if row and row[0] is not None else None,
        "audit_event_time": str(row[1]) if row and row[1] is not None else "",
    }


def collect_security_evidence(
    question: str,
    token_preset: VpdTokenPreset | None,
    *,
    failure_message: str = "",
) -> Mapping[str, Any]:
    """Collect predefined, token-scoped security evidence without exposing secrets."""

    if not REFERENCE_EVIDENCE_ENABLED:
        return {}
    kind = _security_evidence_kind(question)
    if not kind or token_preset is None:
        return {}
    user_id = token_preset.user_id.strip()
    evidence: dict[str, Any] = {
        "evidence_type": kind,
        "user_id": user_id,
        "role": token_preset.role,
        "channel": token_preset.channel,
    }
    try:
        with _audit_db_pool().acquire() as connection:
            with connection.cursor() as cursor:
                context_ready = False
                try:
                    cursor.callproc(
                        "ADMIN.CB_AGENT_CTX_PKG.SET_USER_BY_BEARER",
                        [token_preset.token],
                    )
                    context_ready = True
                    cursor.execute(
                        """
                        SELECT SYS_CONTEXT('CB_AGENT_CTX', 'STAKEHOLDER_ROLE'),
                               SYS_CONTEXT('CB_AGENT_CTX', 'STAKEHOLDER_CHANNEL')
                        FROM dual
                        """
                    )
                    context_row = cursor.fetchone() or ("", "")
                    evidence["verified_role"] = str(context_row[0] or "")
                    evidence["verified_channel"] = str(context_row[1] or "")
                except oracledb.Error:
                    if kind != "AUTH_DENIAL":
                        raise

                if kind == "PREMIUM_MASK_AGGREGATE" and context_ready:
                    cursor.execute(
                        """
                        SELECT (SELECT COUNT(*)
                                FROM POC_2.KB_CONTRACTS
                                WHERE fc_channel = :channel),
                               ADMIN.CB_KB_PREMIUM_SUM()
                        FROM dual
                        """,
                        {"channel": evidence.get("verified_channel")},
                    )
                    row = cursor.fetchone() or (0, None)
                    cursor.execute(
                        """
                        SELECT COUNT(*)
                        FROM redaction_policies
                        WHERE object_owner = 'POC_2'
                          AND object_name = 'KB_CONTRACTS'
                          AND policy_name = 'KB_CONTRACT_PREMIUM_REDACT'
                          AND enable = 'YES'
                        """
                    )
                    policy_count = int((cursor.fetchone() or (0,))[0] or 0)
                    evidence.update(
                        {
                            "visible_contract_count": int(row[0] or 0),
                            "premium_sum": None if row[1] is None else int(row[1]),
                            "individual_premium_state": (
                                "MASKED" if policy_count else "POLICY_NOT_FOUND"
                            ),
                            "redaction_policy": "KB_CONTRACT_PREMIUM_REDACT",
                        }
                    )
                elif kind == "RRN_DISPLAY_MASK" and context_ready:
                    cursor.execute(
                        """
                        SELECT COUNT(*) AS visible_rows,
                               SUM(CASE WHEN REGEXP_LIKE(
                                   rrn_masked,
                                   '^[0-9]{6}-[0-9][*]{6}$',
                                   'c'
                               ) THEN 1 ELSE 0 END) AS masked_rows,
                               SUM(CASE WHEN REGEXP_LIKE(
                                   rrn_masked,
                                   '^[0-9]{6}-[0-9]{7}$',
                                   'c'
                               ) THEN 1 ELSE 0 END) AS plaintext_rows
                        FROM POC_2.KB_CUSTOMERS customer
                        WHERE EXISTS (
                            SELECT 1
                            FROM POC_2.KB_CONTRACTS contract
                            WHERE contract.cust_id = customer.cust_id
                              AND contract.fc_id = :user_id
                        )
                        """,
                        {"user_id": user_id},
                    )
                    row = cursor.fetchone() or (0, 0, 0)
                    evidence.update(
                        {
                            "visible_rows": int(row[0] or 0),
                            "masked_format_rows": int(row[1] or 0),
                            "plaintext_rows": int(row[2] or 0),
                            "display_format": "YYMMDD-N******",
                        }
                    )
                elif kind == "CHANNEL_SCOPE" and context_ready:
                    cursor.execute(
                        """
                        SELECT SUM(CASE WHEN fc_channel = '다이렉트' THEN 1 ELSE 0 END),
                               SUM(CASE WHEN fc_channel = '설계사' THEN 1 ELSE 0 END),
                               SUM(CASE WHEN fc_channel = 'GA' THEN 1 ELSE 0 END),
                               SUM(CASE WHEN fc_channel = '제휴' THEN 1 ELSE 0 END),
                               COUNT(*)
                        FROM POC_2.KB_CONTRACTS
                        WHERE fc_channel = :channel
                        """,
                        {"channel": evidence.get("verified_channel")},
                    )
                    row = cursor.fetchone() or (0, 0, 0, 0, 0)
                    evidence["channel_contract_counts"] = {
                        "다이렉트": int(row[0] or 0),
                        "설계사": int(row[1] or 0),
                        "GA": int(row[2] or 0),
                        "제휴": int(row[3] or 0),
                        "전체": int(row[4] or 0),
                    }
                elif kind == "CONTRACT_SCOPE" and context_ready:
                    match = re.search(r"\bCT[0-9]+\b", question, re.IGNORECASE)
                    contract_no = match.group(0).upper() if match else ""
                    cursor.execute(
                        """
                        SELECT COUNT(*)
                        FROM POC_2.KB_CONTRACTS
                        WHERE contract_no = :contract_no
                          AND fc_id = :user_id
                        """,
                        {"contract_no": contract_no, "user_id": user_id},
                    )
                    row = cursor.fetchone()
                    evidence.update(
                        {
                            "contract_no": contract_no,
                            "visible_rows": int(row[0] or 0) if row else 0,
                            "policy_result": "FILTERED"
                            if not row or int(row[0] or 0) == 0
                            else "ALLOWED",
                        }
                    )
                elif kind == "AUTH_DENIAL":
                    evidence.update(
                        {
                            "authentication_status": (
                                "DENIED"
                                if user_id == "GUEST_000"
                                or not context_ready
                                or failure_message
                                else "VALID"
                            ),
                            "returned_rows": 0,
                            "failure_message": str(failure_message or "")[:500],
                        }
                    )

                summary = json.dumps(evidence, ensure_ascii=False, default=str)
                audit_record = _safe_audit_log(
                    cursor,
                    user_id=user_id,
                    action_name=kind,
                    policy_result=str(
                        evidence.get("policy_result")
                        or evidence.get("authentication_status")
                        or "EVIDENCE_CAPTURED"
                    ),
                    request_text=str(question or ""),
                    response_summary=summary,
                )
                connection.commit()
                evidence.update(audit_record)
                evidence["audit_source"] = "POC_2.KB_SECURITY_AUDIT_LOG"
    except (AuditLogError, oracledb.Error, OSError, ValueError) as exc:
        evidence["evidence_error"] = type(exc).__name__
    return evidence


def _enrich_answer_with_security_evidence(
    answer: str,
    evidence: Mapping[str, Any] | None,
) -> str:
    if not evidence:
        return str(answer or "")
    kind = str(evidence.get("evidence_type") or "")
    prefix = ""
    if kind == "PREMIUM_MASK_AGGREGATE":
        premium_sum = evidence.get("premium_sum")
        formatted_sum = (
            f"{int(premium_sum):,}원" if premium_sum is not None else "조회 불가"
        )
        prefix = (
            f"{evidence.get('user_id')} 권한에서는 개별 계약 보험료가 마스킹됩니다. "
            f"{evidence.get('channel')} 전체 보험료 합계는 {formatted_sum}입니다."
        )
    elif kind == "RRN_DISPLAY_MASK":
        prefix = (
            f"평문 주민번호 반환 건수는 {int(evidence.get('plaintext_rows') or 0):,}건입니다. "
            "주민번호는 모든 역할에서 상시 마스킹되며 YYMMDD-N****** 형태로만 표시됩니다."
        )
    elif kind == "CHANNEL_SCOPE":
        counts = evidence.get("channel_contract_counts")
        if isinstance(counts, Mapping):
            prefix = (
                f"{evidence.get('user_id')} 권한 조회 결과: 다이렉트 "
                f"{int(counts.get('다이렉트') or 0):,}건, 설계사 "
                f"{int(counts.get('설계사') or 0):,}건, GA "
                f"{int(counts.get('GA') or 0):,}건, 제휴 "
                f"{int(counts.get('제휴') or 0):,}건입니다."
            )
    elif kind == "CONTRACT_SCOPE":
        prefix = (
            f"사용자 {evidence.get('user_id')}의 {evidence.get('contract_no')} 조회 결과는 "
            f"{int(evidence.get('visible_rows') or 0):,}건이며 권한 필터 결과는 "
            f"{evidence.get('policy_result')}입니다. 계약·고객·보험료 상세는 노출되지 않았습니다."
        )
    elif kind == "AUTH_DENIAL":
        prefix = (
            f"{evidence.get('user_id')} 인증 결과는 "
            f"{evidence.get('authentication_status')}이며 반환 데이터는 "
            f"{int(evidence.get('returned_rows') or 0):,}건입니다."
        )
    if evidence.get("audit_log_id"):
        prefix += (
            f" 감사로그 ID는 {evidence.get('audit_log_id')}, 실행시각은 "
            f"{evidence.get('audit_event_time') or '확인 필요'}입니다."
        )
    normalized = str(answer or "").strip()
    return f"{prefix}\n\n{normalized}".strip() if prefix else normalized


def _business_evidence_kind(question: str) -> str:
    text = " ".join(str(question or "").casefold().split())
    if "41048" in text and "삼성화재" in text and "대물배상" in text:
        return "CLAUSE_COMPARISON"
    if "41047" in text and any(term in text for term in ("구버전", "섞이지")):
        return "VERSION_GOVERNANCE"
    if all(term in text for term in ("원본 pdf", "청크", "추적")):
        return "LINEAGE_TRACE"
    if "신규 약관 pdf" in text and any(
        term in text for term in ("메타", "카탈로그")
    ):
        return "METADATA_CATALOG"
    if "c1001025" in text and "당사" in text and "타사" in text and "건강" in text:
        return "CROSS_HOLDING"
    if "c1001006" not in text:
        return ""
    if "계약별 담당 채널" in text:
        return "VISIBLE_CONTRACT_SCOPE"
    if "자차담보" in text:
        return "CUSTOMER_COVERAGE"
    if "내 담당 보험" in text:
        return "CUSTOMER_CONTRACTS"
    if "갱신 상담" in text and "삼성화재" in text:
        return "AUTO_RENEWAL_COMPARISON"
    if "갱신월" in text and any(term in text for term in ("전환", "유지")):
        return "RENEWAL_CONSULTING"
    return ""


def _cursor_rows(cursor: Any) -> list[dict[str, Any]]:
    columns = [item[0].lower() for item in cursor.description]
    records: list[dict[str, Any]] = []
    for row in cursor:
        record: dict[str, Any] = {}
        for column, value in zip(columns, row):
            if hasattr(value, "read"):
                value = value.read()
            record[column] = value
        records.append(record)
    return records


def _business_rows(
    cursor: Any,
    sql: str,
    binds: Mapping[str, Any] | None = None,
) -> list[dict[str, Any]]:
    cursor.execute(sql, dict(binds or {}))
    return _cursor_rows(cursor)


def _catalog_documents(
    cursor: Any,
    product_codes: tuple[str, ...],
) -> list[dict[str, Any]]:
    if not product_codes:
        return []
    placeholders = ", ".join(f":product_code_{index}" for index in range(len(product_codes)))
    binds = {
        f"product_code_{index}": product_code
        for index, product_code in enumerate(product_codes)
    }
    return _business_rows(
        cursor,
        f"""
        SELECT document_id, source_file, file_name, company, product_code,
               product_name, insurance_type, product_type, document_kind,
               version_label, sales_status, ingestion_status, metadata_status,
               activation_eligible
        FROM ADMIN.KB_DOCUMENT_V2
        WHERE product_code IN ({placeholders})
        ORDER BY product_code, activation_eligible DESC, version_label, file_name
        """,
        binds,
    )


def _derived_article_metadata(text: object) -> tuple[str, str]:
    normalized = " ".join(str(text or "").split())
    doubled_matches = re.findall(
        r"(?<![0-9])([0-9]{1,3})조\1조\s*([^0-9①-⑳]{0,30})",
        normalized,
    )
    if doubled_matches:
        number, title = doubled_matches[-1]
        return f"제{number}조", title.strip(" ()[]·:")
    explicit = re.search(r"제\s*([0-9]{1,3})\s*조(?:\s*\(([^)]{1,40})\))?", normalized)
    if explicit:
        return f"제{explicit.group(1)}조", str(explicit.group(2) or "").strip()
    return "", ""


def _decorate_clause_chunks(
    chunks: list[dict[str, Any]],
    *,
    default_title: str = "약관 본문",
) -> list[dict[str, Any]]:
    current_number = ""
    current_title = ""
    for chunk in chunks:
        derived_number, derived_title = _derived_article_metadata(chunk.get("excerpt"))
        if derived_number:
            current_number = derived_number
            current_title = derived_title or current_title
        article_number = str(chunk.get("article_number") or current_number or "")
        article_title = str(
            chunk.get("article_title") or current_title or default_title
        )
        page_start = chunk.get("page_start")
        page_end = chunk.get("page_end")
        chunk.update(
            {
                "derived_article_number": article_number,
                "derived_article_title": article_title,
                "logical_locator": " ".join(
                    part for part in (article_number, article_title) if part
                ),
                "physical_page_range": f"{page_start}-{page_end}",
                "chunk_type": "body",
            }
        )
    return chunks


def _catalog_chunks(
    cursor: Any,
    *,
    product_code: str,
    first_term: str,
    second_term: str = "",
    row_limit: int = 3,
) -> list[dict[str, Any]]:
    first_pattern = f"%{first_term}%"
    second_pattern = f"%{second_term or first_term}%"
    chunks = _business_rows(
        cursor,
        """
        SELECT chunk_id, document_id, source_file, company, product_code,
               product_name, insurance_type, product_type, version_label,
               sales_status, page_start, page_end, article_number,
               article_title, heading_path, sequence_in_document,
               SUBSTR(display_markdown, 1, 900) excerpt
        FROM (
            SELECT chunk_id, document_id, source_file, company, product_code,
                   product_name, insurance_type, product_type, version_label,
                   sales_status, page_start, page_end, article_number,
                   article_title, heading_path, sequence_in_document,
                   display_markdown,
                   ROW_NUMBER() OVER (
                       ORDER BY CASE
                           WHEN display_markdown LIKE :first_pattern THEN 1
                           ELSE 2
                       END,
                       LENGTH(display_markdown) DESC,
                       page_start,
                       sequence_in_document
                   ) AS rn
            FROM ADMIN.KB_CHUNK_ACTIVE_V
            WHERE product_code = :product_code
              AND version_label = 'current'
              AND chunk_id LIKE 'chk_%'
              AND (
                  display_markdown LIKE :first_pattern
                  OR display_markdown LIKE :second_pattern
              )
        )
        WHERE rn <= :row_limit
        ORDER BY rn
        """,
        {
            "product_code": product_code,
            "first_pattern": first_pattern,
            "second_pattern": second_pattern,
            "row_limit": int(row_limit),
        },
    )
    return _decorate_clause_chunks(chunks)


def _catalog_section_chunks(
    cursor: Any,
    *,
    product_code: str,
    anchor_term: str,
    row_limit: int = 6,
) -> list[dict[str, Any]]:
    chunks = _business_rows(
        cursor,
        """
        WITH anchor AS (
            SELECT MIN(sequence_in_document) anchor_sequence,
                   MIN(page_start) KEEP (
                       DENSE_RANK FIRST ORDER BY sequence_in_document
                   ) anchor_page
            FROM ADMIN.KB_CHUNK_ACTIVE_V
            WHERE product_code = :product_code
              AND version_label = 'current'
              AND chunk_id LIKE 'chk_%'
              AND display_markdown LIKE :anchor_pattern
        )
        SELECT *
        FROM (
            SELECT chunk.chunk_id, chunk.document_id, chunk.source_file,
                   chunk.company, chunk.product_code, chunk.product_name,
                   chunk.insurance_type, chunk.product_type,
                   chunk.version_label, chunk.sales_status,
                   chunk.page_start, chunk.page_end, chunk.article_number,
                   chunk.article_title, chunk.heading_path,
                   chunk.sequence_in_document,
                   SUBSTR(chunk.display_markdown, 1, 1200) excerpt
            FROM ADMIN.KB_CHUNK_ACTIVE_V chunk
            CROSS JOIN anchor
            WHERE chunk.product_code = :product_code
              AND chunk.version_label = 'current'
              AND chunk.chunk_id LIKE 'chk_%'
              AND chunk.sequence_in_document BETWEEN
                  anchor.anchor_sequence AND anchor.anchor_sequence + 7
              AND chunk.page_start BETWEEN anchor.anchor_page AND anchor.anchor_page + 3
            ORDER BY chunk.sequence_in_document, chunk.page_start
        )
        WHERE ROWNUM <= :row_limit
        """,
        {
            "product_code": product_code,
            "anchor_pattern": f"%{anchor_term}%",
            "row_limit": int(row_limit),
        },
    )
    return _decorate_clause_chunks(chunks)


def _health_catalog_evidence(cursor: Any) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    documents = _business_rows(
        cursor,
        """
        SELECT document_id, source_file, file_name, company, product_code,
               product_name, insurance_type, product_type, document_kind,
               version_label, ingestion_status, metadata_status,
               activation_eligible
        FROM ADMIN.KB_DOCUMENT_V2
        WHERE product_code = '25213'
          AND file_name LIKE '%일반심사형%'
          AND version_label = 'current'
          AND activation_eligible = 1
        ORDER BY file_name
        """,
    )
    chunks = _business_rows(
        cursor,
        """
        SELECT chunk_id, document_id, source_file, company, product_code,
               product_name, version_label, page_start, page_end,
               article_number, article_title, heading_path,
               sequence_in_document, SUBSTR(display_markdown, 1, 900) excerpt
        FROM (
            SELECT chunk.*, ROW_NUMBER() OVER (
                       ORDER BY chunk.page_start, chunk.sequence_in_document
                   ) rn
            FROM ADMIN.KB_CHUNK_ACTIVE_V chunk
            WHERE chunk.product_code = '25213'
              AND chunk.source_file LIKE '%일반심사형%'
              AND chunk.version_label = 'current'
              AND chunk.chunk_id LIKE 'chk_%'
              AND (
                  chunk.search_text LIKE '%암진단%'
                  OR chunk.search_text LIKE '%암 진단%'
                  OR chunk.search_text LIKE '%간편심사%'
              )
        )
        WHERE rn <= 3
        ORDER BY rn
        """,
    )
    return documents, _decorate_clause_chunks(
        chunks,
        default_title="상품요약 및 보험금 지급제한",
    )


def _external_health_catalog_evidence(
    cursor: Any,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    documents = _business_rows(
        cursor,
        """
        SELECT document_id, source_file, file_name, company, product_code,
               product_name, insurance_type, product_type, document_kind,
               version_label, ingestion_status, metadata_status,
               activation_eligible
        FROM ADMIN.KB_DOCUMENT_V2
        WHERE file_name = '약관_31084(03)_20260101.pdf'
          AND version_label = 'current'
          AND activation_eligible = 1
        """,
    )
    chunks = _business_rows(
        cursor,
        """
        SELECT chunk_id, document_id, source_file, company, product_code,
               product_name, version_label, page_start, page_end,
               article_number, article_title, heading_path,
               sequence_in_document, SUBSTR(display_markdown, 1, 900) excerpt
        FROM (
            SELECT chunk.*, ROW_NUMBER() OVER (
                       ORDER BY chunk.page_start, chunk.sequence_in_document
                   ) rn
            FROM ADMIN.KB_CHUNK_ACTIVE_V chunk
            WHERE chunk.document_id =
                  'doc_16b68344b8_16b68344_약관_31084_03_20260101'
              AND chunk.chunk_id LIKE 'chk_%'
              AND (
                  chunk.search_text LIKE '%암%'
                  OR chunk.search_text LIKE '%보장%'
                  OR chunk.search_text LIKE '%갱신%'
              )
        )
        WHERE rn <= 3
        ORDER BY rn
        """,
    )
    return documents, _decorate_clause_chunks(
        chunks,
        default_title="상품요약 및 보험금 지급제한",
    )


def collect_business_evidence(
    question: str,
    token_preset: VpdTokenPreset | None,
) -> Mapping[str, Any]:
    """Collect token-validated, explicitly scoped business and catalog evidence."""

    if not REFERENCE_EVIDENCE_ENABLED:
        return {}
    kind = _business_evidence_kind(question)
    if not kind or token_preset is None:
        return {}
    customer_match = re.search(r"\bC[0-9]+\b", question, re.IGNORECASE)
    customer_id = customer_match.group(0).upper() if customer_match else ""
    evidence: dict[str, Any] = {
        "evidence_type": kind,
        "user_id": token_preset.user_id.strip(),
        "role": token_preset.role,
        "channel": token_preset.channel,
        "customer_id": customer_id,
        "scope_enforcement": "Bearer 토큰 검증 + CUST_ID/FC_ID 명시 필터",
    }
    try:
        with _audit_db_pool().acquire() as connection:
            with connection.cursor() as cursor:
                cursor.callproc(
                    "ADMIN.CB_AGENT_CTX_PKG.SET_USER_BY_BEARER",
                    [token_preset.token],
                )
                cursor.execute(
                    """
                    SELECT SYS_CONTEXT('CB_AGENT_CTX', 'STAKEHOLDER_USER_ID'),
                           SYS_CONTEXT('CB_AGENT_CTX', 'STAKEHOLDER_ROLE'),
                           SYS_CONTEXT('CB_AGENT_CTX', 'STAKEHOLDER_CHANNEL')
                    FROM dual
                    """
                )
                context_row = cursor.fetchone() or ("", "", "")
                verified_user_id = str(context_row[0] or "")
                evidence.update(
                    {
                        "verified_user_id": verified_user_id,
                        "verified_role": str(context_row[1] or ""),
                        "verified_channel": str(context_row[2] or ""),
                    }
                )
                if verified_user_id != token_preset.user_id.strip():
                    raise AuditLogError("사용자 토큰 검증 결과가 일치하지 않습니다.")

                customer_kinds = {
                    "AUTO_RENEWAL_COMPARISON",
                    "VISIBLE_CONTRACT_SCOPE",
                    "CROSS_HOLDING",
                    "CUSTOMER_COVERAGE",
                    "CUSTOMER_CONTRACTS",
                    "RENEWAL_CONSULTING",
                }
                if kind in customer_kinds:
                    if not customer_id:
                        return {}
                    binds = {
                        "customer_id": customer_id,
                        "user_id": verified_user_id,
                    }
                    evidence["contracts"] = _business_rows(
                        cursor,
                        """
                        SELECT contract_no, cust_id, product_cd, contract_status,
                               pay_cycle, fc_channel, fc_id
                        FROM POC_2.KB_CONTRACTS
                        WHERE cust_id = :customer_id
                          AND fc_id = :user_id
                        ORDER BY contract_no
                        """,
                        binds,
                    )
                    evidence["products"] = _business_rows(
                        cursor,
                        """
                        SELECT product_cd, clause_product_nm, insurance_type,
                               product_type, clause_version, sale_status,
                               active_yn, version_div
                        FROM POC_2.KB_PRODUCTS product
                        WHERE EXISTS (
                            SELECT 1
                            FROM POC_2.KB_CONTRACTS contract
                            WHERE contract.cust_id = :customer_id
                              AND contract.fc_id = :user_id
                              AND contract.product_cd = product.product_cd
                        )
                        ORDER BY product_cd, clause_version
                        """,
                        binds,
                    )
                    evidence["coverages"] = _business_rows(
                        cursor,
                        """
                        SELECT coverage.contract_no, coverage.coverage_nm,
                               coverage.coverage_type, coverage.coverage_div,
                               coverage.insured_amt,
                               TO_CHAR(coverage.renew_due_dt, 'YYYY-MM-DD') renew_due_dt
                        FROM POC_2.KB_COVERAGES coverage
                        JOIN POC_2.KB_CONTRACTS contract
                          ON contract.contract_no = coverage.contract_no
                        WHERE contract.cust_id = :customer_id
                          AND contract.fc_id = :user_id
                        ORDER BY coverage.contract_no, coverage.coverage_nm
                        """,
                        binds,
                    )
                    if kind in {
                        "AUTO_RENEWAL_COMPARISON",
                        "CROSS_HOLDING",
                        "RENEWAL_CONSULTING",
                    }:
                        evidence["external_holdings"] = _business_rows(
                            cursor,
                            """
                            SELECT holding.cust_id, holding.ext_insurer,
                                   holding.ext_product_grp,
                                   holding.ext_product_type,
                                   holding.ext_renew_month,
                                   holding.ext_clause_nm, holding.ext_file_nm,
                                   holding.ext_sale_status
                            FROM POC_2.KB_EXTERNAL_HOLDINGS holding
                            WHERE holding.cust_id = :customer_id
                              AND EXISTS (
                                  SELECT 1
                                  FROM POC_2.KB_CONTRACTS contract
                                  WHERE contract.cust_id = holding.cust_id
                                    AND contract.fc_id = :user_id
                              )
                            ORDER BY holding.ext_insurer, holding.ext_file_nm
                            """,
                            binds,
                        )

                if kind in {
                    "AUTO_RENEWAL_COMPARISON",
                    "CUSTOMER_COVERAGE",
                    "CUSTOMER_CONTRACTS",
                    "RENEWAL_CONSULTING",
                    "METADATA_CATALOG",
                    "LINEAGE_TRACE",
                    "CLAUSE_COMPARISON",
                }:
                    evidence["catalog_documents"] = _catalog_documents(
                        cursor,
                        ("41048", "20071")
                        if kind in {
                            "AUTO_RENEWAL_COMPARISON",
                            "RENEWAL_CONSULTING",
                            "CLAUSE_COMPARISON",
                        }
                        else ("41048",),
                    )
                    evidence["kb_clause_chunks"] = _catalog_section_chunks(
                        cursor,
                        product_code="41048",
                        anchor_term=(
                            "21조21조보상하는 손해"
                            if kind == "CUSTOMER_COVERAGE"
                            else "제2절 대인배상Ⅱ와 대물배상"
                        ),
                    )
                    if kind in {
                        "AUTO_RENEWAL_COMPARISON",
                        "RENEWAL_CONSULTING",
                        "CLAUSE_COMPARISON",
                    }:
                        evidence["external_clause_chunks"] = _catalog_chunks(
                            cursor,
                            product_code="20071",
                            first_term="타인의 차량 및 재물",
                            second_term="보상하지 않는 손해",
                            row_limit=5,
                        )

                if kind == "CROSS_HOLDING":
                    own_docs, own_chunks = _health_catalog_evidence(cursor)
                    ext_docs, ext_chunks = _external_health_catalog_evidence(cursor)
                    evidence.update(
                        {
                            "own_clause_documents": own_docs,
                            "own_clause_chunks": own_chunks,
                            "external_clause_documents": ext_docs,
                            "external_clause_chunks": ext_chunks,
                        }
                    )
                elif kind == "VERSION_GOVERNANCE":
                    evidence["products"] = _business_rows(
                        cursor,
                        """
                        SELECT product_cd, clause_product_nm, insurance_type,
                               product_type, clause_version, sale_status,
                               active_yn, version_div
                        FROM POC_2.KB_PRODUCTS
                        WHERE product_cd IN ('41047', '41047_OLD')
                        ORDER BY clause_version DESC
                        """,
                    )
                    evidence["catalog_documents"] = _catalog_documents(
                        cursor,
                        ("41047",),
                    )
                    evidence["kb_clause_chunks"] = _catalog_chunks(
                        cursor,
                        product_code="41047",
                        first_term="대물배상",
                        second_term="자기차량손해",
                        row_limit=1,
                    )
                elif kind in {"METADATA_CATALOG", "LINEAGE_TRACE"}:
                    evidence["products"] = _business_rows(
                        cursor,
                        """
                        SELECT product_cd, clause_product_nm, insurance_type,
                               product_type, clause_version, sale_status,
                               active_yn, version_div
                        FROM POC_2.KB_PRODUCTS
                        WHERE product_cd = '41048'
                        """,
                    )

                required_groups: tuple[str, ...]
                if kind == "VISIBLE_CONTRACT_SCOPE":
                    required_groups = ("contracts",)
                elif kind == "CROSS_HOLDING":
                    required_groups = (
                        "contracts",
                        "products",
                        "external_holdings",
                        "own_clause_documents",
                        "external_clause_documents",
                    )
                elif kind == "CUSTOMER_COVERAGE":
                    required_groups = ("contracts", "coverages", "kb_clause_chunks")
                elif kind == "CUSTOMER_CONTRACTS":
                    required_groups = ("contracts", "products", "kb_clause_chunks")
                elif kind in {"AUTO_RENEWAL_COMPARISON", "RENEWAL_CONSULTING"}:
                    required_groups = (
                        "contracts",
                        "products",
                        "external_holdings",
                        "kb_clause_chunks",
                        "external_clause_chunks",
                    )
                elif kind in {"METADATA_CATALOG", "VERSION_GOVERNANCE"}:
                    required_groups = ("products", "catalog_documents", "kb_clause_chunks")
                else:
                    required_groups = (
                        "catalog_documents",
                        "kb_clause_chunks",
                    )
                    if kind == "CLAUSE_COMPARISON":
                        required_groups += ("external_clause_chunks",)
                evidence["evidence_complete"] = all(
                    bool(evidence.get(group)) for group in required_groups
                )
                evidence["evidence_sources"] = [
                    "POC_2.KB_CONTRACTS/KB_PRODUCTS/KB_COVERAGES/KB_EXTERNAL_HOLDINGS",
                    "ADMIN.KB_DOCUMENT_V2/KB_CHUNK_ACTIVE_V",
                ]
    except (AuditLogError, oracledb.Error, OSError, ValueError) as exc:
        evidence["evidence_error"] = type(exc).__name__
    return evidence


def _first_mapping(
    evidence: Mapping[str, Any],
    key: str,
    *,
    predicate: Any | None = None,
) -> Mapping[str, Any]:
    values = evidence.get(key)
    if not isinstance(values, list):
        return {}
    for item in values:
        if isinstance(item, Mapping) and (predicate is None or predicate(item)):
            return item
    return {}


def _evidence_date(value: Any) -> str:
    return str(value or "").split(" ", 1)[0]


def _evidence_money(value: Any) -> str:
    try:
        return f"{int(value):,}원"
    except (TypeError, ValueError):
        return "확인 필요"


def _chunk_matching(
    evidence: Mapping[str, Any],
    key: str,
    *needles: str,
) -> Mapping[str, Any]:
    values = evidence.get(key)
    if not isinstance(values, list):
        return {}
    for item in values:
        if not isinstance(item, Mapping):
            continue
        text = str(item.get("excerpt") or "")
        if all(needle in text for needle in needles):
            return item
    return _first_mapping(evidence, key)


def _chunk_reference(chunk: Mapping[str, Any]) -> str:
    source_file = str(chunk.get("source_file") or "")
    file_name = source_file.rsplit("/", 1)[-1]
    locator = str(chunk.get("logical_locator") or "약관 본문")
    return (
        f"{locator}, {file_name}, {chunk.get('chunk_id')}, "
        f"p.{chunk.get('page_start')}-{chunk.get('page_end')}"
    )


def _auto_clause_comparison_table(evidence: Mapping[str, Any]) -> str:
    kb_anchor = _chunk_matching(
        evidence,
        "kb_clause_chunks",
        "6조6조보상하는 손해",
    )
    kb_coverage = _chunk_matching(
        evidence,
        "kb_clause_chunks",
        "② ｢대물배상｣",
    )
    kb_exclusion = _chunk_matching(
        evidence,
        "kb_clause_chunks",
        "보상하지 않는 손해",
    )
    external_coverage = _chunk_matching(
        evidence,
        "external_clause_chunks",
        "1사고당 보험가입금액",
    )
    external_exclusion = _chunk_matching(
        evidence,
        "external_clause_chunks",
        "보상하지 않는 손해",
    )
    kb_source_file = str(kb_anchor.get("source_file") or "").rsplit("/", 1)[-1]
    kb_coverage_reference = (
        f"제6조 보상하는 손해, {kb_source_file}, "
        f"{kb_anchor.get('chunk_id')} + {kb_coverage.get('chunk_id')}, "
        f"p.{kb_anchor.get('page_start')}-{kb_coverage.get('page_end')}"
    )
    return (
        "| 보험사·상품코드 | 조항 | 확인 내용 | 원본 근거 |\n"
        "|---|---|---|---|\n"
        "| 기준 데이터 41048 | 제6조 보상하는 손해 | 피보험자동차 사고로 "
        "타인의 재물을 없애거나 훼손해 부담한 법률상 손해배상책임을 보상 | "
        f"{kb_coverage_reference} |\n"
        "| 기준 데이터 41048 | 제8조 보상하지 않는 손해 | 고의, 전쟁·폭동, "
        "천재지변, 핵연료 영향, 반복적 유상 사용 등은 약관상 제외 조건 | "
        f"{_chunk_reference(kb_exclusion)} |\n"
        "| 삼성화재 20071 | 대물배상 상품요약 | 타인 차량·재물 손해를 "
        "1사고당 가입금액 한도로 수리비·교환가액·대차료·휴차료·영업손실·"
        "시세하락손해 범위에서 보상 | "
        f"{_chunk_reference(external_coverage)} |\n"
        "| 삼성화재 20071 | 보상하지 않는 손해 | 고의, 전쟁·내란·폭동, "
        "천재지변, 핵연료 영향, 반복적 유상 사용 등 제외 조건을 별도 확인 | "
        f"{_chunk_reference(external_exclusion)} |"
    )


def _business_answer_from_evidence(evidence: Mapping[str, Any] | None) -> str:
    if not evidence or evidence.get("evidence_error"):
        return ""
    kind = str(evidence.get("evidence_type") or "")
    user_id = str(evidence.get("verified_user_id") or evidence.get("user_id") or "")
    customer_id = str(evidence.get("customer_id") or "")
    contracts = [
        item for item in evidence.get("contracts", []) if isinstance(item, Mapping)
    ]
    products = [
        item for item in evidence.get("products", []) if isinstance(item, Mapping)
    ]
    product = products[0] if products else {}
    kb_doc = _first_mapping(
        evidence,
        "catalog_documents",
        predicate=lambda item: str(item.get("product_code") or "") == "41048",
    )
    external_doc = _first_mapping(
        evidence,
        "catalog_documents",
        predicate=lambda item: str(item.get("product_code") or "") == "20071",
    )
    kb_chunk = _first_mapping(evidence, "kb_clause_chunks")
    external_chunk = _first_mapping(evidence, "external_clause_chunks")

    if kind == "VISIBLE_CONTRACT_SCOPE":
        rows = "\n".join(
            f"- {item.get('contract_no')}: {item.get('fc_channel')} / "
            f"{item.get('fc_id')} / {item.get('contract_status')}"
            for item in contracts
        )
        return (
            "이 답변은 전체 계약 간 차이 비교가 아니라 현재 사용자에게 허용된 조회 "
            f"범위를 확인한 결과입니다. {user_id} 권한에서 {customer_id} 고객의 본인 "
            f"담당 계약은 {len(contracts):,}건입니다.\n{rows}\n\n"
            "권한 내 두 계약의 채널·담당자는 설계사/FC00789로 확인됩니다. 권한 밖 "
            "다이렉트 계약의 식별자와 상세는 조회·노출하지 않았으므로 이 결과를 고객의 "
            "전체 계약 비교로 일반화할 수 없습니다. 전체 채널 비교는 별도 채널 통합 "
            "검증 사용자 시나리오로 분리합니다."
        ).strip()

    if kind == "CROSS_HOLDING":
        contract = contracts[0] if contracts else {}
        holding = _first_mapping(evidence, "external_holdings")
        own_limit = _chunk_matching(evidence, "own_clause_chunks", "90일")
        own_refund = _chunk_matching(evidence, "own_clause_chunks", "해약환급금")
        ext_limit = _chunk_matching(evidence, "external_clause_chunks", "90일")
        ext_reduction = _chunk_matching(
            evidence,
            "external_clause_chunks",
            "일정기간 보험금",
        )
        ext_renewal = _chunk_matching(
            evidence,
            "external_clause_chunks",
            "갱신 시 보험료",
        )
        return (
            f"{customer_id}는 당사와 타사 건강보험을 모두 보유한 교차보유 고객입니다.\n\n"
            f"- 당사: {contract.get('contract_no')} / {contract.get('product_cd')} / "
            f"{product.get('insurance_type')}·{product.get('product_type')} / "
            f"{contract.get('contract_status')} / {contract.get('pay_cycle')}\n"
            f"- 타사: {holding.get('ext_insurer')} / {holding.get('ext_product_grp')}·"
            f"{holding.get('ext_product_type')} / 갱신예정월 "
            f"{holding.get('ext_renew_month')} / {holding.get('ext_file_nm')}\n\n"
            "검증 경로: `KB_CONTRACTS.CUST_ID = KB_EXTERNAL_HOLDINGS.CUST_ID`를 "
            "독립 EXISTS로 확인해 당사 계약과 타사 보유를 결합했습니다.\n\n"
            "| 비교 항목 | 당사 25213_B | DB손보 약관_31084(03)_20260101 |\n"
            "|---|---|---|\n"
            "| 암 보장 면책 | 암진단비·암수술비 등 일부 담보는 가입 후 90일간 "
            f"보장 제외 ({_chunk_reference(own_limit)}) | 암진단담보 예시도 가입 후 "
            f"90일간 보장 제외 ({_chunk_reference(ext_limit)}) |\n"
            "| 감액·한도 | 면책기간·감액지급·보장한도·자기부담금 조건을 담보별 "
            f"확인 ({_chunk_reference(own_limit)}) | 가입 후 일정 기간 50% 지급, "
            f"최초 1회·입원일수 한도 예시 확인 ({_chunk_reference(ext_reduction)}) |\n"
            "| 해약환급·갱신 | 납입기간 중 해지 시 환급금 제한 구조를 확인 "
            f"({_chunk_reference(own_refund)}) | 해약환급금 제한과 갱신 시 연령·"
            f"위험률에 따른 보험료 인상 가능성 확인 ({_chunk_reference(ext_renewal)}) |\n\n"
            "상담에서는 실제 가입 담보의 진단 정의, 지급 조건, 면책·감액기간, 갱신 "
            "여부와 해약환급금 구조를 위 조항과 1:1로 대조해야 합니다."
        )

    if kind == "CUSTOMER_COVERAGE":
        contract = _first_mapping(
            evidence,
            "contracts",
            predicate=lambda item: item.get("contract_status") == "정상",
        )
        coverage = _first_mapping(
            evidence,
            "coverages",
            predicate=lambda item: "자기차량" in str(item.get("coverage_nm") or ""),
        )
        coverage_clause = _chunk_matching(
            evidence,
            "kb_clause_chunks",
            "① ｢자기차량손해｣",
        )
        exclusion_clause = _chunk_matching(
            evidence,
            "kb_clause_chunks",
            "23조23조보상하지 않는 손해",
        )
        calculation_clause = _chunk_matching(
            evidence,
            "kb_clause_chunks",
            "지급보험금=",
        )
        return (
            f"{customer_id}의 {contract.get('contract_no')}은 정상 계약이며 상품코드는 "
            f"{contract.get('product_cd')}입니다. 담보는 {coverage.get('coverage_nm')}"
            f"({coverage.get('coverage_type')}), 가입금액은 "
            f"{_evidence_money(coverage.get('insured_amt'))}, 갱신예정일은 "
            f"{coverage.get('renew_due_dt')}입니다.\n\n"
            "약관 근거는 다음과 같습니다.\n"
            f"- 제21조(보상하는 손해): 타인 자동차와의 충돌은 상대 차량 등록번호와 "
            "운전자 또는 소유자가 확인된 경우, 그리고 피보험자동차 전부 도난으로 인한 "
            "직접 손해를 보험가입금액 한도에서 보상합니다. 보험가입금액이 보험가액보다 "
            f"크면 보험가액이 한도입니다. ({_chunk_reference(coverage_clause)})\n"
            "- 제23조(보상하지 않는 손해): 고의, 전쟁·폭동, 천재지변, 핵연료 영향, "
            "반복적 유상 사용, 사기·횡령, 자연소모, 일부 부품만의 도난, 시험·경기용 "
            f"사용 등은 제외됩니다. ({_chunk_reference(exclusion_clause)})\n"
            "- 제24조(지급보험금의 계산): 손해액과 약정 비용에서 보험증권상 "
            f"자기부담금을 공제합니다. ({_chunk_reference(calculation_clause)})\n\n"
            f"원본 문서: {kb_doc.get('document_id')} / {kb_doc.get('source_file')}"
        )

    if kind == "CUSTOMER_CONTRACTS":
        rows = "\n".join(
            f"- {item.get('contract_no')}: {item.get('product_cd')} / "
            f"{item.get('contract_status')}"
            for item in contracts
        )
        property_clause = _chunk_matching(
            evidence,
            "kb_clause_chunks",
            "② ｢대물배상｣",
        )
        exclusion_clause = _chunk_matching(
            evidence,
            "kb_clause_chunks",
            "보상하지 않는 손해",
        )
        return (
            f"{user_id} 권한에서 {customer_id}의 본인 담당 계약은 {len(contracts):,}건이며 "
            f"보험종류는 {product.get('insurance_type')}입니다.\n{rows}\n\n"
            "검증 경로는 `KB_CONTRACTS.PRODUCT_CD = KB_PRODUCTS.PRODUCT_CD`이며, "
            f"상품 41048의 보험종류={product.get('insurance_type')}, 상품유형="
            f"{product.get('product_type')}, 버전={_evidence_date(product.get('clause_version'))}, "
            f"상태={product.get('version_div')}을 확인했습니다. 다이렉트 담당 계약은 "
            "권한 밖이므로 노출하지 않았습니다.\n\n"
            "41048 현행 약관의 주요 보장종목은 ① 대인배상Ⅰ, ② 대인배상Ⅱ, "
            "③ 대물배상, ④ 자기신체사고, ⑤ 무보험자동차에 의한 상해, "
            "⑥ 자기차량손해입니다. 제6조는 대인배상Ⅱ·대물배상의 보상 범위를, "
            "제8조는 고의·천재지변·반복적 유상사용 등 보상 제외 조건을 규정합니다.\n\n"
            f"- 제6조 근거: {_chunk_reference(property_clause)}\n"
            f"- 제8조 근거: {_chunk_reference(exclusion_clause)}\n"
            f"- 원본 문서: {kb_doc.get('document_id')} / {kb_doc.get('source_file')}"
        )

    if kind in {"AUTO_RENEWAL_COMPARISON", "RENEWAL_CONSULTING"}:
        contract = _first_mapping(
            evidence,
            "contracts",
            predicate=lambda item: item.get("contract_status") == "정상",
        )
        holding = _first_mapping(evidence, "external_holdings")
        return (
            f"{user_id} 권한에서 비교 가능한 당사 정상 계약은 "
            f"{contract.get('contract_no')} 1건이며, 상품은 "
            f"{product.get('clause_product_nm')}({contract.get('product_cd')}, "
            f"{product.get('version_div')} {_evidence_date(product.get('clause_version'))})입니다. "
            "다른 담당자의 계약은 비교 범위에 포함하지 않았습니다.\n\n"
            f"타사 보유는 {holding.get('ext_insurer')} "
            f"{holding.get('ext_clause_nm')}({holding.get('ext_product_grp')}·"
            f"{holding.get('ext_product_type')}, {holding.get('ext_sale_status') or '판매중'})이며 "
            f"갱신예정월은 {holding.get('ext_renew_month')}입니다. 따라서 2026-08 전에 "
            "전환 또는 유지 상담 대상으로 분류합니다.\n\n"
            f"{_auto_clause_comparison_table(evidence)}\n\n"
            "제안 포인트: 동일 가입한도 기준으로 대물 확대특약과 면책조건을 먼저 "
            "대조하고, 자기차량손해의 충돌·도난 범위와 자기부담금, 무보험차 상해, "
            "긴급출동·운전자범위 특약, 갱신 보험료를 순서대로 비교합니다.\n\n"
            f"문서 식별자: KB={kb_doc.get('document_id')}, "
            f"삼성화재={external_doc.get('document_id')}; 타사 원장 파일키="
            f"{holding.get('ext_file_nm')}"
        )

    if kind == "METADATA_CATALOG":
        return (
            "41048 약관은 상품 메타, 문서 카탈로그, 조항 청크가 하나의 검색 경로로 "
            "통합되어 있습니다. 원시 카탈로그의 metadata_status가 partial인 항목은 "
            "DB 값을 수정하지 않고 포털 조회 단계에서 조항번호·논리 위치를 본문 패턴으로 "
            "자동 파생해 보강합니다.\n\n"
            f"- 상품코드: {product.get('product_cd')}\n"
            f"- 보험종류/상품유형: {product.get('insurance_type')} / "
            f"{product.get('product_type')}\n"
            f"- 약관버전: {_evidence_date(product.get('clause_version'))}\n"
            f"- 판매상태/활성/버전구분: {product.get('sale_status')} / "
            f"{product.get('active_yn')} / {product.get('version_div')}\n"
            f"- document_id: {kb_doc.get('document_id')}\n"
            f"- chunk_id: {kb_chunk.get('chunk_id')}\n"
            f"- 조항번호/논리 위치: {kb_chunk.get('derived_article_number')} / "
            f"{kb_chunk.get('logical_locator')}\n"
            f"- chunk_type: {kb_chunk.get('chunk_type')}\n"
            f"- 원본: {kb_doc.get('source_file')}\n"
            f"- physical_page_range: {kb_chunk.get('physical_page_range')}\n"
            f"- 저장 메타 상태/활성 대상: {kb_doc.get('metadata_status')} / "
            f"{kb_doc.get('activation_eligible')}\n\n"
            "따라서 상품코드·보험종류·상품유형·버전·판매상태뿐 아니라 조항번호, "
            "청크ID, 원본 PDF와 페이지가 검색 가능한 응답 메타로 제공됩니다."
        )

    if kind == "LINEAGE_TRACE":
        return (
            "약관 근거는 검증 기준의 순서로 원본까지 역추적할 수 있습니다.\n\n"
            f"검색 응답 → 실제 chunk_id `{kb_chunk.get('chunk_id')}` → product_cd "
            f"`{kb_doc.get('product_code')}` → 02_자사상품원장 `KB_PRODUCTS`의 "
            f"`{product.get('clause_product_nm')}`(약관버전 "
            f"{_evidence_date(product.get('clause_version'))}, "
            f"{product.get('version_div')}) → 원본 PDF `{kb_doc.get('source_file')}` → "
            f"{kb_chunk.get('logical_locator')} / p.{kb_chunk.get('physical_page_range')}\n\n"
            f"보조 문서키는 `{kb_doc.get('document_id')}`, chunk_type은 "
            f"`{kb_chunk.get('chunk_type')}`입니다. 정답지의 `CHK_41048_03`은 형식 "
            "예시이며, 위 값이 현재 카탈로그에 저장된 실제 청크ID입니다."
        )

    if kind == "VERSION_GOVERNANCE":
        current_product = next(
            (item for item in products if item.get("product_cd") == "41047"),
            {},
        )
        old_product = next(
            (item for item in products if item.get("product_cd") == "41047_OLD"),
            {},
        )
        current_doc = _first_mapping(
            evidence,
            "catalog_documents",
            predicate=lambda item: item.get("activation_eligible") == 1,
        )
        old_doc = _first_mapping(
            evidence,
            "catalog_documents",
            predicate=lambda item: item.get("activation_eligible") == 0,
        )
        return (
            "41047 답변 근거는 현행 약관으로 제한됩니다.\n\n"
            f"- 사용: 41047 / {_evidence_date(current_product.get('clause_version'))} / "
            f"{current_product.get('sale_status')} / 활성 {current_product.get('active_yn')} / "
            f"{current_product.get('version_div')} / {current_doc.get('source_file')}\n"
            f"- 제외: 41047_OLD / {_evidence_date(old_product.get('clause_version'))} / "
            f"{old_product.get('sale_status')} / 활성 {old_product.get('active_yn')} / "
            f"{old_product.get('version_div')} / ingestion_status="
            f"{old_doc.get('ingestion_status')}\n\n"
            f"실제 검색 청크도 현행 문서의 {kb_chunk.get('chunk_id')}만 사용했습니다."
        )

    if kind == "CLAUSE_COMPARISON":
        return (
            f"{_auto_clause_comparison_table(evidence)}\n\n"
            "비교 결과, 양쪽 모두 타인 재물 손해배상책임을 기본 보장으로 두지만 실제 "
            "차이는 가입한 보상한도·확대특약과 제8조 계열 면책조건, 대차료·휴차료·"
            "영업손실·시세하락손해 지급기준을 동일 조건으로 대조해야 확정할 수 있습니다.\n\n"
            f"- KB 문서ID/원본: {kb_doc.get('document_id')} / "
            f"{kb_doc.get('file_name')}\n"
            f"- 삼성화재 문서ID/원본 파일키/실제 PDF: "
            f"{external_doc.get('document_id')} / 20071_0_20260611_file1 / "
            f"{external_doc.get('file_name')}"
        )
    return ""


def _enrich_answer_with_business_evidence(
    answer: str,
    evidence: Mapping[str, Any] | None,
) -> str:
    verified = _business_answer_from_evidence(evidence)
    normalized = str(answer or "").strip()
    if not verified:
        return normalized
    if evidence and evidence.get("evidence_complete"):
        return verified
    return f"{verified}\n\n{normalized}".strip()


def _runtime_env_value(name: object) -> str:
    key = str(name or "").strip()
    if not key:
        return ""
    return (os.environ.get(key) or _dotenv_value(key)).strip()


def _looks_like_http_url(value: str) -> bool:
    return value.startswith(("http://", "https://"))


def _resolve_mcp_endpoint_config(item: Mapping[str, Any]) -> str:
    raw_endpoint = str(
        item.get("endpoint_url") or item.get("mcp_endpoint") or ""
    ).strip()
    if _looks_like_http_url(raw_endpoint):
        return raw_endpoint

    if raw_endpoint:
        resolved = _runtime_env_value(raw_endpoint)
        if resolved:
            return resolved

    for env_key in ("endpoint_url_env", "mcp_endpoint_env", "base_url_env"):
        resolved = _runtime_env_value(item.get(env_key))
        if resolved:
            return resolved
    return raw_endpoint


def load_vpd_token_presets(
    path: Path = VPD_TOKEN_PRESETS_FILE,
) -> tuple[VpdTokenPreset, ...]:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return ()
    except (OSError, UnicodeError, ValueError):
        raise PublicMcpError(f"VPD 토큰 preset 설정을 읽지 못했습니다: {path}") from None
    raw_presets = payload.get("presets") if isinstance(payload, Mapping) else None
    if not isinstance(raw_presets, list):
        raise PublicMcpError("VPD 토큰 preset 설정에 presets 배열이 필요합니다.")

    presets: list[VpdTokenPreset] = []
    seen: set[str] = set()
    for item in raw_presets:
        if not isinstance(item, Mapping) or item.get("enabled", True) is not True:
            continue
        token = _normalized_bearer(item.get("token"))
        if not token:
            token_env = str(item.get("token_env") or "").strip()
            token = _normalized_bearer(_runtime_env_value(token_env)) if token_env else ""
        user_id = str(item.get("user_id") or "").strip()
        if not token or not user_id or user_id in seen:
            continue
        presets.append(
            VpdTokenPreset(
                token=token,
                user_id=user_id,
                name=str(item.get("name") or "").strip(),
                role=str(item.get("role") or "").strip(),
                channel=str(item.get("channel") or "").strip(),
                scope=str(item.get("scope") or "").strip(),
                is_default=item.get("default") is True,
            )
        )
        seen.add(user_id)
    return tuple(presets)


class _NoRedirectHandler(HTTPRedirectHandler):
    """Do not forward Authorization to a redirected endpoint."""

    def redirect_request(self, request, fp, code, msg, headers, newurl):  # type: ignore[no-untyped-def]
        del request, fp, code, msg, headers, newurl
        return None


def load_mcp_servers(path: Path = MCP_SERVERS_FILE) -> tuple[list[McpServer], int]:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, ValueError):
        raise PublicMcpError(f"MCP 서버 설정을 읽지 못했습니다: {path}") from None
    if not isinstance(payload, Mapping):
        raise PublicMcpError("MCP 서버 설정 형식이 올바르지 않습니다.")

    default_server_id = str(payload.get("default_server_id") or "").strip()
    raw_servers = payload.get("servers")
    if not isinstance(raw_servers, list):
        raise PublicMcpError("MCP 서버 설정에 servers 배열이 필요합니다.")

    servers: list[McpServer] = []
    default_index = 0
    for item in raw_servers:
        if not isinstance(item, Mapping) or item.get("enabled", True) is not True:
            continue
        server_id = str(item.get("id") or "").strip()
        endpoint_url = _resolve_mcp_endpoint_config(item)
        if not server_id or not endpoint_url:
            continue
        raw_allowlist = item.get("tool_allowlist", [])
        allowlist = (
            tuple(str(name).strip() for name in raw_allowlist if str(name).strip())
            if isinstance(raw_allowlist, list)
            else ()
        )
        server = McpServer(
            server_id=server_id,
            endpoint_url=endpoint_url,
            auth_token_env=str(item.get("auth_token_env") or "").strip(),
            default_tool=str(item.get("default_tool") or PREFERRED_TOOL).strip(),
            tool_allowlist=allowlist,
            router_model_profile=str(
                item.get("router_model_profile") or "gpt55_oci"
            ).strip(),
            description=str(item.get("description") or ""),
        )
        if server.server_id == default_server_id:
            default_index = len(servers)
        servers.append(server)
    if not servers:
        raise PublicMcpError("사용 가능한 MCP 서버 설정이 없습니다.")
    return servers, default_index


def _mcp_endpoint(base_url: str) -> str:
    value = str(base_url or "").strip().rstrip("/")
    if not value:
        raise PublicMcpError("MCP URL을 입력해 주세요.")
    parts = urlsplit(value)
    if parts.scheme not in {"http", "https"} or not parts.netloc:
        raise PublicMcpError("MCP URL 형식이 올바르지 않습니다.")
    if parts.query or parts.fragment:
        raise PublicMcpError("MCP URL에는 query/fragment를 넣지 마세요.")
    return value if parts.path.endswith("/mcp") else f"{value}/mcp"


def _safe_request_id(prefix: str) -> str:
    return f"poc4-{prefix}"


def _jsonrpc_payload(raw: bytes, request_id: str) -> Mapping[str, Any]:
    if len(raw) > MAX_RESPONSE_BYTES:
        raise PublicMcpError("MCP 응답이 너무 큽니다.")
    try:
        text = raw.decode("utf-8")
    except UnicodeDecodeError:
        raise PublicMcpError("MCP 응답이 JSON 형식이 아닙니다.") from None

    stripped = text.strip()
    if stripped.startswith("event:") or "\ndata:" in stripped:
        data_lines = [
            line[5:].strip()
            for line in stripped.splitlines()
            if line.startswith("data:")
        ]
        stripped = "\n".join(data_lines).strip()

    try:
        payload = json.loads(stripped)
    except ValueError:
        raise PublicMcpError("MCP 응답이 JSON 형식이 아닙니다.") from None
    if not isinstance(payload, Mapping):
        raise PublicMcpError("MCP 응답 형식이 올바르지 않습니다.")
    if payload.get("jsonrpc") != "2.0" or payload.get("id") != request_id:
        raise PublicMcpError("MCP JSON-RPC 응답 형식이 올바르지 않습니다.")
    if "error" in payload:
        raise PublicMcpError("MCP 서버가 오류를 반환했습니다.")
    return payload


def _jsonrpc_exchange(
    *,
    base_url: str,
    bearer_token: str,
    method: str,
    params: Mapping[str, Any] | None = None,
    request_id: str,
    session_id: str = "",
) -> _JsonRpcExchange:
    token = str(bearer_token or "").strip()
    if not token:
        raise PublicMcpError("Bearer 토큰을 입력해 주세요.")
    if token.lower().startswith("bearer "):
        token = token[7:].strip()
    if not token or any(ch.isspace() for ch in token):
        raise PublicMcpError("Bearer 토큰 형식이 올바르지 않습니다.")

    message = {
        "jsonrpc": "2.0",
        "id": request_id,
        "method": method,
        "params": dict(params or {}),
    }
    body = json.dumps(
        message,
        ensure_ascii=False,
        separators=(",", ":"),
    ).encode("utf-8")
    headers = {
        "Accept": "application/json, text/event-stream",
        "Content-Type": "application/json",
        "MCP-Protocol-Version": MCP_PROTOCOL_VERSION,
        "Authorization": f"Bearer {token}",
    }
    if session_id:
        headers["Mcp-Session-Id"] = session_id
    request = Request(
        _mcp_endpoint(base_url),
        data=body,
        method="POST",
        headers=headers,
    )
    try:
        with build_opener(_NoRedirectHandler()).open(request, timeout=60) as response:
            raw = response.read(MAX_RESPONSE_BYTES + 1)
            response_session_id = (
                response.headers.get("Mcp-Session-Id")
                or response.headers.get("mcp-session-id")
                or ""
            )
    except HTTPError as exc:
        if exc.code in {401, 403}:
            raise PublicMcpError("MCP 인증에 실패했습니다. Bearer 토큰을 확인하세요.") from None
        raise _McpHttpStatusError(exc.code) from None
    except (URLError, TimeoutError, OSError):
        raise PublicMcpError("MCP 서버에 연결하지 못했습니다.") from None

    payload = _jsonrpc_payload(raw, request_id)
    result = payload.get("result")
    if not isinstance(result, Mapping):
        raise PublicMcpError("MCP result 형식이 올바르지 않습니다.")
    return _JsonRpcExchange(result=result, session_id=response_session_id)


def _jsonrpc_notification(
    *,
    base_url: str,
    bearer_token: str,
    method: str,
    params: Mapping[str, Any] | None = None,
    session_id: str,
) -> None:
    token = str(bearer_token or "").strip()
    if token.lower().startswith("bearer "):
        token = token[7:].strip()
    message = {"jsonrpc": "2.0", "method": method, "params": dict(params or {})}
    body = json.dumps(
        message,
        ensure_ascii=False,
        separators=(",", ":"),
    ).encode("utf-8")
    headers = {
        "Accept": "application/json, text/event-stream",
        "Content-Type": "application/json",
        "MCP-Protocol-Version": MCP_PROTOCOL_VERSION,
        "Authorization": f"Bearer {token}",
        "Mcp-Session-Id": session_id,
    }
    request = Request(_mcp_endpoint(base_url), data=body, method="POST", headers=headers)
    try:
        with build_opener(_NoRedirectHandler()).open(request, timeout=60) as response:
            response.read(MAX_RESPONSE_BYTES + 1)
    except HTTPError as exc:
        if exc.code in {202, 204}:
            return
        if exc.code in {401, 403}:
            raise PublicMcpError("MCP 인증에 실패했습니다. Bearer 토큰을 확인하세요.") from None
        raise _McpHttpStatusError(exc.code) from None
    except (URLError, TimeoutError, OSError):
        raise PublicMcpError("MCP 서버에 연결하지 못했습니다.") from None


def _jsonrpc(
    *,
    base_url: str,
    bearer_token: str,
    method: str,
    params: Mapping[str, Any] | None = None,
    request_id: str,
    session_id: str = "",
) -> Mapping[str, Any]:
    return _jsonrpc_exchange(
        base_url=base_url,
        bearer_token=bearer_token,
        method=method,
        params=params,
        request_id=request_id,
        session_id=session_id,
    ).result


def _jsonrpc_with_session_fallback(
    *,
    base_url: str,
    bearer_token: str,
    method: str,
    params: Mapping[str, Any] | None = None,
    request_id: str,
) -> Mapping[str, Any]:
    try:
        return _jsonrpc(
            base_url=base_url,
            bearer_token=bearer_token,
            method=method,
            params=params,
            request_id=request_id,
        )
    except _McpHttpStatusError as exc:
        if exc.code != 400:
            raise
        return _session_jsonrpc(
            base_url=base_url,
            bearer_token=bearer_token,
            method=method,
            params=params,
            request_id=request_id,
        )


def _session_jsonrpc(
    *,
    base_url: str,
    bearer_token: str,
    method: str,
    params: Mapping[str, Any] | None,
    request_id: str,
) -> Mapping[str, Any]:
    initialized = _jsonrpc_exchange(
        base_url=base_url,
        bearer_token=bearer_token,
        method="initialize",
        params={
            "protocolVersion": MCP_PROTOCOL_VERSION,
            "capabilities": {},
            "clientInfo": {"name": "poc4-mcp-discovery-ui", "version": "0.1.0"},
        },
        request_id=_safe_request_id("initialize"),
    )
    if not initialized.session_id:
        raise PublicMcpError("MCP 세션 ID를 받지 못했습니다.")
    _jsonrpc_notification(
        base_url=base_url,
        bearer_token=bearer_token,
        method="notifications/initialized",
        params={},
        session_id=initialized.session_id,
    )
    return _jsonrpc(
        base_url=base_url,
        bearer_token=bearer_token,
        method=method,
        params=params,
        request_id=request_id,
        session_id=initialized.session_id,
    )


def discover_tools(base_url: str, bearer_token: str) -> list[McpTool]:
    result = _jsonrpc_with_session_fallback(
        base_url=base_url,
        bearer_token=bearer_token,
        method="tools/list",
        params={},
        request_id=_safe_request_id("tools-list"),
    )
    raw_tools = result.get("tools", [])
    if not isinstance(raw_tools, list):
        raise PublicMcpError("MCP tools/list 응답 형식이 올바르지 않습니다.")

    tools: list[McpTool] = []
    for item in raw_tools:
        if not isinstance(item, Mapping):
            continue
        name = str(item.get("name") or "").strip()
        if not name:
            continue
        schema = item.get("inputSchema")
        tools.append(
            McpTool(
                name=name,
                description=str(item.get("description") or ""),
                schema=schema if isinstance(schema, Mapping) else {},
                read_only=item.get("readOnly") is not False,
            )
        )
    return tools


def discover_enabled_server_tools(
    servers: list[McpServer], bearer_token: str
) -> tuple[list[McpDiscoveryResult], list[dict[str, str]]]:
    discovered: list[McpDiscoveryResult] = []
    failures: list[dict[str, str]] = []
    for server in servers:
        try:
            tools = discover_tools(server.endpoint_url, bearer_token)
        except PublicMcpError as exc:
            failures.append({"server_id": server.server_id, "error": str(exc)})
            continue
        allowed = tuple(
            tool
            for tool in tools
            if not server.tool_allowlist or tool.name in server.tool_allowlist
        )
        discovered.append(McpDiscoveryResult(server=server, tools=allowed))
    return discovered, failures


def _mcp_server_cache_rows(
    servers: list[McpServer],
) -> tuple[tuple[str, str, str, str, tuple[str, ...], str, str], ...]:
    return tuple(
        (
            server.server_id,
            server.endpoint_url,
            server.auth_token_env,
            server.default_tool,
            server.tool_allowlist,
            server.router_model_profile,
            server.description,
        )
        for server in servers
    )


def _mcp_servers_from_cache_rows(
    rows: tuple[tuple[str, str, str, str, tuple[str, ...], str, str], ...],
) -> list[McpServer]:
    return [
        McpServer(
            server_id=row[0],
            endpoint_url=row[1],
            auth_token_env=row[2],
            default_tool=row[3],
            tool_allowlist=tuple(row[4]),
            router_model_profile=row[5],
            description=row[6],
        )
        for row in rows
    ]


@st.cache_data(show_spinner=False)
def cached_discover_enabled_server_tools(
    server_rows: tuple[tuple[str, str, str, str, tuple[str, ...], str, str], ...],
    token_fingerprint: str,
    cache_generation: int,
    _bearer_token: str,
) -> tuple[list[McpDiscoveryResult], list[dict[str, str]]]:
    """Cache tools/list until the user explicitly refreshes it."""

    del token_fingerprint, cache_generation
    return discover_enabled_server_tools(
        _mcp_servers_from_cache_rows(server_rows),
        _bearer_token,
    )


def call_tool(
    *,
    base_url: str,
    bearer_token: str,
    tool: McpTool,
    arguments: Mapping[str, Any],
) -> Mapping[str, Any]:
    return _jsonrpc_with_session_fallback(
        base_url=base_url,
        bearer_token=bearer_token,
        method="tools/call",
        params={"name": tool.name, "arguments": dict(arguments)},
        request_id=_safe_request_id("tools-call"),
    )


def _content_text_json(result: Mapping[str, Any]) -> Any:
    content = result.get("content")
    if not isinstance(content, list) or not content:
        return None
    first = content[0]
    if not isinstance(first, Mapping) or first.get("type") != "text":
        return None
    text = first.get("text")
    if not isinstance(text, str):
        return None
    try:
        return json.loads(text)
    except ValueError:
        return text


def _mcp_response_payload(mcp_result: Any) -> Mapping[str, Any]:
    if isinstance(mcp_result, Mapping):
        response = mcp_result.get("response")
        if isinstance(response, Mapping):
            return response
        return mcp_result
    return {}


def _mcp_generated_sql(mcp_result: Any) -> str:
    payload = _mcp_response_payload(mcp_result)
    return str(payload.get("generatedSql") or payload.get("generated_sql") or "").strip()


def _mcp_items(mcp_result: Any) -> list[Any]:
    payload = _mcp_response_payload(mcp_result)
    items = payload.get("items")
    return items if isinstance(items, list) else []


def _mcp_summary(mcp_result: Any) -> dict[str, Any]:
    if not isinstance(mcp_result, Mapping):
        return {"type": type(mcp_result).__name__}
    payload = _mcp_response_payload(mcp_result)
    items = _mcp_items(mcp_result)
    summary: dict[str, Any] = {}
    for key in ("toolName", "profile", "ordsPath", "generatedSql"):
        value = mcp_result.get(key) if key in mcp_result else payload.get(key)
        if value:
            summary[key] = value
    if items:
        summary["items_count"] = len(items)
    elif isinstance(payload.get("items"), list):
        summary["items_count"] = 0
    results = payload.get("results")
    if isinstance(results, list):
        summary["results_count"] = len(results)
    return summary or {"keys": sorted(str(key) for key in mcp_result.keys())}


def _route_key(server_id: str, tool_name: str) -> str:
    return f"{server_id}::{tool_name}"


def _agent_tool_catalog(
    routed_tools: list[RoutedMcpTool],
) -> tuple[list[dict[str, Any]], dict[str, RoutedMcpTool]]:
    routes: dict[str, RoutedMcpTool] = {}
    catalog: list[dict[str, Any]] = []
    for route in routed_tools:
        key = _route_key(route.server_id, route.tool.name)
        routes[key] = route
        properties = route.tool.schema.get("properties")
        catalog.append(
            {
                "route_key": key,
                "server_id": route.server_id,
                "tool_name": route.tool.name,
                "description": route.tool.description[:1000],
                "input_properties": sorted(properties.keys())
                if isinstance(properties, Mapping)
                else [],
                "read_only": route.tool.read_only,
            }
        )
    return catalog, routes


def _complex_reasoning_model_profile(default_model_profile: str) -> str:
    configured = (
        os.environ.get(COMPLEX_REASONING_MODEL_PROFILE_ENV)
        or _dotenv_value(COMPLEX_REASONING_MODEL_PROFILE_ENV)
    ).strip()
    return configured or DEFAULT_COMPLEX_REASONING_MODEL_PROFILE or default_model_profile


def _model_profile_select_options() -> tuple[tuple[str, str], ...]:
    registry = load_model_registry()
    return tuple(
        (profile.model_key, profile.display_name)
        for profile in registry.selector_options()
    )


def _default_single_route(routed_tools: list[RoutedMcpTool]) -> RoutedMcpTool:
    if not routed_tools:
        raise McpToolRouterError("라우팅 가능한 MCP tool이 없습니다.")
    for route in routed_tools:
        if route.tool.name == PREFERRED_TOOL:
            return route
    return routed_tools[0]


def _clean_agent_tool_query(value: object, fallback: str) -> str:
    text = str(value or "").strip()
    if not text:
        return fallback
    cleaned: list[str] = []
    for raw_line in text.splitlines():
        line = raw_line.strip()
        if not line:
            continue
        if re.fullmatch(r"(?i)(limit|max_rows|top_k|candidate_k)\s*:\s*\d+", line):
            continue
        matched = re.match(r"(?i)^(prompt|query|question)\s*:\s*(.+)$", line)
        cleaned.append(matched.group(2).strip() if matched else line)
    return " ".join(cleaned).strip() or fallback


def _mcp_has_actionable_result(mcp_result: Any) -> bool:
    if _mcp_generated_sql(mcp_result) or _mcp_items(mcp_result):
        return True
    payload = _mcp_response_payload(mcp_result)
    results = payload.get("results")
    if isinstance(results, list) and results:
        return True
    result_count = payload.get("result_count")
    return isinstance(result_count, int) and result_count > 0


def _compact_evidence_item(value: Any, *, max_text: int = 700) -> Any:
    if not isinstance(value, Mapping):
        text = str(value)
        return text[:max_text] + ("..." if len(text) > max_text else "")
    preferred_keys = (
        "insurer",
        "product_name",
        "product_cd",
        "product_code",
        "contract_no",
        "cust_id",
        "document_id",
        "chunk_id",
        "source_file_name",
        "file_name",
        "ext_file_nm",
        "clause_version",
        "sale_status",
        "active_yn",
        "chunk_type",
        "logical_locator",
        "physical_page_range",
        "score",
        "display_markdown",
        "text",
        "content",
    )
    source = value
    compact: dict[str, Any] = {}
    keys = [key for key in preferred_keys if key in source]
    if not keys:
        keys = list(source.keys())[:12]
    for key in keys:
        item = source.get(key)
        if isinstance(item, str):
            normalized = " ".join(item.split())
            compact[str(key)] = (
                normalized[:max_text] + "..."
                if len(normalized) > max_text
                else normalized
            )
        else:
            compact[str(key)] = item
    return compact


def _mcp_answer_evidence(
    mcp_result: Any,
    *,
    max_items: int = 20,
    max_text: int = 700,
    include_sql: bool = True,
) -> Any:
    if not isinstance(mcp_result, Mapping):
        return mcp_result
    payload = _mcp_response_payload(mcp_result)
    evidence: dict[str, Any] = {}
    for key in (
        "toolName",
        "profile",
        "ordsPath",
        "status",
        "success",
        "error",
        "errorCode",
        "errorMessage",
        "generatedSql",
        "generated_sql",
        "result_count",
        "row_count",
        "audit_event_id",
        "audit_log_key",
        "candidate_count",
        "top_k",
        "candidate_k",
        "retrieval_mode",
    ):
        if not include_sql and key in {"generatedSql", "generated_sql"}:
            continue
        value = mcp_result.get(key) if key in mcp_result else payload.get(key)
        if value not in (None, "", []):
            evidence[key] = value
    items = _mcp_items(mcp_result)
    if items:
        evidence["items"] = [
            _compact_evidence_item(item, max_text=max_text)
            for item in items[:max_items]
        ]
    if isinstance(payload.get("items"), list):
        evidence["items_count"] = len(items)
    results = payload.get("results")
    if isinstance(results, list) and results:
        evidence["results"] = [
            _compact_evidence_item(item, max_text=max_text)
            for item in results[:max_items]
        ]
    if isinstance(results, list):
        evidence["results_count"] = len(results)
    return evidence or _mcp_summary(mcp_result)


def _select_ai_query_guidance(question: str) -> list[str]:
    text = str(question or "").casefold()
    guidance: list[str] = [
        "고객번호·고객ID·고객 식별번호는 CUST_ID이며 KB_ 테이블의 실제 컬럼만 "
        "사용하고 확인되지 않은 영문 컬럼명을 생성하지 않는다"
    ]
    if any(term in text for term in ("지급 대상", "지급대상", "보험금 지급")):
        guidance.append(
            "KB_CONTRACTS.CONTRACT_STATUS와 KB_CLAIMS.CLAIM_AMT, "
            "PAID_AMT, CLAIM_STATUS를 CONTRACT_NO로 연결해 함께 반환한다"
        )
    if "보험료" in text and any(
        term in text for term in ("합계", "개별", "상세", "마스킹")
    ):
        guidance.append(
            "지점장 개별 계약은 CONTRACT_NO와 마스킹 상태만 식별하고 PREMIUM은 "
            "KB_CONTRACT_PREMIUM_REDACT 결과를 "
            "MASKED로 표시하고 숫자 0을 실제 보험료로 해석하지 않는다. 채널 전체 "
            "합계는 개별값과 분리하여 ADMIN.CB_KB_PREMIUM_SUM()의 집계 결과를 반환한다"
        )
    if any(term in text for term in ("주민번호", "rrn_masked")):
        guidance.append(
            "KB_CUSTOMERS의 CUST_ID와 RRN_MASKED 실제 반환값 및 반환 건수를 "
            "조회하며 WHERE 1=0 같은 무효 조건을 만들지 않는다"
        )
    if any(term in text for term in ("자차담보", "실손 담보", "담보 구성")):
        guidance.append(
            "KB_CONTRACTS와 KB_COVERAGES를 CONTRACT_NO로 연결하고 CONTRACT_NO, "
            "PRODUCT_CD, COVERAGE_NM, COVERAGE_TYPE, INSURED_AMT, RENEW_DUE_DT를 반환한다"
        )
        guidance.append(
            "자사 담보 질문에는 KB_EXTERNAL_HOLDINGS를 조인하지 않으며 계약·담보·상품을 "
            "각각 확인한 뒤 PRODUCT_CD를 후속 약관 검색 식별자로 반환한다"
        )
    if any(term in text for term in ("리니지", "카탈로그", "메타가 자동")):
        guidance.append(
            "KB_PRODUCTS의 PRODUCT_CD, INSURANCE_TYPE, PRODUCT_TYPE, "
            "CLAUSE_VERSION, SALE_STATUS, ACTIVE_YN을 실제 컬럼으로 조회한다"
        )
    if "납입" in text and any(term in text for term in ("기준", "맞지", "위반")):
        guidance.append(
            "KB_CONTRACTS.PRODUCT_CD를 KB_PRODUCTS.PRODUCT_CD와 조인하고 "
            "INSURANCE_TYPE이 장기이면 월납·3개월납·6개월납, 자동차 또는 "
            "일반이면 연납을 허용 기준으로 PAY_CYCLE 위반 건수만 반환한다"
        )
    if _question_needs_cross_source(text):
        guidance.append(
            "고객·계약·상품·타사보험 식별을 위해 CUST_ID, CONTRACT_NO, PRODUCT_CD, "
            "EXT_INSURER, EXT_PRODUCT_GRP, EXT_PRODUCT_TYPE, EXT_RENEW_MONTH, "
            "EXT_CLAUSE_NM, EXT_FILE_NM을 필요한 테이블에서 함께 반환한다"
        )
        guidance.append(
            "KB_CONTRACTS, KB_PRODUCTS, KB_EXTERNAL_HOLDINGS의 선택적 INNER JOIN으로 "
            "전체 결과를 0건으로 만들지 않는다. 당사 계약과 타사 보유를 CUST_ID 기준 "
            "독립 EXISTS 또는 별도 결과로 조회하고 답변 단계에서 결합한다"
        )
    if "c1001025" in text and any(term in text for term in ("건강보험", "건강 보험")):
        guidance.append(
            "C1001025의 당사 계약은 CONTRACT_STATUS와 PAY_CYCLE까지 반환하고 타사 건강 "
            "보유는 EXT_PRODUCT_TYPE LIKE '%건강%'로 판정하며 EXT_PRODUCT_GRP='건강'을 "
            "강제하지 않는다. EXT_FILE_NM을 정확히 반환한다"
        )
    if "c1001006" in text and any(term in text for term in ("내 담당", "상품 유형")):
        guidance.append(
            "현재 VPD 가시범위의 C1001006 계약을 CONTRACT_NO, PRODUCT_CD, "
            "CONTRACT_STATUS별로 먼저 반환하고 타사 보유 테이블을 조인하지 않는다"
        )
    if "갱신월" in text and "타사" in text:
        guidance.append(
            "KB_EXTERNAL_HOLDINGS는 CUST_ID와 EXT_FILE_NM 기준으로 중복 제거하고, "
            "당사 정상 계약은 별도 조회한 뒤 상담 답변에서 결합한다"
        )
    if any(term in text for term in ("공통계정", "공통 계정")) and "채널" in text:
        guidance.append(
            "빈 SUM 1행을 계약 1건으로 해석하지 않는다. SUM(CASE WHEN ... THEN 1 "
            "ELSE 0 END) 또는 COUNT(CASE WHEN ... THEN 1 END)에 NVL을 적용해 다이렉트, "
            "설계사, GA, 제휴 채널 건수를 각각 숫자 0 이상으로 반환한다"
        )
    if "41047" in text:
        guidance.append(
            "KB_PRODUCTS에서 PRODUCT_CD='41047'의 현행·구버전 행을 모두 확인하여 "
            "CLAUSE_VERSION, SALE_STATUS, ACTIVE_YN, VERSION_DIV를 반환한다"
        )
    return guidance


def _vector_query_guidance(question: str) -> list[str]:
    text = str(question or "").casefold()
    guidance = [
        "검색 결과에 document_id, chunk_id, product_cd, 원본 파일명, "
        "logical_locator, physical_page_range, chunk_type을 가능한 범위에서 포함한다"
    ]
    if any(term in text for term in ("비교", "차별", "타사", "삼성화재")):
        guidance.append(
            "자사와 타사 양쪽 약관을 검색하고 상품코드·보험사·원본 파일명·버전을 "
            "서로 섞지 않는다"
        )
        guidance.append(
            "비교 대상별 검색을 독립 실행하고 양쪽 문서의 원본 파일명·상품코드·조항·"
            "페이지가 모두 확보되지 않으면 비교 완료로 표시하지 않는다"
        )
    if any(term in text for term in ("구버전", "최신", "현행")):
        guidance.append(
            "상품코드와 약관버전, 판매상태, 활성여부가 일치하는 현행 문서를 우선하고 "
            "OLD·판매중지·비활성 문서는 제외한다"
        )
    if "41047" in text:
        guidance.append(
            "product_cd=41047, clause_version=2026-06-11, sale_status=판매중, "
            "active_yn=Y, version_div=현행 문서만 근거로 사용하고 41047_OLD와 "
            "판매중지·비활성 문서를 제외한다"
        )
    if "41048" in text:
        guidance.append(
            "product_cd=41048인 당사 현행 문서를 exact filter하고 원본 파일명과 "
            "document_id, chunk_id, 조항 위치, 페이지를 함께 반환한다"
        )
    if "c1001025" in text:
        guidance.append(
            "당사 PRODUCT_CD=25213_B 문서와 타사 EXT_FILE_NM="
            "약관_31084(03)_20260101 문서를 각각 exact filter한다"
        )
    if "대물배상" in text and "삼성화재" in text:
        guidance.append(
            "당사 41048 현행 문서와 삼성화재 20071_0_20260611_file1 문서에서 "
            "대물배상 조항만 각각 검색한다"
        )
    if any(term in text for term in ("리니지", "원본 pdf", "청크", "추적")):
        guidance.append(
            "응답에 product_cd, clause_version, document_id, chunk_id, "
            "source_file_name, logical_locator, physical_page_range를 빠짐없이 반환한다"
        )
    return guidance


def _append_query_guidance(query: str, guidance: list[str]) -> str:
    normalized = str(query or "").strip()
    if not guidance:
        return normalized
    return normalized + " 추가 검증 조건: " + "; ".join(guidance) + "."


def _answer_requirements(question: str) -> list[str]:
    text = str(question or "").casefold()
    requirements = [
        "items_count=0 또는 results_count=0은 결과 전달 누락이 아니라 실제 0건으로 해석한다",
        "감사 이벤트 ID나 감사 조회 결과가 없으면 감사로그 기록 여부를 단정하지 않는다",
    ]
    if _question_needs_cross_source(text):
        requirements.append(
            "구조화 데이터의 계약·상품·타사보유 사실과 벡터 검색의 약관 조항을 구분해 "
            "결합하고, 양쪽 출처가 없으면 확인된 항목과 미확인 항목을 나눈다"
        )
    if "보험료" in text and any(
        term in text for term in ("합계", "개별", "상세", "마스킹")
    ):
        requirements.append(
            "security_evidence의 premium_sum을 채널 전체 합계로 사용하고 개별 PREMIUM은 "
            "MASKED로 표시한다. 마스킹된 숫자 0을 실제 금액이나 합계로 해석하지 않는다"
        )
    if any(term in text for term in ("지급 대상", "지급대상", "보험금 지급")):
        requirements.append(
            "지급 판정에는 CONTRACT_STATUS, CLAIM_AMT, PAID_AMT, CLAIM_STATUS를 "
            "모두 확인하고 누락 시 재조회 필요 항목을 명시한다"
        )
    if any(term in text for term in ("주민번호", "rrn_masked")):
        requirements.append(
            "security_evidence의 plaintext_rows와 masked_format_rows를 근거로 평문 "
            "반환 건수와 YYMMDD-N****** 표시 형식을 답하고 원문을 추정하거나 복원하지 않는다"
        )
    if any(term in text for term in ("담보", "보장 개요", "가입금액")):
        requirements.append(
            "계약번호, 상품코드, 담보명, 가입금액을 계약 사실로 먼저 제시하고 약관 "
            "요약은 별도 근거로 표시한다"
        )
    if any(term in text for term in ("리니지", "청크", "원본 pdf", "추적")):
        requirements.append(
            "product_cd, document_id, chunk_id, 원본 파일명, 조항 위치 중 실제 제공된 "
            "식별자를 빠짐없이 표시하고 누락 식별자를 명시한다"
        )
    if _security_evidence_kind(text) in {"CONTRACT_SCOPE", "AUTH_DENIAL"}:
        requirements.append(
            "security_evidence의 차단 결과, 반환 건수, audit_log_id와 실행 사용자를 "
            "함께 제시하고 권한 밖 계약·고객·보험료 상세를 노출하지 않는다"
        )
    if _security_evidence_kind(text) == "CHANNEL_SCOPE":
        requirements.append(
            "security_evidence.channel_contract_counts의 네 채널 숫자를 모두 제시하고 "
            "전체는 실제 행 COUNT로 계산한다. 빈 집계 행을 1건으로 세지 않는다"
        )
    if "41047" in text:
        requirements.append(
            "41047 현행 2026-06-11·판매중·활성 Y 근거만 사용하고 41047_OLD, "
            "2022-03-16, 판매중지, 활성 N은 제외됐음을 명시한다"
        )
    if "대물배상" in text and "삼성화재" in text:
        requirements.append(
            "41048과 20071_0_20260611_file1의 대물배상 조항을 표로 비교하고 각 행에 "
            "상품코드, 원본 파일명, 조항 또는 페이지를 표시한다"
        )
    return requirements


def _question_needs_cross_source(question: str) -> bool:
    text = str(question or "").casefold()
    source_terms = (
        "약관",
        "삼성화재",
        "db손보",
        "타사",
        "경쟁사",
        "원본 pdf",
        "청크",
        "보유 자동차보험",
        "갱신 상담",
    )
    return any(keyword in text for keyword in source_terms)


def _question_prefers_structured_single(question: str) -> bool:
    text = str(question or "").casefold()
    return any(
        term in text
        for term in (
            "지급 대상",
            "지급대상",
            "주민번호",
            "rrn_masked",
            "납입구분",
            "납입 기준",
            "보험료 합계",
            "보험료 상세",
        )
    )


def _is_search_route(route: RoutedMcpTool) -> bool:
    properties = route.tool.schema.get("properties")
    if not isinstance(properties, Mapping):
        properties = {}
    route_text = f"{route.server_id} {route.tool.name} {route.tool.description}".casefold()
    return (
        _is_vector_route(route)
        or "query" in properties
        or "search" in route_text
        or "vector" in route_text
        or "rerank" in route_text
        or "hybrid" in route_text
    )


def _is_vector_route(route: RoutedMcpTool) -> bool:
    route_text = f"{route.server_id} {route.tool.name} {route.tool.description}".casefold()
    return any(
        marker in route_text
        for marker in ("vector", "hybrid", "rerank", "약관 검색")
    )


def _is_structured_route(route: RoutedMcpTool) -> bool:
    route_text = f"{route.server_id} {route.tool.name} {route.tool.description}".casefold()
    return not _is_vector_route(route) and any(
        marker in route_text for marker in ("select_ai", "select ai", "ords.query")
    )


def _select_unvisited_route(
    question: str,
    routes_by_key: Mapping[str, RoutedMcpTool],
    completed_route_keys: set[str],
) -> tuple[str, RoutedMcpTool] | None:
    candidates = [
        (key, route)
        for key, route in routes_by_key.items()
        if key not in completed_route_keys
    ]
    if not candidates:
        return None
    if _question_needs_cross_source(question):
        if not completed_route_keys:
            for key, route in candidates:
                if _is_structured_route(route):
                    return key, route
        for key, route in candidates:
            if _is_vector_route(route):
                return key, route
    return candidates[0]


def _fallback_tool_query_for_route(
    question: str,
    route: RoutedMcpTool,
    observations: list[Mapping[str, Any]],
) -> str:
    text = str(question or "").strip()
    if _is_structured_route(route):
        return _append_query_guidance(text, _select_ai_query_guidance(text))
    if _is_vector_route(route):
        structured_context = " ".join(
            str(item.get("result_excerpt") or "")
            for item in observations
            if "kb_mcp" in str(item.get("route_key") or "")
        )
        if structured_context:
            text = (
                f"{text} 구조화 조회에서 확인된 식별자와 파일명은 다음과 같다: "
                f"{structured_context[:1600]}"
            )
        return _append_query_guidance(text, _vector_query_guidance(text))
    return text


def _cross_source_vector_queries(
    question: str,
    observations: list[Mapping[str, Any]],
) -> list[str]:
    text = str(question or "").strip()
    folded = text.casefold()
    structured_context = " ".join(
        str(item.get("result_excerpt") or "")
        for item in observations
        if "kb_mcp" in str(item.get("route_key") or "")
    )[:1800]
    context_suffix = (
        f" 구조화 조회 식별자: {structured_context}" if structured_context else ""
    )
    queries: list[str]
    if "c1001025" in folded:
        queries = [
            "당사 상품코드 25213_B 현행 건강보험 약관 주요 보장 조항",
            "DB손보 원본 파일 약관_31084(03)_20260101 건강보험 약관 주요 보장 조항",
        ]
    elif "대물배상" in folded and "삼성화재" in folded:
        queries = [
            "당사 상품코드 41048 현행 자동차보험 대물배상 약관 조항",
            "삼성화재 원본 파일 20071_0_20260611_file1 개인용애니카다이렉트자동차보험 대물배상 약관 조항",
        ]
    elif "삼성화재" in folded and any(
        term in folded for term in ("비교", "차별", "갱신")
    ):
        queries = [
            "당사 상품코드 41048 현행 KB개인용자동차보험 관련 보장 약관 조항",
            "삼성화재 개인용애니카다이렉트자동차보험 현행 원본 약관 관련 보장 조항",
        ]
    else:
        queries = [text]
    return [
        _append_query_guidance(
            query + context_suffix,
            _vector_query_guidance(text),
        )
        for query in queries
    ]


def plan_mcp_execution_mode(
    *,
    question: str,
    routed_tools: list[RoutedMcpTool],
    model_profile_key: str,
    mode_override: str,
) -> Mapping[str, Any]:
    tool_catalog, routes_by_key = _agent_tool_catalog(routed_tools)
    route_keys = list(routes_by_key)
    fallback = _default_single_route(routed_tools)
    fallback_key = _route_key(fallback.server_id, fallback.tool.name)
    override = str(mode_override or "auto").strip().lower()
    if override in {"single", "agent"}:
        return {
            "mode": override,
            "route_key": fallback_key,
            "reason": f"사용자 선택: {override}",
            "model_profile": "",
        }
    if len(routed_tools) <= 1:
        return {
            "mode": "single",
            "route_key": fallback_key,
            "reason": "사용 가능한 MCP route가 1개라 단일 호출로 처리합니다.",
            "model_profile": "",
        }
    if _question_needs_cross_source(question):
        structured = next(
            (route for route in routed_tools if _is_structured_route(route)),
            fallback,
        )
        return {
            "mode": "agent",
            "route_key": _route_key(structured.server_id, structured.tool.name),
            "reason": (
                "계약·상품 데이터와 약관 근거를 함께 요구하여 구조화 MCP부터 "
                "벡터 MCP까지 순차 호출합니다."
            ),
            "model_profile": model_profile_key,
        }
    if _question_prefers_structured_single(question):
        structured = next(
            (route for route in routed_tools if _is_structured_route(route)),
            fallback,
        )
        return {
            "mode": "single",
            "route_key": _route_key(structured.server_id, structured.tool.name),
            "reason": "업무 테이블의 실제 컬럼과 결과 행만으로 검증 가능한 질의입니다.",
            "model_profile": model_profile_key,
        }
    try:
        profile = resolve_model_profile(model_profile_key)
        client = build_oci_genai_completion_client(
            profile.model_id,
            profile.answer_model_region,
            profile.answer_model_endpoint,
        )
        text = client.complete(
            system_prompt=(
                "You are a lightweight MCP execution planner. Decide whether "
                "the user's Korean question should be answered with one MCP "
                "tool call or with a multi-tool ReAct loop. Prefer mode=single "
                "unless the question explicitly requires comparing, combining, "
                "or validating evidence across different MCP tools/sources. "
                "Return only JSON matching the schema. Never request or expose "
                "bearer tokens."
            ),
            user_prompt=json.dumps(
                {
                    "question": question,
                    "routes": tool_catalog,
                    "single_policy": (
                        "counts, lists, summaries, and direct DB lookups should "
                        "use single unless another source is clearly required"
                    ),
                    "agent_policy": (
                        "use agent only for cross-source comparison, contract "
                        "plus terms/document search, or multi-step validation"
                    ),
                },
                ensure_ascii=False,
            ),
            response_schema={
                "type": "object",
                "additionalProperties": False,
                "required": ["mode", "route_key", "reason"],
                "properties": {
                    "mode": {"type": "string", "enum": ["single", "agent"]},
                    "route_key": {"type": "string", "enum": route_keys},
                    "reason": {"type": "string"},
                },
            },
            max_tokens=300,
            temperature=temperature_for_model_profile(profile),
        )
        parsed = json.loads(text)
    except Exception:
        return {
            "mode": "single",
            "route_key": fallback_key,
            "reason": "실행 방식 판단 실패로 단일 MCP 호출로 폴백했습니다.",
            "model_profile": model_profile_key,
        }
    if not isinstance(parsed, Mapping):
        return {
            "mode": "single",
            "route_key": fallback_key,
            "reason": "실행 방식 판단 응답 형식 오류로 단일 MCP 호출로 폴백했습니다.",
            "model_profile": model_profile_key,
        }
    mode = str(parsed.get("mode") or "single").strip().lower()
    route_key = str(parsed.get("route_key") or fallback_key).strip()
    if mode not in {"single", "agent"}:
        mode = "single"
    if route_key not in routes_by_key:
        route_key = fallback_key
    return {
        "mode": mode,
        "route_key": route_key,
        "reason": str(parsed.get("reason") or "").strip(),
        "model_profile": model_profile_key,
    }


def _plan_agent_step(
    *,
    question: str,
    conversation: list[Mapping[str, str]],
    tool_catalog: list[Mapping[str, Any]],
    observations: list[Mapping[str, Any]],
    model_profile_key: str,
) -> Mapping[str, Any]:
    route_keys = [str(item["route_key"]) for item in tool_catalog]
    try:
        profile = resolve_model_profile(model_profile_key)
        client = build_oci_genai_completion_client(
            profile.model_id,
            profile.answer_model_region,
            profile.answer_model_endpoint,
        )
        text = client.complete(
            system_prompt=(
                "You are a concise ReAct-style MCP tool planner. "
                "Use the available MCP tools to answer the Korean business question. "
                "If no tool has been called yet, choose action=call_tool. "
                "After each observation, decide whether another tool call is needed "
                "or action=final_answer is enough. Do not expose or request bearer tokens. "
                "tool_query must be plain natural-language query text only; do not include "
                "argument labels such as limit:, prompt:, query:, top_k:, or candidate_k:. "
                "Do not write SQL. Preserve identifiers exactly. If the user says "
                "계약번호, write it as 계약번호(CONTRACT_NO); if the user says 상품코드 "
                "or product code, write it as 상품코드(PRODUCT_CD). Do not convert one "
                "identifier type into the other. 고객번호, 고객ID, 고객 식별번호는 "
                "반드시 고객번호(CUST_ID)로 작성한다. For a cross-source question, "
                "call the structured kb_mcp route first to identify CUST_ID, CONTRACT_NO, "
                "PRODUCT_CD, insurer, clause name, and source file. Then call the vector "
                "route using those exact identifiers. Do not finish before both routes "
                "have been attempted. "
                "Return only JSON matching the schema."
            ),
            user_prompt=json.dumps(
                {
                    "question": question,
                    "conversation": conversation,
                    "available_routes": tool_catalog,
                    "observations": observations,
                    "max_tool_steps": MAX_AGENT_TOOL_STEPS,
                },
                ensure_ascii=False,
            ),
            response_schema={
                "type": "object",
                "additionalProperties": False,
                "required": ["thought", "action", "route_key", "tool_query"],
                "properties": {
                    "thought": {"type": "string"},
                    "action": {"type": "string", "enum": ["call_tool", "final_answer"]},
                    "route_key": {
                        "type": "string",
                        "enum": [*route_keys, AGENT_FINAL_ROUTE],
                    },
                    "tool_query": {"type": "string"},
                },
            },
            max_tokens=700,
            temperature=temperature_for_model_profile(profile),
        )
        parsed = json.loads(text)
    except Exception:
        raise McpToolRouterError("MCP agent planner 호출에 실패했습니다.") from None
    if not isinstance(parsed, Mapping):
        raise McpToolRouterError("MCP agent planner 응답 형식이 올바르지 않습니다.")
    return parsed


def run_mcp_agent_loop(
    *,
    question: str,
    conversation: list[Mapping[str, str]],
    routed_tools: list[RoutedMcpTool],
    servers: list[McpServer],
    bearer_token: str,
    limit: int,
    model_profile_key: str,
    progress_callback: Any = None,
) -> dict[str, Any]:
    tool_catalog, routes_by_key = _agent_tool_catalog(routed_tools)
    servers_by_id = {server.server_id: server for server in servers}
    observations: list[dict[str, Any]] = []
    steps: list[dict[str, Any]] = []
    last: dict[str, Any] | None = None
    attempted_route_keys: set[str] = set()
    actionable_route_keys: set[str] = set()
    completed_vector_queries: set[str] = set()
    stop_reason = ""

    for step_no in range(1, MAX_AGENT_TOOL_STEPS + 1):
        forced_plan = None
        if step_no == 1 and _question_needs_cross_source(question):
            forced_plan = _select_unvisited_route(
                question,
                routes_by_key,
                attempted_route_keys,
            )
        if (
            forced_plan is None
            and _question_needs_cross_source(question)
            and any(
                _is_structured_route(routes_by_key[key])
                for key in attempted_route_keys
                if key in routes_by_key
            )
        ):
            vector_route = next(
                (
                    (key, route)
                    for key, route in routes_by_key.items()
                    if _is_vector_route(route)
                ),
                None,
            )
            pending_queries = [
                query
                for query in _cross_source_vector_queries(question, observations)
                if query not in completed_vector_queries
            ]
            if vector_route is not None and pending_queries:
                forced_key, forced_route = vector_route
                forced_plan = (forced_key, forced_route)
        if forced_plan is not None:
            forced_key, forced_route = forced_plan
            pending_vector_queries = [
                query
                for query in _cross_source_vector_queries(question, observations)
                if query not in completed_vector_queries
            ]
            forced_query = (
                pending_vector_queries[0]
                if _is_vector_route(forced_route) and pending_vector_queries
                else _fallback_tool_query_for_route(
                    question,
                    forced_route,
                    observations,
                )
            )
            plan = {
                "thought": (
                    "비교 대상별 약관 근거를 독립 검색"
                    if _is_vector_route(forced_route)
                    else "교차 조회 식별자를 확보하기 위한 구조화 MCP 우선 호출"
                ),
                "action": "call_tool",
                "route_key": forced_key,
                "tool_query": forced_query,
            }
        else:
            try:
                plan = _plan_agent_step(
                    question=question,
                    conversation=conversation,
                    tool_catalog=tool_catalog,
                    observations=observations,
                    model_profile_key=model_profile_key,
                )
            except McpToolRouterError:
                fallback_route = _select_unvisited_route(
                    question,
                    routes_by_key,
                    attempted_route_keys,
                )
                if fallback_route is None or not _question_needs_cross_source(question):
                    raise
                fallback_key, route = fallback_route
                plan = {
                    "thought": "Agent planner 장애로 미호출 MCP route를 순차 실행",
                    "action": "call_tool",
                    "route_key": fallback_key,
                    "tool_query": _fallback_tool_query_for_route(
                        question,
                        route,
                        observations,
                    ),
                }
        action = str(plan.get("action") or "").strip()
        route_key = str(plan.get("route_key") or "").strip()
        thought = str(plan.get("thought") or "").strip()
        tool_query = _clean_agent_tool_query(plan.get("tool_query"), question)

        if action == "final_answer" and observations:
            forced = _select_unvisited_route(
                question,
                routes_by_key,
                attempted_route_keys,
            )
            if forced is None or not _question_needs_cross_source(question):
                stop_reason = "planner가 추가 MCP 호출이 불필요하다고 판단했습니다."
                break
            route_key, route = forced
            action = "call_tool"
            tool_query = _fallback_tool_query_for_route(
                question,
                route,
                observations,
            )
            thought = (
                f"{thought} / 비교·약관 질문인데 미호출 MCP route가 있어 "
                f"{route_key} 호출로 전환합니다."
            ).strip(" /")
        if action != "call_tool" and not observations:
            action = "call_tool"
        route = routes_by_key.get(route_key)
        if route is None:
            route = next(iter(routes_by_key.values()))
            route_key = _route_key(route.server_id, route.tool.name)
        is_distinct_vector_query = (
            _is_vector_route(route)
            and tool_query not in completed_vector_queries
        )
        if (
            route_key in attempted_route_keys
            and last is not None
            and not is_distinct_vector_query
        ):
            forced = _select_unvisited_route(
                question,
                routes_by_key,
                attempted_route_keys,
            )
            if forced is None:
                stop_reason = f"중복 MCP route 재호출 차단: {route_key}"
                break
            repeated_route_key = route_key
            route_key, route = forced
            tool_query = _fallback_tool_query_for_route(
                question,
                route,
                observations,
            )
            thought = (
                f"{thought} / 중복 MCP route {repeated_route_key} 대신 "
                f"미호출 route {route_key}로 전환합니다."
            ).strip(" /")
        server = servers_by_id.get(route.server_id)
        if server is None:
            raise PublicMcpError("선택된 MCP 서버 설정을 찾지 못했습니다.")

        started = perf_counter()
        arguments = build_mcp_tool_arguments(
            route.tool,
            tool_query,
            int(limit),
            preferred_tool=server.default_tool,
        )
        raw_result = call_tool(
            base_url=server.endpoint_url,
            bearer_token=bearer_token,
            tool=route.tool,
            arguments=arguments,
        )
        parsed = _content_text_json(raw_result)
        mcp_result = parsed if parsed is not None else raw_result
        elapsed = perf_counter() - started
        summary = _mcp_summary(mcp_result)
        step = {
            "step": step_no,
            "thought": thought,
            "action": "call_tool",
            "server_id": server.server_id,
            "tool_name": route.tool.name,
            "route_key": route_key,
            "tool_query": tool_query,
            "arguments": arguments,
            "mcp_result": mcp_result,
            "result_summary": summary,
            "elapsed_seconds": round(elapsed, 3),
        }
        steps.append(step)
        observations.append(
            {
                "step": step_no,
                "route_key": route_key,
                "tool_query": tool_query,
                "result_summary": summary,
                "result_excerpt": _bounded_json(mcp_result, max_chars=6000),
            }
        )
        last = {
            "server": server,
            "tool": route.tool,
            "arguments": arguments,
            "mcp_result": mcp_result,
        }
        attempted_route_keys.add(route_key)
        if _is_vector_route(route):
            completed_vector_queries.add(tool_query)
        if _mcp_has_actionable_result(mcp_result):
            actionable_route_keys.add(route_key)
        if progress_callback:
            progress_callback(step)

    if last is None:
        raise PublicMcpError("MCP agent가 실행한 tool 호출이 없습니다.")
    return {
        "server": last["server"],
        "tool": last["tool"],
        "arguments": last["arguments"],
        "mcp_result": last["mcp_result"],
        "agent_steps": steps,
        "stop_reason": stop_reason,
        "attempted_route_keys": sorted(attempted_route_keys),
        "actionable_route_keys": sorted(actionable_route_keys),
    }


def _render_mcp_result_sections(
    details: Mapping[str, Any],
    message_key: str,
) -> None:
    mcp_result = details.get("mcp_result", {})
    agent_steps = details.get("agent_steps")
    execution_events = details.get("execution_events")
    generated_sql = _mcp_generated_sql(mcp_result)
    items = _mcp_items(mcp_result)

    if isinstance(execution_events, list) and execution_events:
        with st.expander(f"처리 시간 로그 · {len(execution_events)}건"):
            rows: list[dict[str, Any]] = []
            for event in execution_events:
                if not isinstance(event, Mapping):
                    continue
                rows.append(
                    {
                        "진행률": event.get("percent"),
                        "단계": str(event.get("label") or ""),
                        "단계소요(s)": event.get("elapsed_seconds"),
                        "누적소요(s)": event.get("total_elapsed_seconds"),
                        "상세": str(event.get("detail") or ""),
                    }
                )
            if rows:
                st.dataframe(rows, hide_index=True, width="stretch")

    if isinstance(agent_steps, list) and agent_steps:
        with st.expander(f"Agent 실행 단계 · {len(agent_steps)}회"):
            stop_reason = str(details.get("agent_stop_reason") or "").strip()
            if stop_reason:
                st.caption(f"종료 사유: {stop_reason}")
            for step in agent_steps:
                if not isinstance(step, Mapping):
                    continue
                st.markdown(
                    f"**Step {step.get('step')} · "
                    f"{step.get('server_id')} / {step.get('tool_name')}**"
                )
                thought = str(step.get("thought") or "").strip()
                tool_query = str(step.get("tool_query") or "").strip()
                if thought:
                    st.caption(f"판단: {thought}")
                if tool_query:
                    st.caption(f"툴 질의: {tool_query}")
                st.markdown("전달 arguments")
                st.json(step.get("arguments", {}))
                st.markdown("응답 요약")
                st.json(step.get("result_summary", {}))

    if generated_sql:
        with st.expander("실행 SQL", expanded=True):
            st.code(generated_sql, language="sql")

    if items:
        with st.expander(f"조회 결과 · {len(items)}건", expanded=True):
            display_items = items[:100]
            st.dataframe(display_items, use_container_width=True)
            if len(items) > len(display_items):
                st.caption(f"화면에는 최초 {len(display_items)}건만 표시합니다.")
    elif generated_sql:
        st.info("실행 SQL은 생성되었지만 반환된 조회 결과 행이 없습니다.")

    with st.expander("MCP 호출 상세"):
        st.write(f"route: {details.get('server_id')} / {details.get('tool_name')}")
        execution_mode = str(details.get("execution_mode") or "").strip()
        execution_reason = str(details.get("execution_mode_reason") or "").strip()
        if execution_mode:
            st.caption(
                "실행 방식: "
                + execution_mode
                + (f" · {execution_reason}" if execution_reason else "")
            )
        reasoning_model = str(details.get("reasoning_model_profile") or "").strip()
        answer_model = str(details.get("answer_model_profile") or "").strip()
        fallback_model = str(
            details.get("answer_synthesis_fallback_model") or ""
        ).strip()
        planner_model = str(details.get("execution_mode_model_profile") or "").strip()
        if planner_model or reasoning_model or answer_model:
            st.caption(
                "모델: "
                + f"planner={planner_model or '-'} · "
                + f"reasoning={reasoning_model or '-'} · "
                + f"answer={answer_model or '-'}"
                + (f" · fallback={fallback_model}" if fallback_model else "")
            )
        if details.get("standalone_question"):
            st.caption(f"MCP 질의: {details['standalone_question']}")
        if details.get("tool_query"):
            st.caption(f"툴별 정제 질의: {details['tool_query']}")
        st.markdown("전달 arguments")
        st.json(details.get("arguments", {}))
        st.markdown("MCP 응답 요약")
        st.json(_mcp_summary(mcp_result))
        synthesis_error = details.get("answer_synthesis_error")
        if isinstance(synthesis_error, Mapping) and synthesis_error:
            st.markdown("최종 답변 합성 실패 진단")
            st.json(dict(synthesis_error))
        raw_key = f"poc4_show_raw_mcp_{message_key}"
        if st.button("MCP 원본 JSON 보기/숨기기", key=f"{raw_key}_button"):
            st.session_state[raw_key] = not bool(st.session_state.get(raw_key))
        if st.session_state.get(raw_key):
            st.json(mcp_result)
        routes = details.get("discovered_routes")
        failures = details.get("discovery_failures")
        if routes or failures:
            st.markdown("Discovered routes")
            st.json({"routes": routes or [], "failures": failures or []})


def _conversation_context(messages: list[Mapping[str, Any]]) -> list[dict[str, str]]:
    context: list[dict[str, str]] = []
    for message in messages[-MAX_CONVERSATION_MESSAGES:]:
        role = str(message.get("role") or "").strip()
        content = str(message.get("content") or "").strip()
        if role in {"user", "assistant"} and content:
            context.append({"role": role, "content": content[:2000]})
    return context


def resolve_standalone_question(
    *,
    question: str,
    messages: list[Mapping[str, Any]],
    model_profile_key: str,
) -> str:
    context = _conversation_context(messages)
    if not context:
        return question
    try:
        profile = resolve_model_profile(model_profile_key)
        client = build_oci_genai_completion_client(
            profile.model_id,
            profile.answer_model_region,
            profile.answer_model_endpoint,
        )
        text = client.complete(
            system_prompt=(
                "Rewrite the user's latest Korean question into one standalone "
                "MCP tool query. Use the conversation only to resolve references. "
                "Do not answer the question."
            ),
            user_prompt=json.dumps(
                {"conversation": context, "latest_question": question},
                ensure_ascii=False,
            ),
            response_schema={
                "type": "object",
                "additionalProperties": False,
                "required": ["standalone_question"],
                "properties": {"standalone_question": {"type": "string"}},
            },
            max_tokens=500,
            temperature=temperature_for_model_profile(profile),
        )
        parsed = json.loads(text)
        rewritten = str(parsed.get("standalone_question") or "").strip()
    except Exception:
        return question
    return rewritten or question


def prepare_tool_query_for_mcp(
    *,
    question: str,
    server: McpServer,
    tool: McpTool,
    model_profile_key: str,
    selected_user_id: str = "",
    selected_user_role: str = "",
    selected_user_channel: str = "",
    selected_user_scope: str = "",
) -> str:
    """Rewrite the user question into a selected-tool-specific natural query."""

    fallback = str(question or "").strip()
    if not fallback:
        return fallback
    normalized_question = " ".join(fallback.casefold().split())
    individual_scope_terms = (
        "내 담당이 아닌",
        "다른 설계사",
        "다른 채널",
        "타 설계사",
        "타 채널",
    )
    is_select_ai_tool = (
        server.server_id == "kb_mcp"
        and "select_ai" in tool.name.casefold()
    )
    query_guidance = (
        _select_ai_query_guidance(fallback)
        if is_select_ai_tool
        else _vector_query_guidance(fallback)
        if "vector" in server.server_id.casefold()
        else []
    )
    is_channel_scope_validation = (
        bool(selected_user_id.strip())
        and is_select_ai_tool
        and (
            "공통계정" in normalized_question
            or "공통 계정" in normalized_question
        )
        and "채널" in normalized_question
        and any(term in normalized_question for term in ("조회", "접근", "권한", "vpd"))
    )
    if is_channel_scope_validation:
        user_id = selected_user_id.strip()
        expected_channel = selected_user_channel.strip() or "선택 채널"
        return (
            f"POC_2.KB_CONTRACTS에서 현재 VPD 공통계정 사용자 {user_id}에게 실제로 "
            f"조회되는 계약을 검증해줘. 이 사용자의 권한 채널은 {expected_channel}이다. "
            "조회 결과 전체를 한 번 집계해서 FC_CHANNEL='다이렉트' 계약 건수, "
            "FC_CHANNEL='설계사' 계약 건수, FC_CHANNEL='GA' 계약 건수, "
            "FC_CHANNEL='제휴' 계약 건수를 각각 별도 컬럼으로 반드시 보여줘. "
            "계약번호는 CONTRACT_NO, 고객번호는 CUST_ID, 담당자는 FC_ID, "
            "담당 채널은 FC_CHANNEL 실제 컬럼만 사용해줘. "
            "VPD 정책, 시스템 카탈로그, 사용자 또는 권한 메타데이터는 조회하지 말고 "
            "현재 토큰으로 보이는 KB_CONTRACTS 업무 데이터만 사용해줘."
        )
    is_individual_scope_validation = (
        bool(selected_user_id.strip())
        and is_select_ai_tool
        and any(term in normalized_question for term in individual_scope_terms)
        and any(term in normalized_question for term in ("조회", "접근", "권한", "vpd"))
    )
    if is_individual_scope_validation:
        user_id = selected_user_id.strip()
        return (
            f"POC_2.KB_CONTRACTS에서 현재 VPD 사용자 {user_id}에게 실제로 조회되는 "
            "계약을 FC_CHANNEL별로 집계해줘. 전체 계약 건수, "
            f"FC_ID='{user_id}'인 본인 담당 계약 건수, "
            f"FC_ID<>'{user_id}'인 다른 설계사 계약 건수를 각각 보여줘. "
            "계약번호는 CONTRACT_NO, 고객번호는 CUST_ID, 담당자는 FC_ID, "
            "담당 채널은 FC_CHANNEL 실제 컬럼만 사용해줘. "
            "VPD 정책, 시스템 카탈로그, 사용자 또는 권한 메타데이터는 조회하지 말고 "
            "현재 토큰으로 보이는 KB_CONTRACTS 업무 데이터만 사용해줘."
        )
    properties = tool.schema.get("properties")
    if not isinstance(properties, Mapping):
        properties = {}
    safe_properties = {
        str(name): {
            "type": value.get("type") if isinstance(value, Mapping) else None,
            "description": (
                str(value.get("description") or "")[:500]
                if isinstance(value, Mapping)
                else ""
            ),
        }
        for name, value in properties.items()
        if str(name).lower()
        not in {"bearertoken", "bearer_token", "token", "authorization"}
    }
    try:
        profile = resolve_model_profile(model_profile_key)
        client = build_oci_genai_completion_client(
            profile.model_id,
            profile.answer_model_region,
            profile.answer_model_endpoint,
        )
        text = client.complete(
            system_prompt=(
                "You prepare one selected-tool-specific MCP query. "
                "Return only JSON matching the schema. Do not answer the user. "
                "Do not write SQL. Do not expose or request bearer tokens. "
                "The output tool_query must be plain Korean natural-language text "
                "for the selected MCP tool, not a JSON string and not argument labels. "
                "Never include labels such as prompt:, query:, question:, limit:, "
                "top_k:, candidate_k:, or max_rows:. "
                "Use the tool description and input properties to make the query "
                "fit that tool. For database Select AI tools, make filters, columns, "
                "and identifier meanings explicit. For document/vector search tools, "
                "produce compact search keywords and document terms. Preserve all IDs "
                "and numbers exactly. If the user says 계약번호, write 계약번호(CONTRACT_NO); "
                "if the user says 상품코드 or product code, write 상품코드(PRODUCT_CD). "
                "고객번호, 고객ID, 고객 식별번호는 반드시 고객번호(CUST_ID)로 작성한다. "
                "담당 채널은 담당 채널(FC_CHANNEL), 설계사는 설계사(FC_ID), "
                "계약상태는 계약상태(CONTRACT_STATUS)로 작성한다. "
                "VPD 적용 여부, 권한 범위, 접근 가능 여부를 묻는 질의는 VPD 정책, "
                "시스템 카탈로그, 사용자 또는 권한 메타데이터를 조회하지 않는다. "
                "현재 선택 사용자의 VPD 컨텍스트로 보이는 POC_2.KB_CONTRACTS의 "
                "업무 데이터 결과만 사용해 실효 접근 범위를 검증한다. "
                "DB Select AI 질의에는 제공된 KB 테이블의 실제 컬럼만 사용하고, "
                "확인되지 않은 영문 컬럼명을 추측하여 생성하지 않는다. "
                "보험금 지급 판단은 KB_CONTRACTS.CONTRACT_STATUS와 "
                "KB_CLAIMS.CLAIM_AMT, PAID_AMT, CLAIM_STATUS를 포함한다. "
                "담보 질문은 KB_COVERAGES.COVERAGE_NM, COVERAGE_TYPE, "
                "INSURED_AMT, RENEW_DUE_DT를 포함한다. 보험료 질문에서 개별값과 "
                "채널 합계는 별도 범위로 계산한다. 주민번호 확인 질문은 "
                "KB_CUSTOMERS.RRN_MASKED 실제 결과를 조회하며 WHERE 1=0을 만들지 않는다. "
                "Do not convert one identifier type into the other."
            ),
            user_prompt=json.dumps(
                {
                    "question": fallback,
                    "selected_server_id": server.server_id,
                    "selected_tool_name": tool.name,
                    "selected_tool_description": tool.description[:1200],
                    "input_properties": safe_properties,
                    "default_tool": server.default_tool,
                    "selected_vpd_user": {
                        "user_id": selected_user_id.strip(),
                        "role": selected_user_role.strip(),
                        "channel": selected_user_channel.strip(),
                        "scope": selected_user_scope.strip(),
                    },
                    "required_query_guidance": query_guidance,
                },
                ensure_ascii=False,
            ),
            response_schema={
                "type": "object",
                "additionalProperties": False,
                "required": ["tool_query"],
                "properties": {"tool_query": {"type": "string"}},
            },
            max_tokens=600,
            temperature=temperature_for_model_profile(profile),
        )
        parsed = json.loads(text)
        rewritten = _clean_agent_tool_query(parsed.get("tool_query"), fallback)
    except Exception:
        return _append_query_guidance(fallback, query_guidance)
    return _append_query_guidance(rewritten or fallback, query_guidance)


def synthesize_answer(
    *,
    question: str,
    conversation: list[Mapping[str, str]],
    server: McpServer,
    tool: McpTool,
    arguments: Mapping[str, Any],
    mcp_result: Any,
    model_profile_key: str,
    agent_steps: list[Mapping[str, Any]] | None = None,
    security_evidence: Mapping[str, Any] | None = None,
    business_evidence: Mapping[str, Any] | None = None,
) -> Mapping[str, Any]:
    def build_user_prompt(*, compact: bool) -> str:
        evidence_items = 10 if compact else 20
        evidence_text = 260 if compact else 380
        step_items = 4 if compact else 6
        step_text = 300 if compact else 480
        prompt_payload = {
            "question": question,
            "required_answer_checks": _answer_requirements(question),
            "conversation": [] if compact else conversation[-4:],
            "selected_route": {
                "server_id": server.server_id,
                "tool_name": tool.name,
            },
            "tool_arguments": dict(arguments),
            "security_evidence": dict(security_evidence or {}),
            "business_evidence": _bounded_json(
                dict(business_evidence or {}),
                max_chars=7000 if compact else 11000,
            ),
            "mcp_result": _bounded_json(
                _mcp_answer_evidence(
                    mcp_result,
                    max_items=evidence_items,
                    max_text=evidence_text,
                    include_sql=False,
                ),
                max_chars=6000 if compact else 8000,
            ),
            "agent_steps": [
                {
                    "step": step.get("step"),
                    "server_id": step.get("server_id"),
                    "tool_name": step.get("tool_name"),
                    "tool_query": step.get("tool_query"),
                    "arguments": step.get("arguments", {}),
                    "mcp_result_evidence": _bounded_json(
                        _mcp_answer_evidence(
                            step.get("mcp_result"),
                            max_items=step_items,
                            max_text=step_text,
                            include_sql=False,
                        ),
                        max_chars=3600 if compact else 5200,
                    ),
                }
                for step in (agent_steps or [])
            ],
        }
        return json.dumps(prompt_payload, ensure_ascii=False)

    user_prompt = build_user_prompt(compact=False)
    response_schema = {
        "type": "object",
        "additionalProperties": False,
        "required": ["answer", "basis", "limitations"],
        "properties": {
            "answer": {"type": "string"},
            "basis": {
                "type": "array",
                "items": {"type": "string"},
                "maxItems": 5,
            },
            "limitations": {"type": "string"},
        },
    }
    diagnostics: dict[str, Any] = {
        "model_profile": str(model_profile_key),
        "payload_chars": len(user_prompt),
        "agent_steps": len(agent_steps or []),
        "security_evidence": bool(security_evidence),
        "business_evidence": bool(business_evidence),
    }
    try:
        profile = resolve_model_profile(model_profile_key)
        client = build_oci_genai_completion_client(
            profile.model_id,
            profile.answer_model_region,
            profile.answer_model_endpoint,
        )
        diagnostics.update(
            {
                "model_key": profile.model_key,
                "model_id": profile.model_id,
                "region": profile.answer_model_region,
            }
        )
    except Exception as exc:
        diagnostics.update(
            {
                "stage": "client_init",
                "error_type": type(exc).__name__,
                "error": str(exc)[:500],
            }
        )
        LOG.warning("poc4_answer_synthesis_client_init_failed %s", diagnostics)
        raise AnswerSynthesisError(
            "MCP 결과 기반 최종 답변 생성에 실패했습니다.",
            diagnostics,
        ) from None

    system_prompt = (
        "You write final Korean answers from MCP tool results. "
        "Use English for internal instructions, but the final answer "
        "field must be written only in Korean. "
        "Use only the provided MCP results as evidence. Do not invent "
        "facts. If the MCP result is an error or permission denial, "
        "say that clearly. Treat items_count=0 or results_count=0 as an actual "
        "zero-row result, not as missing tool delivery. Never claim that an audit "
        "event exists unless its identifier or audit result is in the evidence. "
        "For cross-source questions, keep structured contract/product facts separate "
        "from vector clause evidence, then combine only matching identifiers. If one "
        "source is missing, list confirmed and unconfirmed points separately instead "
        "of saying only that comparison is impossible. Do not confuse row-level "
        "masked values with aggregate values. Preserve product_cd, contract_no, "
        "document_id, chunk_id, source filename, version, and clause locator when "
        "they are provided. Security evidence is a predefined token-scoped database "
        "verification result and takes precedence over ambiguous redacted MCP values. "
        "Business evidence is a Bearer-token-validated result with explicit CUST_ID and "
        "FC_ID scoping for customer data, plus active catalog document/chunk evidence. "
        "It takes precedence when Select AI returns an ambiguous or incorrect zero-row "
        "result. Never expose customer rows outside the business evidence scope. "
        "Show audit_log_id when it is provided. Follow every required_answer_checks item in the user "
        "payload. When the user requests a result list, include every "
        "row provided in the MCP evidence, up to 20 rows, and state the total "
        "returned row count. Do not arbitrarily stop at five rows. Keep the "
        "answer concise and business-readable."
    )
    for attempt in range(1, 3):
        compact = attempt > 1
        if compact:
            user_prompt = build_user_prompt(compact=True)
            diagnostics["payload_chars"] = len(user_prompt)
            diagnostics["compact_payload"] = True
        text = ""
        try:
            text = client.complete(
                system_prompt=system_prompt,
                user_prompt=user_prompt,
                response_schema=response_schema,
                max_tokens=1600,
                temperature=temperature_for_model_profile(profile),
            )
            if not text.strip():
                raise ValueError("empty structured response")
            parsed = json.loads(text)
            if not isinstance(parsed, Mapping) or not isinstance(
                parsed.get("answer"), str
            ):
                raise ValueError("invalid structured answer")
            enriched = dict(parsed)
            enriched["answer"] = _enrich_answer_with_security_evidence(
                str(parsed.get("answer") or ""),
                security_evidence,
            )
            enriched["answer"] = _enrich_answer_with_business_evidence(
                str(enriched.get("answer") or ""),
                business_evidence,
            )
            basis = list(parsed.get("basis") or [])
            if security_evidence and security_evidence.get("audit_log_id"):
                basis.append(
                    "보안 검증 증적: "
                    f"{security_evidence.get('audit_source')} "
                    f"LOG_ID={security_evidence.get('audit_log_id')}"
                )
            if business_evidence and business_evidence.get("evidence_complete"):
                basis.append(
                    "업무 검증 증적: "
                    + ", ".join(
                        str(source)
                        for source in business_evidence.get("evidence_sources", [])
                    )
                )
            evidence_verified = bool(
                business_evidence
                and business_evidence.get("evidence_complete")
                and not business_evidence.get("evidence_error")
            ) or bool(
                security_evidence
                and security_evidence.get("evidence_type")
                and not security_evidence.get("evidence_error")
            )
            if evidence_verified:
                enriched["limitations"] = ""
            enriched["basis"] = basis[:5]
            return enriched
        except Exception as exc:
            last_error = {
                "stage": "completion",
                "attempt": attempt,
                "error_type": type(exc).__name__,
                "error": str(exc)[:500],
                "raw_text_chars": len(text),
                "raw_text_head": text[:200],
            }
            diagnostics.update(last_error)
            LOG.warning("poc4_answer_synthesis_attempt_failed %s", diagnostics)
            if str(exc) == "empty structured response":
                break

    raise AnswerSynthesisError(
        "MCP 결과 기반 최종 답변 생성에 실패했습니다.",
        diagnostics,
    ) from None


def fallback_answer_from_mcp(
    mcp_result: Any,
    agent_steps: list[Mapping[str, Any]] | None = None,
) -> str:
    items = _mcp_items(mcp_result)
    payload = _mcp_response_payload(mcp_result)
    results = payload.get("results")
    lines = ["MCP 조회는 완료됐지만 최종 답변 합성에 실패했습니다."]
    if items:
        lines.append(f"조회 결과는 {len(items)}건입니다.")
        lines.append("")
        lines.append("주요 결과:")
        for item in items[:20]:
            compact = _compact_evidence_item(item, max_text=300)
            lines.append(f"- {_bounded_json(compact, max_chars=500)}")
    elif isinstance(results, list) and results:
        lines.append(f"검색 결과는 {len(results)}건입니다.")
        lines.append("")
        lines.append("주요 검색 결과:")
        for item in results[:20]:
            compact = _compact_evidence_item(item, max_text=320)
            if isinstance(compact, Mapping):
                title = " / ".join(
                    str(compact.get(key) or "")
                    for key in ("insurer", "product_name", "chunk_type")
                    if compact.get(key)
                )
                body = str(
                    compact.get("display_markdown")
                    or compact.get("text")
                    or compact.get("content")
                    or ""
                ).strip()
                page = compact.get("physical_page_range")
                suffix = f" (p.{page})" if page else ""
                lines.append(f"- {title or '검색 결과'}{suffix}: {body}")
            else:
                lines.append(f"- {compact}")
    elif agent_steps:
        lines.append(f"Agent는 MCP tool을 {len(agent_steps)}회 호출했습니다.")
    else:
        lines.append("상세 영역에서 MCP 원본 응답을 확인해 주세요.")
    return "\n".join(lines)


_DISPLAY_MARKDOWN_METADATA_KEYS = (
    "insurer",
    "product_name",
    "product_cd",
    "product_code",
    "document_title",
    "title",
    "document_id",
    "chunk_id",
    "source_file_name",
    "file_name",
    "ext_file_nm",
    "clause_version",
    "chunk_type",
    "logical_locator",
    "physical_page_range",
    "page",
    "score",
)


def _display_markdown_records(details: Mapping[str, Any] | None) -> list[dict[str, Any]]:
    if not isinstance(details, Mapping):
        return []
    sources: list[Any] = [details.get("mcp_result")]
    agent_steps = details.get("agent_steps")
    if isinstance(agent_steps, list):
        sources.extend(
            step.get("mcp_result")
            for step in agent_steps
            if isinstance(step, Mapping)
        )

    records: list[dict[str, Any]] = []
    seen: set[tuple[str, str, str]] = set()

    def collect(value: Any, depth: int = 0) -> None:
        if depth > 12 or len(records) >= 50:
            return
        if isinstance(value, Mapping):
            display_markdown = value.get("display_markdown")
            if isinstance(display_markdown, str) and display_markdown.strip():
                record = {
                    key: value.get(key)
                    for key in _DISPLAY_MARKDOWN_METADATA_KEYS
                    if value.get(key) not in (None, "", [])
                }
                record["display_markdown"] = display_markdown.strip()
                identity = (
                    str(record.get("document_id") or ""),
                    str(record.get("chunk_id") or ""),
                    record["display_markdown"],
                )
                if identity not in seen:
                    seen.add(identity)
                    records.append(record)
            for child in value.values():
                if isinstance(child, (Mapping, list, tuple)):
                    collect(child, depth + 1)
        elif isinstance(value, (list, tuple)):
            for child in value:
                collect(child, depth + 1)

    for source in sources:
        collect(source)
        if isinstance(source, Mapping):
            collect(_mcp_response_payload(source))
    return records


def _display_markdown_match_score(basis: str, record: Mapping[str, Any]) -> int:
    normalized_basis = " ".join(basis.casefold().split())
    score = 0
    for key in _DISPLAY_MARKDOWN_METADATA_KEYS:
        value = " ".join(str(record.get(key) or "").casefold().split())
        if len(value) >= 3 and value in normalized_basis:
            score += min(len(value), 40)
            if key in {
                "document_id",
                "chunk_id",
                "source_file_name",
                "file_name",
                "product_name",
            }:
                score += 30
    return score


def _match_basis_to_display_markdown(
    basis: list[Any],
    records: list[dict[str, Any]],
) -> list[dict[str, Any] | None]:
    remaining = set(range(len(records)))
    matched: list[dict[str, Any] | None] = []
    for item in basis:
        basis_text = str(item or "")
        ranked = sorted(
            (
                (_display_markdown_match_score(basis_text, records[index]), index)
                for index in remaining
            ),
            reverse=True,
        )
        selected_index: int | None = None
        if ranked and ranked[0][0] > 0:
            selected_index = ranked[0][1]
        elif remaining:
            selected_index = min(remaining)
        if selected_index is None:
            matched.append(None)
            continue
        remaining.remove(selected_index)
        matched.append(records[selected_index])
    return matched


def _display_markdown_source_label(record: Mapping[str, Any]) -> str:
    source = next(
        (
            str(record.get(key) or "").strip()
            for key in (
                "source_file_name",
                "file_name",
                "ext_file_nm",
                "product_name",
                "document_title",
                "document_id",
            )
            if str(record.get(key) or "").strip()
        ),
        "MCP 참조 문서",
    )
    details: list[str] = []
    page = str(record.get("physical_page_range") or record.get("page") or "").strip()
    chunk_id = str(record.get("chunk_id") or "").strip()
    clause_version = str(record.get("clause_version") or "").strip()
    if page:
        details.append(f"페이지 {page}")
    if chunk_id:
        details.append(f"청크 {chunk_id}")
    if clause_version:
        details.append(f"약관 버전 {clause_version}")
    return " · ".join([source, *details])


def _readable_basis_summary(value: Any) -> str:
    text = " ".join(str(value or "").split()).strip()
    if not text:
        return ""
    metadata_match = re.search(
        r"\s+(?:문서[_ ]?id|document_id|chunk_id|logical_locator|"
        r"physical_page_range)\s*:",
        text,
        flags=re.IGNORECASE,
    )
    if metadata_match:
        readable = text[: metadata_match.start()].strip(" /·:-")
        if readable:
            return readable
    return text


def _render_assistant_message(
    message: Mapping[str, Any],
    message_key: str,
) -> None:
    st.markdown(str(message.get("content") or ""))
    details = message.get("details")
    basis = message.get("basis")
    if isinstance(basis, list) and basis:
        display_records = _display_markdown_records(
            details if isinstance(details, Mapping) else None
        )
        matched_records = _match_basis_to_display_markdown(basis, display_records)
        with st.expander(f"답변 근거 요약 · {len(basis)}건"):
            for index, (item, record) in enumerate(
                zip(basis, matched_records),
                start=1,
            ):
                with st.container(border=True):
                    readable_summary = _readable_basis_summary(item)
                    source_title = (
                        _display_markdown_source_label(record).split(" · ", 1)[0]
                        if isinstance(record, Mapping)
                        else ""
                    )
                    heading = f"근거 {index}"
                    if source_title:
                        heading += f" · {source_title}"
                    st.markdown(f"#### {heading}")
                    if readable_summary and (
                        not source_title
                        or readable_summary.casefold() != source_title.casefold()
                    ):
                        st.markdown(readable_summary)
                    if isinstance(record, Mapping):
                        st.caption(
                            "참조 문서 · "
                            + _display_markdown_source_label(record)
                        )
                        st.markdown("**MCP 원문 근거 (`display_markdown`)**")
                        st.markdown(str(record.get("display_markdown") or ""))
    limitations = str(message.get("limitations") or "").strip()
    if limitations:
        st.caption(f"제약/주의: {limitations}")
    if isinstance(details, Mapping):
        _render_mcp_result_sections(details, message_key)


def _render_chat_turn(turn: Mapping[str, Any]) -> None:
    user_label = str(turn.get("selected_user_id") or "").strip()
    created_at = str(turn.get("created_at") or "").strip()
    meta = " · ".join(item for item in (created_at, user_label) if item)
    with st.chat_message("user"):
        if meta:
            st.caption(meta)
        st.markdown(str(turn.get("question") or ""))
    with st.chat_message("assistant"):
        _render_assistant_message(
            {
                "content": str(turn.get("answer") or ""),
                "basis": turn.get("basis", []),
                "limitations": str(turn.get("limitations") or ""),
                "details": turn.get("details", {}),
            },
            f"turn_{turn.get('turn_id')}",
        )


def _render_chat_history(conversation_id: str, page_key: str) -> None:
    st.markdown(
        '<div class="kb-section-title result" role="heading" aria-level="3">'
        "질의 결과"
        "</div>",
        unsafe_allow_html=True,
    )
    total_turns = count_chat_turns(conversation_id)
    if total_turns <= 0:
        st.info("아직 저장된 대화가 없습니다.")
        return

    max_page = max((total_turns - 1) // CHAT_TURNS_PER_PAGE, 0)
    current_page = int(st.session_state.get(page_key, 0) or 0)
    current_page = min(max(current_page, 0), max_page)
    st.session_state[page_key] = current_page

    start_no = current_page * CHAT_TURNS_PER_PAGE + 1
    end_no = min(start_no + CHAT_TURNS_PER_PAGE - 1, total_turns)
    st.caption(
        f"총 {total_turns}건 · 최신순 {start_no}~{end_no}건 표시 · "
        f"페이지 {current_page + 1}/{max_page + 1}"
    )

    nav_cols = st.columns([1, 1, 1, 5])
    with nav_cols[0]:
        if st.button(
            "최신",
            key=f"{page_key}_first_{conversation_id}",
            disabled=current_page == 0,
        ):
            st.session_state[page_key] = 0
            st.rerun()
    with nav_cols[1]:
        if st.button(
            "이전 내역",
            key=f"{page_key}_older_{conversation_id}",
            disabled=current_page >= max_page,
        ):
            st.session_state[page_key] = current_page + 1
            st.rerun()
    with nav_cols[2]:
        if st.button(
            "다음 내역",
            key=f"{page_key}_newer_{conversation_id}",
            disabled=current_page <= 0,
        ):
            st.session_state[page_key] = current_page - 1
            st.rerun()

    turns = load_chat_turns(conversation_id, current_page)
    if not turns:
        st.info("아직 저장된 대화가 없습니다.")
        return
    for turn in turns:
        _render_chat_turn(turn)


def _render_architecture_tab() -> None:
    st.subheader("운영 아키텍처")
    st.caption("사용자 권한을 기준으로 AI 질의, MCP 도구, Oracle Database를 연결합니다.")
    experience, integration, security = st.columns(3)
    with experience:
        st.markdown("**01 · 사용자와 권한**")
        st.write("사용자 역할과 담당 범위를 요청 컨텍스트에 적용합니다.")
    with integration:
        st.markdown("**02 · AI와 MCP**")
        st.write("질의에 맞는 데이터·지식 검색 도구를 선택합니다.")
    with security:
        st.markdown("**03 · 데이터와 보안**")
        st.write("VPD와 감사 정책으로 데이터 접근을 통제합니다.")

    st.divider()
    st.markdown("#### 질의 처리 흐름")
    st.markdown(
        "1. 사용자 권한과 질문을 요청에 반영합니다.  \n"
        "2. AI가 필요한 MCP 도구를 선택합니다.  \n"
        "3. Oracle Database에서 근거를 조회하고 결과를 표시합니다."
    )


def _render_fga_audit_tab() -> None:
    st.markdown(
        '<div class="kb-section-title input" role="heading" aria-level="3">'
        '감사로그 ( 오라클 <strong>FGA</strong> )'
        '</div>',
        unsafe_allow_html=True,
    )
    st.markdown(
        '<div class="kb-audit-lead">'
        'Oracle FGA 정책 상태와 민감 컬럼 접근 이력을 시간순으로 확인합니다. '
        '조회 조건을 선택하면 정책·객체·사용자·SQL 원문을 함께 비교할 수 있습니다.'
        '</div>',
        unsafe_allow_html=True,
    )
    try:
        inventory = _load_fga_inventory()
    except AuditLogError as exc:
        st.error(str(exc))
        return

    policies = inventory["policies"]
    catalog = inventory["catalog"]
    policy_names = sorted(
        {
            str(item.get("policy_name") or "").strip()
            for item in (*policies, *catalog)
            if str(item.get("policy_name") or "").strip()
        }
    )
    object_names = sorted(
        {
            str(item.get("object_name") or "").strip()
            for item in (*policies, *catalog)
            if str(item.get("object_name") or "").strip()
        }
    )

    st.markdown(
        '<div class="kb-audit-heading">조회 조건</div>',
        unsafe_allow_html=True,
    )
    with st.container(key="poc4_fga_filters"):
        filter_policy, filter_object = st.columns(2)
        with filter_policy:
            selected_policy = st.selectbox(
                "FGA 정책",
                options=("", *policy_names),
                format_func=lambda value: "전체 정책" if not value else value,
                key="poc4_fga_policy_filter",
            )
        with filter_object:
            selected_object = st.selectbox(
                "감사 객체",
                options=("", *object_names),
                format_func=lambda value: "전체 객체" if not value else value,
                key="poc4_fga_object_filter",
            )
        filter_days, filter_limit, refresh_column = st.columns([1.5, 1, 0.8])
        with filter_days:
            days = st.slider(
                "조회 기간",
                min_value=1,
                max_value=90,
                value=7,
                format="%d일",
                key="poc4_fga_days",
            )
        with filter_limit:
            row_limit = st.number_input(
                "최대 건수",
                min_value=10,
                max_value=500,
                value=100,
                step=10,
                key="poc4_fga_row_limit",
            )
        with refresh_column:
            st.markdown(
                '<div style="height: 28px"></div>',
                unsafe_allow_html=True,
            )
            if st.button(
                "새로고침",
                icon=":material/refresh:",
                width="stretch",
                key="poc4_fga_refresh",
            ):
                _load_fga_inventory.clear()
                _load_fga_audit_events.clear()
                st.rerun()

    try:
        events = _load_fga_audit_events(
            int(days),
            int(row_limit),
            selected_policy,
            selected_object,
        )
    except AuditLogError as exc:
        st.error(str(exc))
        return

    success_count = sum(
        1 for item in events if int(item.get("return_code") or 0) == 0
    )
    failure_count = len(events) - success_count
    with st.container(key="poc4_fga_metrics"):
        metric_policy, metric_event, metric_success, metric_failure = st.columns(4)
        metric_policy.metric("등록 FGA 정책", len(policies))
        metric_event.metric("조회 이벤트", len(events))
        metric_success.metric("성공", success_count)
        metric_failure.metric("실패", failure_count)

    if not policies:
        if catalog:
            st.warning(
                "FGA 정책 카탈로그는 존재하지만 현재 DB에 활성 정책이 등록되어 있지 않습니다."
            )
        else:
            st.warning(
                f"현재 {AUDIT_SCHEMA} 스키마에 등록된 DBMS_FGA 정책이 없습니다."
            )

    st.markdown(
        '<div class="kb-audit-heading">정책 상태</div>'
        f'<div class="kb-audit-caption">현재 활성 정책 {len(policies)}건 · '
        '감사 대상 컬럼과 적용 조건을 확인합니다.</div>',
        unsafe_allow_html=True,
    )
    with st.container(key="poc4_fga_policy_panel"):
        with st.expander("FGA 정책 상태", expanded=True):
            if policies:
                st.dataframe(
                    [
                        {
                            "객체": str(item.get("object_name") or ""),
                            "정책": str(item.get("policy_name") or ""),
                            "감사 컬럼": str(
                                item.get("policy_column") or "전체"
                            ),
                            "조건": str(item.get("policy_text") or "항상"),
                            "활성": str(item.get("enabled") or ""),
                            "SELECT": str(item.get("sel") or ""),
                        }
                        for item in policies
                    ],
                    column_config={
                        "객체": st.column_config.TextColumn(width="medium"),
                        "정책": st.column_config.TextColumn(width="large"),
                        "감사 컬럼": st.column_config.TextColumn(width="large"),
                        "조건": st.column_config.TextColumn(width="large"),
                        "활성": st.column_config.TextColumn(width="small"),
                        "SELECT": st.column_config.TextColumn(width="small"),
                    },
                    hide_index=True,
                    width="stretch",
                    height=min(360, 72 + 36 * len(policies)),
                )
            elif catalog:
                st.dataframe(
                    [
                        {
                            "객체": str(item.get("object_name") or ""),
                            "정책": str(item.get("policy_name") or ""),
                            "대상 컬럼": str(item.get("column_name") or "전체"),
                            "카탈로그 상태": str(item.get("enabled_yn") or ""),
                            "설명": str(item.get("description") or ""),
                        }
                        for item in catalog
                    ],
                    hide_index=True,
                    width="stretch",
                    height=min(360, 72 + 36 * len(catalog)),
                )
            else:
                st.caption("등록된 FGA 정책 정보가 없습니다.")

    st.markdown(
        '<div class="kb-audit-heading">감사 이벤트</div>'
        '<div class="kb-audit-caption">최신 이벤트부터 표시합니다. '
        '성공 여부와 감사 컬럼, 실행 사용자를 먼저 확인하세요.</div>',
        unsafe_allow_html=True,
    )
    with st.container(key="poc4_fga_event_toolbar"):
        show_sql = st.toggle(
            "SQL 원문 표시",
            value=True,
            key="poc4_fga_show_sql",
        )
    if not events:
        st.info("선택한 조건에 해당하는 FGA 감사 이벤트가 없습니다.")
        return

    display_rows: list[dict[str, Any]] = []
    for event in events:
        return_code = int(event.get("return_code") or 0)
        actor = str(event.get("client_identifier") or "").strip()
        if not actor:
            actor = str(event.get("dbusername") or "")
        display_row: dict[str, Any] = {
            "발생시각(KST)": str(event.get("event_time") or ""),
            "정책": str(event.get("fga_policy_name") or ""),
            "감사 컬럼": str(event.get("audit_column") or "전체"),
            "사용자": actor,
            "DB 사용자": str(event.get("dbusername") or ""),
            "접속 호스트": str(event.get("userhost") or ""),
            "객체": (
                f"{event.get('object_schema')}.{event.get('object_name')}"
            ),
            "작업": str(event.get("action_name") or ""),
            "결과": "성공" if return_code == 0 else f"ORA-{return_code:05d}",
        }
        if show_sql:
            display_row["SQL 원문"] = " ".join(
                str(event.get("sql_text") or "").split()
            )
        display_rows.append(display_row)

    def initial_column_width(
        column_name: str,
        minimum: int,
        maximum: int,
    ) -> int:
        values = [column_name]
        values.extend(str(row.get(column_name) or "") for row in display_rows)
        text_units = max(
            sum(2 if ord(character) > 127 else 1 for character in value)
            for value in values
        )
        return max(minimum, min(maximum, 36 + text_units * 8))

    event_column_config: dict[str, Any] = {
        "발생시각(KST)": st.column_config.TextColumn(
            width=initial_column_width("발생시각(KST)", 180, 220)
        ),
        "정책": st.column_config.TextColumn(
            width=initial_column_width("정책", 180, 320)
        ),
        "감사 컬럼": st.column_config.TextColumn(
            width=initial_column_width("감사 컬럼", 150, 300)
        ),
        "사용자": st.column_config.TextColumn(
            width=initial_column_width("사용자", 110, 180)
        ),
        "DB 사용자": st.column_config.TextColumn(
            width=initial_column_width("DB 사용자", 120, 180)
        ),
        "접속 호스트": st.column_config.TextColumn(
            width=initial_column_width("접속 호스트", 150, 240)
        ),
        "객체": st.column_config.TextColumn(
            width=initial_column_width("객체", 180, 300)
        ),
        "작업": st.column_config.TextColumn(
            width=initial_column_width("작업", 90, 140)
        ),
        "결과": st.column_config.TextColumn(
            width=initial_column_width("결과", 90, 140)
        ),
    }
    if show_sql:
        event_column_config["SQL 원문"] = st.column_config.TextColumn(
            width=initial_column_width("SQL 원문", 420, 720)
        )
    with st.container(key="poc4_fga_event_panel"):
        st.dataframe(
            display_rows,
            column_config=event_column_config,
            hide_index=True,
            width="stretch",
            height=min(640, 104 + 38 * len(display_rows)),
        )


def _render_vpd_operations_tab() -> None:
    st.markdown(
        f"""
        <section aria-labelledby="kb-ops-title">
          <div class="kb-section-title input" id="kb-ops-title"
               role="heading" aria-level="3">
            RLS / CLS 설정 ( 오라클 VPD / 마스킹 )
          </div>
          <div class="kb-ops-lead">
            업무 사용자와 데이터 접근 기준을 관리하고,
            DB가 적용한 결과까지 한 흐름에서 확인합니다.
          </div>

          <h3 class="kb-ops-section-heading">보안 설정 업무 프로세스</h3>
          <div class="kb-ops-guide-copy">
            권한을 먼저 만들고 보호 대상을 연결한 뒤, 실제 사용자 토큰으로 결과를 검증합니다.
            권한은 토큰에 복사되지 않아 이후 변경도 다음 요청부터 반영됩니다.
          </div>
          <div class="kb-ops-flow">
            <div class="kb-ops-step">
              <div class="kb-ops-step-no">1</div>
              <strong>사용자·그룹</strong>
              <span>업무 대상을 등록합니다.</span>
            </div>
            <div class="kb-ops-step">
              <div class="kb-ops-step-no">2</div>
              <strong>역할</strong>
              <span>직접·그룹 역할을 부여합니다.</span>
            </div>
            <div class="kb-ops-step accent">
              <div class="kb-ops-step-no">3</div>
              <strong>접근 규칙</strong>
              <span>객체·행·컬럼 접근을 설정합니다.</span>
            </div>
            <div class="kb-ops-step accent">
              <div class="kb-ops-step-no">4</div>
              <strong>보호·연결</strong>
              <span>VPD·ORDS 대상을 확인합니다.</span>
            </div>
            <div class="kb-ops-step success">
              <div class="kb-ops-step-no">5</div>
              <strong>유효 권한·접근 검증</strong>
              <span>토큰으로 실제 결과를 확인합니다.</span>
            </div>
          </div>
          <div class="kb-ops-portal">
            <strong>보안 운영 포털</strong>
            <p>
              권한 등록, 접근 규칙 관리와 검증 결과 확인은 별도 운영 화면에서 수행합니다.
            </p>
            <a class="kb-ops-link" href="{VPD_OPERATIONS_URL}"
               target="_blank" rel="noopener noreferrer">
              권한 운영 화면 열기&nbsp;&#8599;
            </a>
          </div>
        </section>
        """,
        unsafe_allow_html=True,
    )


def _select_ai_execution_summary(payload: Any) -> dict[str, Any]:
    if not isinstance(payload, Mapping):
        return {
            "generated_sql": "",
            "execution_status": "UNKNOWN",
            "execution_succeeded": False,
            "result": {},
        }
    generated_sql = str(
        payload.get("generatedSql") or payload.get("generated_sql") or ""
    ).strip()
    status = str(payload.get("status") or "").strip().upper()
    execution = str(payload.get("execution") or "").strip().upper()
    execution_succeeded = (
        status == "SHOWSQL_AND_EXECUTED" and execution == "READ_ONLY_EXECUTED"
    )
    result = {
        key: payload.get(key)
        for key in (
            "status",
            "execution",
            "rowCount",
            "truncated",
            "columns",
            "items",
            "generatedSql",
        )
        if key in payload
    }
    return {
        "generated_sql": generated_sql,
        "execution_status": status or execution or "UNKNOWN",
        "execution_succeeded": execution_succeeded,
        "result": result,
    }


def _record_qa_history(
    *,
    store: QaHistoryStore | None,
    benchmark_question: QaQuestion | None,
    question_text: str,
    conversation_id: str,
    requested_by: str,
    model_profile: str,
    answer_text: str,
    mcp_result: Any,
    duration_ms: int,
) -> tuple[QaQuestion | None, QaJudgment | None]:
    if store is None:
        return benchmark_question, None
    try:
        history_question = benchmark_question or store.find_or_create_free_text_question(
            question_text
        )
        execution = _select_ai_execution_summary(mcp_result)
        judgment = evaluate_sql(
            benchmark_question,
            execution["generated_sql"],
            execution_succeeded=bool(execution["execution_succeeded"]),
            error_text=_bounded_json(mcp_result, max_chars=8000),
        )
        store.record_answer(
            question_id=int(history_question.question_id or 0),
            answer_kind="LIVE",
            conversation_id=conversation_id,
            requested_by=requested_by,
            model_profile=model_profile,
            generated_sql=execution["generated_sql"],
            answer_text=answer_text,
            result=execution["result"],
            execution_output=_bounded_json(mcp_result, max_chars=24_000),
            execution_status=str(execution["execution_status"]),
            judgment_status=judgment.status,
            judgment_reason=judgment.reason,
            duration_ms=duration_ms,
        )
        return history_question, judgment
    except QaHistoryStoreError as exc:
        st.warning(f"질답 이력 저장에 실패했습니다: {exc}")
        return benchmark_question, None


def _process_submitted_question(
    *,
    question: str,
    conversation_id: str,
    chat_page_key: str,
    query_progress_slot: Any,
    query_progress_notice_key: str,
    bearer_token: str,
    servers: list[McpServer],
    default_router_model_profile: str,
    selected_query_model_profile: str,
    mcp_cache_generation_key: str,
    limit: int,
    selected_token_preset: VpdTokenPreset | None,
    execution_mode_override: str,
    benchmark_question: QaQuestion | None,
    qa_history_store: QaHistoryStore | None,
) -> None:
    normalized_question = question.strip()
    if not normalized_question:
        query_progress_slot.warning("질문을 입력해 주세요.")
        return
    active_model_profile = (
        selected_query_model_profile.strip()
        or _complex_reasoning_model_profile(default_router_model_profile)
    )
    single_tool_query = ""
    process_started = perf_counter()
    execution_events: list[dict[str, Any]] = []

    with query_progress_slot.container():
        with st.chat_message("user"):
            st.markdown(normalized_question)
        processing_status = st.status("질의 진행 상황 · 준비 중", expanded=True)
        with processing_status:
            progress_bar = st.progress(0, text="질의 처리를 준비하고 있습니다.")

    def record_execution_event(
        *,
        percent: int,
        label: str,
        detail: str = "",
        elapsed_seconds: float | None = None,
    ) -> None:
        event: dict[str, Any] = {
            "percent": int(percent),
            "label": str(label),
            "total_elapsed_seconds": round(perf_counter() - process_started, 3),
        }
        if detail:
            event["detail"] = str(detail)
        if elapsed_seconds is not None:
            event["elapsed_seconds"] = round(float(elapsed_seconds), 3)
        execution_events.append(event)

    def refresh_progress(
        percent: int,
        label: str,
        completed_detail: str | None = None,
        elapsed_seconds: float | None = None,
    ) -> None:
        progress_bar.progress(percent, text=label)
        processing_status.update(
            label=f"질의 진행 상황 · {percent}% · {label}",
            state="running",
            expanded=True,
        )
        if completed_detail:
            processing_status.write(completed_detail)
            record_execution_event(
                percent=percent,
                label=label,
                detail=completed_detail,
                elapsed_seconds=elapsed_seconds,
            )

    try:
        with processing_status:
            refresh_progress(5, "대화 문맥을 불러오고 있습니다.")
            step_started = perf_counter()
            conversation = load_chat_context(conversation_id)
            step_elapsed = perf_counter() - step_started
            refresh_progress(
                15,
                "대화 문맥을 불러왔습니다.",
                f"대화 문맥 로드 완료 · {step_elapsed:.1f}s",
                elapsed_seconds=step_elapsed,
            )

            refresh_progress(20, "질문 문맥을 정리하고 있습니다.")
            step_started = perf_counter()
            standalone_question = resolve_standalone_question(
                question=normalized_question,
                messages=conversation,
                model_profile_key=active_model_profile,
            )
            step_elapsed = perf_counter() - step_started
            refresh_progress(
                30,
                "질문 문맥을 정리했습니다.",
                f"후속 질문 정리 완료 · {step_elapsed:.1f}s",
                elapsed_seconds=step_elapsed,
            )

            refresh_progress(35, "사용 가능한 MCP 도구를 확인하고 있습니다.")
            step_started = perf_counter()
            cache_token = _normalized_bearer(bearer_token)
            discovery_results, discovery_failures = cached_discover_enabled_server_tools(
                _mcp_server_cache_rows(servers),
                _token_fingerprint(cache_token),
                int(st.session_state.get(mcp_cache_generation_key, 0) or 0),
                bearer_token,
            )
            step_elapsed = perf_counter() - step_started
            refresh_progress(
                50,
                "MCP 도구 확인을 완료했습니다.",
                f"MCP 툴 디스커버리 완료 · {step_elapsed:.1f}s",
                elapsed_seconds=step_elapsed,
            )

            routed_tools = [
                RoutedMcpTool(server_id=result.server.server_id, tool=tool)
                for result in discovery_results
                for tool in result.tools
            ]
            if discovery_failures:
                st.warning(
                    "일부 MCP 서버 디스커버리 실패: "
                    + ", ".join(
                        f"{failure['server_id']}={failure['error']}"
                        for failure in discovery_failures
                    )
                )
            if not routed_tools:
                raise PublicMcpError("디스커버리된 MCP 툴이 없습니다.")

            refresh_progress(55, "MCP 실행 방식을 판단하고 있습니다.")
            step_started = perf_counter()
            execution_model_profile = active_model_profile
            execution_mode_plan = plan_mcp_execution_mode(
                question=standalone_question,
                routed_tools=routed_tools,
                model_profile_key=execution_model_profile,
                mode_override=execution_mode_override,
            )
            if (
                execution_mode_override == "auto"
                and execution_model_profile != default_router_model_profile
                and str(execution_mode_plan.get("reason") or "").startswith(
                    "실행 방식 판단 실패"
                )
            ):
                execution_mode_plan = plan_mcp_execution_mode(
                    question=standalone_question,
                    routed_tools=routed_tools,
                    model_profile_key=default_router_model_profile,
                    mode_override=execution_mode_override,
                )
            execution_mode = str(execution_mode_plan.get("mode") or "single")
            is_complex_execution = execution_mode == "agent" and len(routed_tools) > 1
            reasoning_model_profile = active_model_profile
            route_key = str(execution_mode_plan.get("route_key") or "")
            _, routes_by_key = _agent_tool_catalog(routed_tools)
            selected_route = routes_by_key.get(route_key) or _default_single_route(
                routed_tools
            )
            if execution_mode_override == "single" and len(routed_tools) > 1:
                selected_route = route_mcp_tool_across_servers_with_llm(
                    routed_tools,
                    standalone_question,
                    router_model_profile=execution_model_profile,
                )
            step_elapsed = perf_counter() - step_started
            execution_mode_detail = (
                "실행 방식 판단 완료 · "
                f"{execution_mode} · "
                f"{execution_mode_plan.get('reason') or 'reason 없음'} · "
                f"{step_elapsed:.1f}s"
            )
            processing_status.write(execution_mode_detail)
            record_execution_event(
                percent=55,
                label="실행 방식 판단 완료",
                detail=execution_mode_detail,
                elapsed_seconds=step_elapsed,
            )

            agent_steps: list[Mapping[str, Any]] = []
            agent_stop_reason = ""
            if is_complex_execution:
                refresh_progress(60, "멀티툴 Agent가 MCP 실행 계획을 세우고 있습니다.")
                step_started = perf_counter()

                def on_agent_step(step: Mapping[str, Any]) -> None:
                    step_elapsed = float(step.get("elapsed_seconds") or 0)
                    agent_detail = (
                        f"Agent step {step.get('step')} · "
                        f"{step.get('server_id')}/{step.get('tool_name')} · "
                        f"{step_elapsed:.1f}s"
                    )
                    processing_status.write(agent_detail)
                    record_execution_event(
                        percent=60,
                        label=f"Agent step {step.get('step')}",
                        detail=agent_detail,
                        elapsed_seconds=step_elapsed,
                    )

                agent_result = run_mcp_agent_loop(
                    question=standalone_question,
                    conversation=conversation,
                    routed_tools=routed_tools,
                    servers=servers,
                    bearer_token=bearer_token,
                    limit=int(limit),
                    model_profile_key=reasoning_model_profile,
                    progress_callback=on_agent_step,
                )
                selected_server = agent_result["server"]
                selected = agent_result["tool"]
                arguments = agent_result["arguments"]
                answer_source = agent_result["mcp_result"]
                agent_steps = list(agent_result["agent_steps"])
                agent_stop_reason = str(agent_result.get("stop_reason") or "")
                step_elapsed = perf_counter() - step_started
                refresh_progress(
                    85,
                    "Agent MCP 실행을 완료했습니다.",
                    f"Agent loop 완료 · {len(agent_steps)} step · "
                    f"{step_elapsed:.1f}s",
                    elapsed_seconds=step_elapsed,
                )
            else:
                refresh_progress(65, "MCP 단일 호출을 실행하고 있습니다.")
                step_started = perf_counter()
                selected_server = next(
                    server
                    for server in servers
                    if server.server_id == selected_route.server_id
                )
                selected = selected_route.tool
                tool_query = prepare_tool_query_for_mcp(
                    question=standalone_question,
                    server=selected_server,
                    tool=selected,
                    model_profile_key=reasoning_model_profile,
                    selected_user_id=(
                        selected_token_preset.user_id
                        if selected_token_preset is not None
                        else ""
                    ),
                    selected_user_role=(
                        selected_token_preset.role
                        if selected_token_preset is not None
                        else ""
                    ),
                    selected_user_channel=(
                        selected_token_preset.channel
                        if selected_token_preset is not None
                        else ""
                    ),
                    selected_user_scope=(
                        selected_token_preset.scope
                        if selected_token_preset is not None
                        else ""
                    ),
                )
                single_tool_query = tool_query
                arguments = build_mcp_tool_arguments(
                    selected,
                    tool_query,
                    int(limit),
                    preferred_tool=selected_server.default_tool,
                )
                raw_result = call_tool(
                    base_url=selected_server.endpoint_url,
                    bearer_token=bearer_token,
                    tool=selected,
                    arguments=arguments,
                )
                parsed = _content_text_json(raw_result)
                answer_source = parsed if parsed is not None else raw_result
                step_elapsed = perf_counter() - step_started
                refresh_progress(
                    85,
                    "MCP 단일 호출을 완료했습니다.",
                    f"MCP 1회 호출 완료 · {selected_server.server_id}/{selected.name} · "
                    f"{step_elapsed:.1f}s",
                    elapsed_seconds=step_elapsed,
                )
    except (PublicMcpError, McpToolRouterError) as exc:
        security_evidence = collect_security_evidence(
            normalized_question,
            selected_token_preset,
            failure_message=str(exc),
        )
        business_evidence = collect_business_evidence(
            normalized_question,
            selected_token_preset,
        )
        security_verified = bool(
            security_evidence and not security_evidence.get("evidence_error")
        )
        business_verified = bool(
            business_evidence
            and business_evidence.get("evidence_complete")
            and not business_evidence.get("evidence_error")
        )
        if security_verified or business_verified:
            verified_answer = _enrich_answer_with_security_evidence(
                "업무 MCP 호출은 완료되지 않았지만 검증 근거는 정상 확인되었습니다.",
                security_evidence,
            )
            verified_answer = _enrich_answer_with_business_evidence(
                verified_answer,
                business_evidence,
            )
            verified_basis: list[str] = []
            if security_verified:
                verified_basis.append(
                    "보안 검증 증적: "
                    f"{security_evidence.get('audit_source')} "
                    f"LOG_ID={security_evidence.get('audit_log_id')}"
                )
            if business_verified:
                verified_basis.append(
                    "업무 검증 증적: "
                    + ", ".join(
                        str(source)
                        for source in business_evidence.get("evidence_sources", [])
                    )
                )
            save_chat_turn(
                conversation_id=conversation_id,
                selected_user_id=(
                    selected_token_preset.user_id
                    if selected_token_preset is not None
                    else ""
                ),
                selected_user_label=(
                    selected_token_preset.display_label
                    if selected_token_preset is not None
                    else "직접 입력"
                ),
                question=normalized_question,
                standalone_question=locals().get(
                    "standalone_question", normalized_question
                ),
                answer=verified_answer,
                basis=verified_basis,
                limitations="",
                details={
                    "server_id": "verified_evidence",
                    "tool_name": str(
                        security_evidence.get("evidence_type") or ""
                    )
                    or str(business_evidence.get("evidence_type") or ""),
                    "standalone_question": locals().get(
                        "standalone_question", normalized_question
                    ),
                    "security_evidence": security_evidence,
                    "business_evidence": business_evidence,
                    "mcp_error": str(exc),
                    "execution_events": execution_events,
                },
            )
            progress_bar.progress(100, text="검증 결과를 확인했습니다.")
            processing_status.update(
                label="질의 진행 상황 · 100% · 근거 검증 완료",
                state="complete",
                expanded=False,
            )
            st.session_state[query_progress_notice_key] = (
                "업무 MCP 오류와 별개로 토큰 범위 검증 결과를 확인했습니다."
            )
            st.session_state[chat_page_key] = 0
            st.rerun()
        progress_bar.empty()
        processing_status.update(label="질문 처리 실패", state="error", expanded=True)
        processing_status.error(str(exc))
        return

    security_evidence = collect_security_evidence(
        normalized_question,
        selected_token_preset,
    )
    business_evidence = collect_business_evidence(
        normalized_question,
        selected_token_preset,
    )
    evidence_verified_for_answer = bool(
        business_evidence
        and business_evidence.get("evidence_complete")
        and not business_evidence.get("evidence_error")
    ) or bool(
        security_evidence
        and security_evidence.get("evidence_type")
        and not security_evidence.get("evidence_error")
    )
    assistant_message: dict[str, Any]
    synthesis_error_details: dict[str, Any] = {}
    answer_model_profile = reasoning_model_profile
    answer_synthesis_fallback_model = ""
    try:
        with processing_status:
            refresh_progress(90, "질의 결과를 정리하고 있습니다.")
            step_started = perf_counter()
            synthesized = synthesize_answer(
                question=normalized_question,
                conversation=conversation,
                server=selected_server,
                tool=selected,
                arguments=arguments,
                mcp_result=answer_source,
                model_profile_key=reasoning_model_profile,
                agent_steps=agent_steps,
                security_evidence=security_evidence,
                business_evidence=business_evidence,
            )
            step_elapsed = perf_counter() - step_started
            refresh_progress(
                96,
                "질의 결과 정리를 완료했습니다.",
                f"최종 답변 생성 완료 · {step_elapsed:.1f}s",
                elapsed_seconds=step_elapsed,
            )
        assistant_message = {
            "role": "assistant",
            "content": str(synthesized["answer"]),
            "basis": synthesized.get("basis", []),
            "limitations": str(synthesized.get("limitations") or ""),
        }
    except AnswerSynthesisError as exc:
        synthesis_error_details = getattr(exc, "diagnostics", {}) or {}
        fallback_profile = DEFAULT_SYNTHESIS_FALLBACK_MODEL_PROFILE
        if fallback_profile and fallback_profile != reasoning_model_profile:
            try:
                with processing_status:
                    refresh_progress(
                        93,
                        f"최종 답변을 {fallback_profile}로 재시도하고 있습니다.",
                    )
                    step_started = perf_counter()
                    synthesized = synthesize_answer(
                        question=normalized_question,
                        conversation=conversation,
                        server=selected_server,
                        tool=selected,
                        arguments=arguments,
                        mcp_result=answer_source,
                        model_profile_key=fallback_profile,
                        agent_steps=agent_steps,
                        security_evidence=security_evidence,
                        business_evidence=business_evidence,
                    )
                    step_elapsed = perf_counter() - step_started
                    refresh_progress(
                        96,
                        "질의 결과 정리를 완료했습니다.",
                        f"fallback 답변 생성 완료 · {fallback_profile} · "
                        f"{step_elapsed:.1f}s",
                        elapsed_seconds=step_elapsed,
                    )
                answer_model_profile = fallback_profile
                answer_synthesis_fallback_model = fallback_profile
                assistant_message = {
                    "role": "assistant",
                    "content": str(synthesized["answer"]),
                    "basis": synthesized.get("basis", []),
                    "limitations": str(synthesized.get("limitations") or ""),
                }
            except AnswerSynthesisError as fallback_exc:
                synthesis_error_details = {
                    "primary": synthesis_error_details,
                    "fallback": getattr(fallback_exc, "diagnostics", {}) or {},
                }
                fallback_content = fallback_answer_from_mcp(answer_source, agent_steps)
                fallback_content = _enrich_answer_with_security_evidence(
                    fallback_content,
                    security_evidence,
                )
                fallback_content = _enrich_answer_with_business_evidence(
                    fallback_content,
                    business_evidence,
                )
                assistant_message = {
                    "role": "assistant",
                    "content": fallback_content,
                    "basis": [],
                    "limitations": (
                        ""
                        if evidence_verified_for_answer
                        else f"{fallback_exc} MCP 원본 응답은 상세에서 확인 가능합니다."
                    ),
                }
                refresh_progress(96, "원본 MCP 응답으로 결과를 구성했습니다.")
        else:
            fallback_content = fallback_answer_from_mcp(answer_source, agent_steps)
            fallback_content = _enrich_answer_with_security_evidence(
                fallback_content,
                security_evidence,
            )
            fallback_content = _enrich_answer_with_business_evidence(
                fallback_content,
                business_evidence,
            )
            assistant_message = {
                "role": "assistant",
                "content": fallback_content,
                "basis": [],
                "limitations": (
                    ""
                    if evidence_verified_for_answer
                    else f"{exc} MCP 원본 응답은 상세에서 확인 가능합니다."
                ),
            }
            refresh_progress(96, "원본 MCP 응답으로 결과를 구성했습니다.")

    pre_save_elapsed = perf_counter() - process_started
    record_execution_event(
        percent=98,
        label="질의 결과 저장 준비 완료",
        detail=f"저장 전 처리 완료 · {pre_save_elapsed:.1f}s",
        elapsed_seconds=pre_save_elapsed,
    )
    assistant_message["details"] = {
        "server_id": selected_server.server_id,
        "tool_name": selected.name,
        "standalone_question": standalone_question,
        "tool_query": single_tool_query,
        "arguments": arguments,
        "mcp_result": answer_source,
        "execution_mode": str(execution_mode_plan.get("mode") or ""),
        "execution_mode_reason": str(execution_mode_plan.get("reason") or ""),
        "execution_mode_model_profile": str(
            execution_mode_plan.get("model_profile") or ""
        ),
        "reasoning_model_profile": reasoning_model_profile,
        "answer_model_profile": answer_model_profile,
        "answer_synthesis_fallback_model": answer_synthesis_fallback_model,
        "answer_synthesis_error": synthesis_error_details,
        "agent_steps": agent_steps,
        "agent_stop_reason": agent_stop_reason,
        "security_evidence": security_evidence,
        "business_evidence": business_evidence,
        "execution_events": execution_events,
        "discovered_routes": [
            {
                "server_id": discovery.server.server_id,
                "tool_name": tool.name,
                "readOnly": tool.read_only,
                "description": tool.description,
            }
            for discovery in discovery_results
            for tool in discovery.tools
        ],
        "discovery_failures": discovery_failures,
    }
    history_question, history_judgment = _record_qa_history(
        store=qa_history_store,
        benchmark_question=benchmark_question,
        question_text=normalized_question,
        conversation_id=conversation_id,
        requested_by=(
            selected_token_preset.user_id
            if selected_token_preset is not None
            else "portal-user"
        ),
        model_profile=answer_model_profile,
        answer_text=str(assistant_message["content"]),
        mcp_result=answer_source,
        duration_ms=round((perf_counter() - process_started) * 1000),
    )
    if history_judgment is not None:
        assistant_message["details"]["qa_history"] = {
            "question_id": history_question.question_id if history_question else None,
            "question_code": history_question.question_code if history_question else "",
            "judgment_status": history_judgment.status,
            "judgment_reason": history_judgment.reason,
        }
        st.info(
            "질답 검증 판정: "
            f"{_judgment_label(history_judgment.status)} · {history_judgment.reason}"
        )
    refresh_progress(98, "질의 결과를 저장하고 있습니다.")
    save_chat_turn(
        conversation_id=conversation_id,
        selected_user_id=(
            selected_token_preset.user_id if selected_token_preset is not None else ""
        ),
        selected_user_label=(
            selected_token_preset.display_label
            if selected_token_preset is not None
            else "직접 입력"
        ),
        question=normalized_question,
        standalone_question=standalone_question,
        answer=str(assistant_message["content"]),
        basis=assistant_message.get("basis", []),
        limitations=str(assistant_message.get("limitations") or ""),
        details=assistant_message["details"],
    )
    total_elapsed = perf_counter() - process_started
    refresh_progress(
        100,
        "질의 처리가 완료되었습니다.",
        f"전체 처리 완료 · {total_elapsed:.1f}s",
        elapsed_seconds=total_elapsed,
    )
    processing_status.update(
        label="질의 진행 상황 · 100% · 처리 완료",
        state="complete",
        expanded=False,
    )
    st.session_state[query_progress_notice_key] = (
        "질의 처리가 완료되었습니다. "
        f"사용 경로: {selected_server.server_id} / {selected.name} · "
        f"{'agent step ' + str(len(agent_steps)) + '회' if agent_steps else 'MCP 1회 호출'}"
    )
    st.session_state[chat_page_key] = 0
    st.rerun()


def main() -> None:
    try:
        profile = load_app_profile(APP_PROFILE_FILE, ENV_FILE)
    except AppProfileError as exc:
        st.set_page_config(page_title="AI 업무 에이전트", layout="wide")
        st.error(str(exc))
        return
    st.set_page_config(
        page_title=profile.page_title, page_icon=profile.page_icon, layout="wide"
    )
    _apply_console_theme(profile)
    _restore_portal_session_from_cookie()
    if not st.session_state.get(PORTAL_AUTHENTICATED_KEY, False):
        _render_portal_login(profile)
        return
    qa_questions, qa_history_store, qa_store_error = _load_qa_questions()
    question_text_key = "poc4_mcp_discovery_question_text"
    loaded_qa_question_key = "poc4_mcp_discovery_loaded_qa_question_code"
    conversation_id_key = "poc4_mcp_discovery_conversation_id"
    chat_page_key = "poc4_mcp_discovery_page"
    query_progress_notice_key = "poc4_query_progress_notice"
    mcp_cache_generation_key = "poc4_mcp_tools_cache_generation"
    init_chat_store()
    if conversation_id_key not in st.session_state:
        st.session_state[conversation_id_key] = new_conversation_id()
    if chat_page_key not in st.session_state:
        st.session_state[chat_page_key] = 0
    if mcp_cache_generation_key not in st.session_state:
        st.session_state[mcp_cache_generation_key] = 0
    conversation_id = str(st.session_state[conversation_id_key])

    try:
        servers, default_server_index = load_mcp_servers()
    except PublicMcpError as exc:
        st.error(str(exc))
        return
    try:
        token_presets = load_vpd_token_presets()
    except PublicMcpError as exc:
        st.error(str(exc))
        return

    default_router_model_profile = servers[default_server_index].router_model_profile
    configured_mcp_bearer = _runtime_env_value(
        servers[default_server_index].auth_token_env
    )
    query_model_profile_key = "poc4_query_model_profile"
    default_query_model_profile = DEFAULT_QUERY_MODEL_PROFILE
    try:
        model_profile_options = _model_profile_select_options()
    except ValueError:
        model_profile_options = (
            (default_query_model_profile, default_query_model_profile),
        )
    model_profile_labels = dict(model_profile_options)
    model_profile_keys = tuple(model_profile_labels)
    if default_query_model_profile not in model_profile_keys:
        default_query_model_profile = (
            default_router_model_profile
            if default_router_model_profile in model_profile_keys
            else model_profile_keys[0]
        )
    if st.session_state.get(query_model_profile_key) not in model_profile_keys:
        st.session_state[query_model_profile_key] = default_query_model_profile
    _render_app_header(profile)
    with st.sidebar:
        st.caption(
            f"포털 사용자 · {st.session_state.get(PORTAL_AUTH_USER_KEY, '')}"
        )
        if st.button("로그아웃", use_container_width=True):
            _logout_portal()
        st.divider()
        st.markdown('<div class="kb-panel-title">AI 사용자 설정</div>', unsafe_allow_html=True)
        selected_query_model_profile = st.selectbox(
            "LLM 모델",
            options=model_profile_keys,
            key=query_model_profile_key,
            format_func=lambda value: (
                f"{model_profile_labels.get(value, value)} · {value}"
            ),
            help=(
                "선택한 모델을 후속 질문 정리, 실행 방식 판단, MCP tool 라우팅, "
                "Agent planning, 최종 답변 합성에 사용합니다."
            ),
        )
        if token_presets:
            vpd_user_key = "poc4_vpd_token_preset"
            token_preset_by_id = {
                preset.user_id: preset for preset in token_presets
            }
            token_preset_ids = tuple(token_preset_by_id)
            stored_vpd_user = st.session_state.get(vpd_user_key)
            if vpd_user_key not in st.session_state:
                st.session_state[vpd_user_key] = _default_vpd_user_id(token_presets)
            elif isinstance(stored_vpd_user, VpdTokenPreset):
                st.session_state[vpd_user_key] = stored_vpd_user.user_id
            elif stored_vpd_user not in token_preset_by_id:
                st.session_state[vpd_user_key] = _default_vpd_user_id(token_presets)

            current_vpd_user_id = st.session_state[vpd_user_key]
            current_vpd_user_label = token_preset_by_id[
                current_vpd_user_id
            ].select_label
            st.markdown(
                '<div class="kb-sidebar-section-title">VPD 사용자</div>',
                unsafe_allow_html=True,
            )
            vpd_popover_key = "poc4_vpd_user_popover_open"

            def close_vpd_user_popover() -> None:
                st.session_state[vpd_popover_key] = False

            with st.popover(
                f"**{current_vpd_user_label}**",
                use_container_width=True,
                key=vpd_popover_key,
                on_change="rerun",
            ):
                selected_vpd_user_id = st.radio(
                    "VPD 사용자 선택",
                    options=token_preset_ids,
                    key=vpd_user_key,
                    format_func=lambda user_id: token_preset_by_id[
                        user_id
                    ].select_label,
                    label_visibility="collapsed",
                    on_change=close_vpd_user_popover,
                )
            selected_token_preset = (
                token_preset_by_id.get(selected_vpd_user_id)
                if selected_vpd_user_id is not None
                else None
            )
        else:
            selected_token_preset = None
        if configured_mcp_bearer:
            # The Smilegate MCP gateway uses a server-side credential.  UI-only
            # context must never be forwarded as a second bearer token.
            st.caption("MCP 인증: 서버 관리 토큰 적용")
            manual_bearer_token = ""
        elif selected_token_preset is not None:
            manual_bearer_token = st.text_input(
                "Bearer token",
                value=selected_token_preset.token,
                type="default",
                disabled=True,
                key=f"poc4_bearer_token_{selected_token_preset.user_id}",
                help="선택한 VPD 사용자의 Bearer Token입니다.",
            )
        else:
            manual_bearer_token = st.text_input(
                "Bearer token",
                type="default",
                key="poc4_manual_bearer_token",
            )
        bearer_token = configured_mcp_bearer or (
            selected_token_preset.token
            if selected_token_preset is not None
            else manual_bearer_token
        )
        if selected_token_preset is not None:
            _render_vpd_user_card(selected_token_preset)

        all_conversation_rows = list_conversations(limit=200)
        current_row = next(
            (
                row
                for row in all_conversation_rows
                if row["conversation_id"] == conversation_id
            ),
            {
                "conversation_id": conversation_id,
                "created_at": "",
                "updated_at": "",
                "title": "",
                "turn_count": 0,
                "latest_question": "",
            },
        )
        conversation_search = st.text_input(
            "대화 검색",
            placeholder="질문, 제목, 세션 ID",
            key="poc4_conversation_search",
        )
        conversation_rows = filter_conversations(
            all_conversation_rows,
            conversation_search,
        )
        search_query = conversation_search.strip()
        current_in_results = any(
            row["conversation_id"] == conversation_id for row in conversation_rows
        )
        if search_query and conversation_rows and not current_in_results:
            st.session_state[conversation_id_key] = str(
                conversation_rows[0]["conversation_id"]
            )
            st.session_state[chat_page_key] = 0
            st.rerun()
        if search_query and not conversation_rows:
            st.info("검색 결과가 없습니다.")
        if not search_query and not current_in_results:
            conversation_rows.insert(0, current_row)
        conversation_ids = [str(row["conversation_id"]) for row in conversation_rows]
        conversation_by_id = {
            str(row["conversation_id"]): row for row in conversation_rows
        }
        if conversation_ids:
            selected_conversation_id = st.selectbox(
                "대화 세션",
                options=conversation_ids,
                index=(
                    conversation_ids.index(conversation_id)
                    if conversation_id in conversation_ids
                    else 0
                ),
                format_func=lambda value: conversation_label(conversation_by_id[value]),
            )
        else:
            st.caption("저장된 대화가 없습니다.")
            selected_conversation_id = conversation_id
            conversation_by_id = {conversation_id: current_row}
        if selected_conversation_id != conversation_id:
            st.session_state[conversation_id_key] = selected_conversation_id
            st.session_state[chat_page_key] = 0
            st.rerun()
        if st.button("새 대화 시작"):
            st.session_state[conversation_id_key] = new_conversation_id()
            st.session_state[chat_page_key] = 0
            st.rerun()

        with st.expander("대화 관리"):
            selected_row = conversation_by_id.get(conversation_id, current_row)
            title_key = f"poc4_conversation_title_{conversation_id}"
            title_value = st.text_input(
                "대화 이름",
                value=str(selected_row.get("title") or ""),
                key=title_key,
                max_chars=80,
            )
            if st.button("대화 이름 저장"):
                rename_conversation(conversation_id, title_value)
                st.rerun()
            st.download_button(
                "대화 JSON 다운로드",
                data=json.dumps(
                    {
                        "conversation_id": conversation_id,
                        "title": title_value.strip(),
                        "turns": load_all_chat_turns(conversation_id),
                    },
                    ensure_ascii=False,
                    indent=2,
                    default=str,
                ),
                file_name=f"{conversation_id}.json",
                mime="application/json",
            )
            delete_confirmed = st.checkbox("현재 대화 삭제 확인")
            if st.button("현재 대화 삭제", disabled=not delete_confirmed):
                delete_conversation(conversation_id)
                st.session_state[conversation_id_key] = new_conversation_id()
                st.session_state[chat_page_key] = 0
                st.rerun()

        with st.expander("MCP 고급 설정", expanded=False):
            st.markdown(
                '<div class="kb-sidebar-section-title">MCP Servers</div>',
                unsafe_allow_html=True,
            )
            st.caption(", ".join(server.server_id for server in servers))
            if st.button("MCP 툴 새로고침"):
                st.session_state[mcp_cache_generation_key] = (
                    int(st.session_state.get(mcp_cache_generation_key, 0) or 0) + 1
                )
                st.toast("MCP tools/list 캐시를 새로고침합니다.")
            st.caption(
                f"tools/list cache generation: "
                f"{int(st.session_state.get(mcp_cache_generation_key, 0) or 0)}"
            )
            limit = st.number_input(
                "limit",
                min_value=1,
                max_value=1000,
                value=50,
            )
            execution_mode_override = st.selectbox(
                "MCP 실행 모드",
                options=("auto", "single", "agent"),
                index=0,
                format_func=lambda value: {
                    "auto": "자동 · 작은 모델이 단일/멀티툴 판단",
                    "single": "단일 · MCP 1회 호출",
                    "agent": "멀티툴 Agent · ReAct loop",
                }[value],
                help=(
                    "속도 우선이면 자동 또는 단일을 사용하세요. "
                    "자동 모드는 선택한 LLM 모델로 실행 방식을 판단합니다."
                ),
            )
            st.markdown("**Resolved MCP endpoints**")
            for server in servers:
                st.caption(f"{server.server_id}: {server.endpoint_url}")
            if st.checkbox("MCP 설정 JSON 보기"):
                try:
                    st.json(
                        json.loads(MCP_SERVERS_FILE.read_text(encoding="utf-8"))
                    )
                except (OSError, UnicodeError, ValueError):
                    st.warning("MCP 설정 JSON을 읽지 못했습니다.")
            st.caption(f"config: {MCP_SERVERS_FILE}")
            st.caption(f"QA benchmark: {QA_BENCHMARK_FILE}")
            st.caption(f"token presets: {VPD_TOKEN_PRESETS_FILE}")
            st.caption(f"selected LLM model: {selected_query_model_profile}")
            st.caption(f"default model: {default_query_model_profile}")
            st.caption(
                "synthesis fallback model: "
                f"{DEFAULT_SYNTHESIS_FALLBACK_MODEL_PROFILE}"
            )

    st.markdown(
        '<div id="kb-main-tabs-anchor" style="scroll-margin-top: 0.75rem;"></div>',
        unsafe_allow_html=True,
    )
    architecture_tab, scenario_tab = st.tabs(["아키텍처", "시나리오"])
    with architecture_tab:
        _render_architecture_tab()

    with scenario_tab:
        selected_qa_question = _render_qa_benchmark_panel(
            qa_questions,
            qa_history_store,
            qa_store_error,
        )
        if question_text_key not in st.session_state:
            st.session_state[question_text_key] = DEFAULT_QUESTION
        selected_qa_code = (
            selected_qa_question.question_code if selected_qa_question is not None else ""
        )
        if st.session_state.get(loaded_qa_question_key, "") != selected_qa_code:
            st.session_state[question_text_key] = (
                selected_qa_question.question_text
                if selected_qa_question is not None
                else DEFAULT_QUESTION
            )
            st.session_state[loaded_qa_question_key] = selected_qa_code

        st.markdown(
            '<div class="kb-section-title input" role="heading" aria-level="3">'
            "질문 실행"
            "</div>",
            unsafe_allow_html=True,
        )
        if selected_qa_question is not None:
            st.caption(
                "선택한 후보 문장을 수정하지 않고 실행하면 자동 판정합니다. 문장을 수정하면 자유 질의로 저장되어 수동 검토 대상이 됩니다."
            )
        else:
            st.caption("자유 질의는 실행 이력으로 남기며, 고객 기준 정답과 자동 비교하지 않습니다.")

        with st.form("poc4_mcp_question_form"):
            question = st.text_area(
                "질문",
                label_visibility="collapsed",
                key=question_text_key,
                height=120,
                max_chars=1_000,
                placeholder="고객 검증 후보를 선택하거나 업무 질문을 직접 입력하세요.",
            )
            submitted = st.form_submit_button(
                "질문 전송",
                type="primary",
                icon=":material/send:",
                icon_position="right",
                width="stretch",
            )

        if submitted:
            components.html(
                """
                <script>
                  const target = window.parent.document.getElementById(
                    "kb-main-tabs-anchor"
                  );
                  if (target) {
                    window.parent.requestAnimationFrame(() => {
                      target.scrollIntoView({ behavior: "smooth", block: "start" });
                    });
                  }
                </script>
                """,
                height=0,
            )

        st.markdown(
            '<div id="kb-query-results-anchor" '
            'style="scroll-margin-top: 0.75rem;"></div>',
            unsafe_allow_html=True,
        )
        query_progress_slot = st.empty()
        query_progress_notice = st.session_state.pop(
            query_progress_notice_key,
            None,
        )
        if query_progress_notice:
            query_progress_slot.success(str(query_progress_notice))

        _render_chat_history(conversation_id, chat_page_key)

        if query_progress_notice:
            components.html(
                """
                <script>
                  window.parent.setTimeout(() => {
                    const target = window.parent.document.getElementById(
                      "kb-query-results-anchor"
                    );
                    if (target) {
                      target.scrollIntoView({ behavior: "smooth", block: "start" });
                    }
                  }, 250);
                </script>
                """,
                height=0,
            )

    if not submitted:
        return

    with scenario_tab:
        _process_submitted_question(
            question=question,
            conversation_id=conversation_id,
            chat_page_key=chat_page_key,
            query_progress_slot=query_progress_slot,
            query_progress_notice_key=query_progress_notice_key,
            bearer_token=bearer_token,
            servers=servers,
            default_router_model_profile=default_router_model_profile,
            selected_query_model_profile=selected_query_model_profile,
            mcp_cache_generation_key=mcp_cache_generation_key,
            limit=int(limit),
            selected_token_preset=selected_token_preset,
            execution_mode_override=execution_mode_override,
            benchmark_question=(
                selected_qa_question
                if selected_qa_question is not None
                and question.strip() == selected_qa_question.question_text.strip()
                else None
            ),
            qa_history_store=qa_history_store,
        )


if __name__ == "__main__":
    main()
