Import PoC4 MCP test UI source snapshot

poc4_active_source_20260714/src/mcp_tool_router.py

@@ -0,0 +1,190 @@
+"""LLM-based MCP tool routing.
+
+The router receives only the user question and the discovered MCP tool
+descriptors.  It never receives MCP bearer tokens or provider credentials.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+from typing import Any, Mapping
+
+from src.oci_genai_sdk import (
+    build_oci_genai_completion_client,
+    temperature_for_model_profile,
+)
+from src.poc3.model_registry import resolve_model_profile
+
+
+@dataclass(frozen=True)
+class McpTool:
+    name: str
+    description: str
+    schema: Mapping[str, Any]
+    read_only: bool
+
+
+@dataclass(frozen=True)
+class RoutedMcpTool:
+    server_id: str
+    tool: McpTool
+
+
+class McpToolRouterError(RuntimeError):
+    """Safe routing error. Must not contain secrets or provider traces."""
+
+
+def route_mcp_tool_across_servers_with_llm(
+    tools: list[RoutedMcpTool],
+    question: str,
+    *,
+    router_model_profile: str,
+) -> RoutedMcpTool:
+    """Select one discovered MCP server/tool pair with OCI GenAI."""
+
+    candidates = list(tools)
+    if not candidates:
+        raise McpToolRouterError("라우팅 가능한 MCP tool이 없습니다.")
+
+    by_key = {
+        "{}::{}".format(candidate.server_id, candidate.tool.name): candidate
+        for candidate in candidates
+    }
+    route_keys = list(by_key)
+    try:
+        profile = resolve_model_profile(router_model_profile)
+        client = build_oci_genai_completion_client(
+            profile.model_id,
+            profile.answer_model_region,
+            profile.answer_model_endpoint,
+        )
+        tool_catalog = [
+            {
+                "route_key": "{}::{}".format(candidate.server_id, candidate.tool.name),
+                "server_id": candidate.server_id,
+                "tool_name": candidate.tool.name,
+                "description": candidate.tool.description[:1000],
+                "input_properties": sorted(
+                    (
+                        candidate.tool.schema.get("properties", {})
+                        if isinstance(
+                            candidate.tool.schema.get("properties"), Mapping
+                        )
+                        else {}
+                    ).keys()
+                ),
+                "read_only": candidate.tool.read_only,
+            }
+            for candidate in candidates
+        ]
+        text = client.complete(
+            system_prompt=(
+                "You are an MCP server and tool router. Choose exactly one "
+                "server/tool route for the user question from the discovered "
+                "routes. Return only JSON that matches the schema. Never "
+                "request or expose bearer tokens. Do not invent server ids or "
+                "tool names."
+            ),
+            user_prompt=json.dumps(
+                {
+                    "question": question,
+                    "routes": tool_catalog,
+                },
+                ensure_ascii=False,
+            ),
+            response_schema={
+                "type": "object",
+                "additionalProperties": False,
+                "required": ["route_key"],
+                "properties": {
+                    "route_key": {
+                        "type": "string",
+                        "enum": route_keys,
+                    }
+                },
+            },
+            max_tokens=256,
+            temperature=temperature_for_model_profile(profile),
+        )
+        routed = json.loads(text)
+    except Exception:
+        raise McpToolRouterError("LLM tool router 호출에 실패했습니다.") from None
+
+    if not isinstance(routed, Mapping):
+        raise McpToolRouterError("LLM tool router 응답 형식이 올바르지 않습니다.")
+    selected_key = str(routed.get("route_key") or "").strip()
+    selected = by_key.get(selected_key)
+    if selected is None:
+        raise McpToolRouterError("LLM tool router가 허용되지 않은 route를 선택했습니다.")
+    return selected
+
+
+def route_mcp_tool_with_llm(
+    tools: list[McpTool],
+    question: str,
+    *,
+    preferred_tool: str,
+    tool_allowlist: tuple[str, ...],
+    router_model_profile: str,
+) -> McpTool:
+    """Backward-compatible single-server routing helper."""
+
+    candidates = [
+        tool for tool in tools if not tool_allowlist or tool.name in tool_allowlist
+    ]
+    routed = route_mcp_tool_across_servers_with_llm(
+        [RoutedMcpTool(server_id="default", tool=tool) for tool in candidates],
+        question,
+        router_model_profile=router_model_profile,
+    )
+    return routed.tool
+
+
+def build_mcp_tool_arguments(
+    tool: McpTool,
+    question: str,
+    limit: int,
+    *,
+    preferred_tool: str,
+) -> dict[str, Any]:
+    """Build bounded tool arguments from the selected tool schema."""
+
+    properties = tool.schema.get("properties")
+    if not isinstance(properties, Mapping):
+        properties = {}
+
+    if tool.name == preferred_tool:
+        return {"prompt": question, "limit": limit}
+    if "prompt" in properties:
+        args: dict[str, Any] = {"prompt": question}
+        if "limit" in properties:
+            args["limit"] = limit
+        elif "max_rows" in properties:
+            args["max_rows"] = limit
+        return args
+    if "question" in properties:
+        args = {"question": question}
+        if "max_rows" in properties:
+            args["max_rows"] = limit
+        elif "limit" in properties:
+            args["limit"] = limit
+        return args
+    if "query" in properties:
+        args = {"query": question}
+        if "max_evidence" in properties:
+            args["max_evidence"] = min(limit, 10)
+        elif "limit" in properties:
+            args["limit"] = limit
+        return args
+    return {"prompt": question, "limit": limit}
+
+
+__all__ = [
+    "McpTool",
+    "McpToolRouterError",
+    "RoutedMcpTool",
+    "build_mcp_tool_arguments",
+    "route_mcp_tool_across_servers_with_llm",
+    "route_mcp_tool_with_llm",
+]

poc4_active_source_20260714/src/oci_genai_sdk.py

@@ -0,0 +1,274 @@
+"""Common OCI Generative AI chat-completion SDK boundary.
+
+This module is intentionally small: callers provide a validated model route
+and a JSON schema, and this boundary performs one OCI GenAI chat call.  It
+does not know about MCP, Streamlit, business payloads, or bearer tokens.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from functools import lru_cache
+import os
+from pathlib import Path
+import re
+from typing import Dict, Optional, Protocol
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DOTENV_PATH = ROOT / ".env"
+ALLOWED_OCI_SETTINGS = frozenset(
+    {
+        "OCI_AUTH_TYPE",
+        "OCI_CONFIG_FILE",
+        "OCI_GENAI_COMPARTMENT_ID",
+        "OCI_PROFILE",
+    }
+)
+_COMPARTMENT_ID = re.compile(r"^ocid1\.compartment\.[A-Za-z0-9._-]+$")
+
+
+class CompletionClient(Protocol):
+    """Minimal completion client contract shared by app layers."""
+
+    def complete(
+        self,
+        system_prompt: str,
+        user_prompt: str,
+        response_schema: Mapping[str, object],
+        max_tokens: int,
+        temperature: Optional[float],
+    ) -> str:
+        """Return the assistant message text."""
+
+
+def read_allowed_dotenv(path: Optional[Path] = None) -> Dict[str, str]:
+    """Read only non-secret OCI routing/auth-mode settings from .env."""
+
+    selected_path = DOTENV_PATH if path is None else path
+    try:
+        lines = selected_path.read_text(encoding="utf-8").splitlines()
+    except OSError:
+        return {}
+    values: Dict[str, str] = {}
+    for raw_line in lines:
+        line = raw_line.strip()
+        if not line or line.startswith("#"):
+            continue
+        if line.startswith("export "):
+            line = line[7:].lstrip()
+        key, separator, raw_value = line.partition("=")
+        key = key.strip()
+        if not separator or key not in ALLOWED_OCI_SETTINGS:
+            continue
+        value = raw_value.strip()
+        if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
+            value = value[1:-1]
+        if "\x00" not in value and "\n" not in value and "\r" not in value:
+            values[key] = value
+    return values
+
+
+@dataclass(frozen=True, repr=False)
+class OCISettings:
+    auth_type: str
+    config_file: str
+    profile: str
+    compartment_id: str
+
+
+def load_oci_settings() -> OCISettings:
+    """Resolve OCI GenAI settings from safe .env keys and environment."""
+
+    values = read_allowed_dotenv()
+    for key in ALLOWED_OCI_SETTINGS:
+        value = os.environ.get(key)
+        if isinstance(value, str) and value.strip():
+            values[key] = value.strip()
+
+    auth_type = values.get("OCI_AUTH_TYPE", "config_file").strip().casefold()
+    auth_type = auth_type.replace("-", "_")
+    if auth_type in {"api_key", "config", "config_file"}:
+        auth_type = "config_file"
+    elif auth_type not in {"instance_principal", "resource_principal"}:
+        raise ValueError("unsupported OCI authentication mode")
+
+    compartment_id = values.get("OCI_GENAI_COMPARTMENT_ID", "").strip()
+    if not _COMPARTMENT_ID.fullmatch(compartment_id):
+        raise ValueError("OCI Generative AI compartment is not configured")
+    return OCISettings(
+        auth_type=auth_type,
+        config_file=values.get("OCI_CONFIG_FILE", "~/.oci/config").strip(),
+        profile=values.get("OCI_PROFILE", "DEFAULT").strip() or "DEFAULT",
+        compartment_id=compartment_id,
+    )
+
+
+class OCICompletionClient:
+    """Minimal OCI GenericChatRequest adapter."""
+
+    def __init__(
+        self,
+        settings: OCISettings,
+        model_id: str,
+        region: str,
+        endpoint: str,
+    ) -> None:
+        try:
+            import oci
+            from oci.generative_ai_inference import GenerativeAiInferenceClient
+        except (ImportError, AttributeError):
+            raise RuntimeError("OCI SDK is unavailable") from None
+
+        kwargs: Dict[str, object] = {}
+        if settings.auth_type == "config_file":
+            try:
+                config = oci.config.from_file(
+                    file_location=os.path.expandvars(
+                        os.path.expanduser(settings.config_file)
+                    ),
+                    profile_name=settings.profile,
+                )
+            except Exception:
+                raise RuntimeError("OCI SDK configuration is unavailable") from None
+            config["region"] = region
+        elif settings.auth_type == "instance_principal":
+            try:
+                signer = oci.auth.signers.InstancePrincipalsSecurityTokenSigner()
+            except Exception:
+                raise RuntimeError("OCI signer is unavailable") from None
+            config = {"region": region}
+            kwargs["signer"] = signer
+        else:
+            try:
+                signer = oci.auth.signers.get_resource_principals_signer()
+            except Exception:
+                raise RuntimeError("OCI signer is unavailable") from None
+            config = {"region": region}
+            kwargs["signer"] = signer
+        if not config.get("region"):
+            raise RuntimeError("OCI region is unavailable")
+        kwargs["service_endpoint"] = endpoint
+        try:
+            self._client = GenerativeAiInferenceClient(config, **kwargs)
+        except Exception:
+            raise RuntimeError("OCI Generative AI client is unavailable") from None
+        self._compartment_id = settings.compartment_id
+        self._model_id = model_id
+
+    def complete(
+        self,
+        system_prompt: str,
+        user_prompt: str,
+        response_schema: Mapping[str, object],
+        max_tokens: int,
+        temperature: Optional[float],
+    ) -> str:
+        try:
+            from oci.generative_ai_inference.models import (
+                ChatDetails,
+                GenericChatRequest,
+                JsonSchemaResponseFormat,
+                OnDemandServingMode,
+                ResponseJsonSchema,
+                SystemMessage,
+                TextContent,
+                UserMessage,
+            )
+
+            schema = ResponseJsonSchema(
+                name="oci_genai_json_response",
+                description="Strict JSON response",
+                schema=dict(response_schema),
+                is_strict=True,
+            )
+            request_options: Dict[str, object] = {
+                "api_format": "GENERIC",
+                "messages": [
+                    SystemMessage(content=[TextContent(text=system_prompt)]),
+                    UserMessage(content=[TextContent(text=user_prompt)]),
+                ],
+                "max_completion_tokens": max_tokens,
+                "is_stream": False,
+                "response_format": JsonSchemaResponseFormat(json_schema=schema),
+            }
+            if temperature is not None:
+                request_options["temperature"] = temperature
+            request = GenericChatRequest(**request_options)
+            details = ChatDetails(
+                compartment_id=self._compartment_id,
+                serving_mode=OnDemandServingMode(model_id=self._model_id),
+                chat_request=request,
+            )
+            response = self._client.chat(details)
+            data = getattr(response, "data", None)
+            chat_response = getattr(data, "chat_response", None)
+            choices = getattr(chat_response, "choices", None)
+            if not isinstance(choices, Sequence) or not choices:
+                raise RuntimeError("OCI response has no choice")
+            message = getattr(choices[0], "message", None)
+            content = getattr(message, "content", None)
+            if not isinstance(content, Sequence) or isinstance(
+                content, (str, bytes, bytearray)
+            ) or not content:
+                raise RuntimeError("OCI response has no content")
+            text = getattr(content[0], "text", None)
+            if not isinstance(text, str):
+                raise RuntimeError("OCI response content is invalid")
+            return text
+        except Exception:
+            raise RuntimeError("OCI GenAI completion call failed") from None
+
+
+def build_oci_genai_completion_client(
+    model_id: str,
+    region: str,
+    endpoint: str,
+) -> CompletionClient:
+    """Build a completion client for one validated model route."""
+
+    return _cached_oci_genai_completion_client(
+        load_oci_settings(),
+        model_id,
+        region,
+        endpoint,
+    )
+
+
+@lru_cache(maxsize=16)
+def _cached_oci_genai_completion_client(
+    settings: OCISettings,
+    model_id: str,
+    region: str,
+    endpoint: str,
+) -> CompletionClient:
+    """Reuse OCI GenAI clients within one Python process."""
+
+    return OCICompletionClient(settings, model_id, region, endpoint)
+
+
+def temperature_for_model_key(model_key: object) -> Optional[float]:
+    """Return provider-compatible temperature for one registered model key."""
+
+    key = str(model_key or "").strip().lower()
+    if key.startswith("gpt"):
+        return None
+    return 0.1
+
+
+def temperature_for_model_profile(profile: object) -> Optional[float]:
+    return temperature_for_model_key(getattr(profile, "model_key", profile))
+
+
+__all__ = [
+    "ALLOWED_OCI_SETTINGS",
+    "CompletionClient",
+    "OCICompletionClient",
+    "OCISettings",
+    "build_oci_genai_completion_client",
+    "load_oci_settings",
+    "read_allowed_dotenv",
+    "temperature_for_model_key",
+    "temperature_for_model_profile",
+]

poc4_active_source_20260714/src/poc3/model_registry.py

@@ -0,0 +1,434 @@
+"""8512/8513 전용 PoC_3 model profile registry.
+
+이 registry는 모델 metadata만 관리한다. ``provider=oci``는 모델의 출처를 뜻하며
+``POC3_MCP_PROVIDER``와 독립적이다. 따라서 기본 model profile이 GPT-5.5여도 현재
+MCP 실행 경로는 계속 ``mock``일 수 있다.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field, replace
+import json
+import os
+from pathlib import Path
+import re
+from typing import Any, Mapping, Optional
+
+
+ROOT = Path(__file__).resolve().parents[2]
+REGISTRY_PATH = ROOT / "config" / "poc3_model_profiles.json"
+DEFAULT_MODEL_PROFILE_KEY = "gpt55_oci"
+MODEL_PROFILE_ENV = "POC3_MODEL_PROFILE"
+MODEL_PROFILE_DEFAULT_ENV = "POC3_MODEL_PROFILE_DEFAULT"
+EXISTING_MODEL_PROFILE_KEYS = ("grok43", "llama4_maverick", "llama33_70b")
+MODEL_PROFILE_ALIASES = {
+    "gpt54_mini": "gpt54_mini_oci",
+    "llama33": "llama33_70b",
+}
+EXPECTED_SOURCE_TAG = "poc_2-gpt55-oci-partial"
+EXPECTED_SOURCE_COMMIT = "7a3b37f175b65ed5eab1d8bf37c9bf6114e7558f"
+_EXPECTED_ANSWER_MODEL_ROUTES = {
+    "gpt55_oci": (
+        "openai.gpt-5.5",
+        "us-chicago-1",
+        "OPENAI_GPT_5_5_CHAT",
+        "OCI_REGIONAL_DEFAULT",
+    ),
+    "gpt54_mini_oci": (
+        "openai.gpt-5.4-mini",
+        "us-chicago-1",
+        "OPENAI_GPT_5_4_MINI_CHAT",
+        "OCI_REGIONAL_DEFAULT",
+    ),
+    "grok43": (
+        "xai.grok-4.3",
+        "us-chicago-1",
+        "XAI_GROK_4_3_CHAT",
+        "OCI_REGIONAL_DEFAULT",
+    ),
+    "llama4_maverick": (
+        "meta.llama-4-maverick-17b-128e-instruct-fp8",
+        "us-chicago-1",
+        "META_LLAMA_4_MAVERICK_CHAT",
+        "OCI_REGIONAL_DEFAULT",
+    ),
+    "llama33_70b": (
+        "meta.llama-3.3-70b-instruct",
+        "us-chicago-1",
+        "META_LLAMA_3_3_70B_CHAT",
+        "OCI_REGIONAL_DEFAULT",
+    ),
+}
+
+_MODEL_KEY = re.compile(r"^[a-z][a-z0-9_]{1,63}$")
+_MODEL_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{1,127}$")
+_ANSWER_MODEL_ID_ALIAS = re.compile(r"^[A-Z][A-Z0-9_]{1,127}$")
+_OCI_REGION = re.compile(r"^[a-z]{2}-[a-z0-9-]+-[1-9][0-9]*$")
+_OCI_REGIONAL_ENDPOINT = re.compile(
+    r"^https://inference\.generativeai\."
+    r"(?P<region>[a-z]{2}-[a-z0-9-]+-[1-9][0-9]*)\.oci\.oraclecloud\.com$"
+)
+_ORACLE_IDENTIFIER = re.compile(r"^[A-Z][A-Z0-9_$#]{0,127}$")
+_SOURCE_TAG = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{1,127}$")
+_VERIFICATION_STATUSES = frozenset(
+    {"VERIFIED", "PARTIAL_VERIFIED", "VERIFIED_WITH_WARNINGS"}
+)
+_REQUIRED_PROFILE_FIELDS = (
+    "model_key",
+    "display_name",
+    "provider",
+    "model_id",
+    "answer_model_id_alias",
+    "answer_model_region",
+    "answer_model_endpoint_mode",
+    "poc2_select_ai_profile",
+    "poc2_native_agent_team",
+    "verification_status",
+    "default_for_poc3",
+    "source_tag",
+    "notes",
+)
+
+_PROFILE_ROUTE_ENV_KEYS = {
+    "gpt55_oci": (
+        "POC3_LLM_GPT55_OCI_MODEL_ID",
+        "POC3_LLM_GPT55_OCI_REGION",
+        "POC3_LLM_GPT55_OCI_ENDPOINT",
+    ),
+    "gpt54_mini_oci": (
+        "POC3_LLM_GPT54_MINI_OCI_MODEL_ID",
+        "POC3_LLM_GPT54_MINI_OCI_REGION",
+        "POC3_LLM_GPT54_MINI_OCI_ENDPOINT",
+    ),
+    "grok43": (
+        "POC3_LLM_GROK43_MODEL_ID",
+        "POC3_LLM_GROK43_REGION",
+        "POC3_LLM_GROK43_ENDPOINT",
+    ),
+    "llama4_maverick": (
+        "POC3_LLM_LLAMA4_MAVERICK_MODEL_ID",
+        "POC3_LLM_LLAMA4_MAVERICK_REGION",
+        "POC3_LLM_LLAMA4_MAVERICK_ENDPOINT",
+    ),
+    "llama33_70b": (
+        "POC3_LLM_LLAMA33_70B_MODEL_ID",
+        "POC3_LLM_LLAMA33_70B_REGION",
+        "POC3_LLM_LLAMA33_70B_ENDPOINT",
+    ),
+}
+
+
+def _regional_endpoint(region: str) -> str:
+    return "https://inference.generativeai.%s.oci.oraclecloud.com" % region
+
+
+def _endpoint_host_alias(region: str) -> str:
+    return "OCI_GENAI_INFERENCE_%s" % region.upper().replace("-", "_")
+
+
+def _env_override(
+    environ: Mapping[str, str],
+    key: str,
+    default: str,
+) -> str:
+    if key not in environ:
+        return default
+    value = environ.get(key)
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError("answer model route override is invalid")
+    return value.strip()
+
+
+@dataclass(frozen=True)
+class ModelProfile:
+    """UI와 workflow가 공유하는 비밀값 없는 model metadata."""
+
+    model_key: str
+    display_name: str
+    provider: str
+    model_id: str
+    answer_model_id_alias: str
+    answer_model_region: str
+    answer_model_endpoint_mode: str
+    answer_model_endpoint: str = field(repr=False)
+    poc2_select_ai_profile: str
+    poc2_native_agent_team: str
+    verification_status: str
+    default_for_poc3: bool
+    source_tag: str
+    notes: str
+    display_order: int = 999
+
+    @classmethod
+    def from_mapping(cls, value: Mapping[str, Any]) -> "ModelProfile":
+        missing = [name for name in _REQUIRED_PROFILE_FIELDS if name not in value]
+        if missing:
+            raise ValueError("PoC_3 model profile fields are missing")
+        if not isinstance(value.get("default_for_poc3"), bool):
+            raise ValueError("default_for_poc3 must be boolean")
+        order = value.get("display_order", 999)
+        if isinstance(order, bool) or not isinstance(order, int) or order < 0:
+            raise ValueError("model profile display_order is invalid")
+
+        answer_model_region = str(value["answer_model_region"]).strip().lower()
+        profile = cls(
+            model_key=str(value["model_key"]).strip().lower(),
+            display_name=str(value["display_name"]).strip(),
+            provider=str(value["provider"]).strip().lower(),
+            model_id=str(value["model_id"]).strip(),
+            answer_model_id_alias=str(value["answer_model_id_alias"])
+            .strip()
+            .upper(),
+            answer_model_region=answer_model_region,
+            answer_model_endpoint_mode=str(
+                value["answer_model_endpoint_mode"]
+            ).strip().upper(),
+            answer_model_endpoint=_regional_endpoint(answer_model_region),
+            poc2_select_ai_profile=str(value["poc2_select_ai_profile"])
+            .strip()
+            .upper(),
+            poc2_native_agent_team=str(value["poc2_native_agent_team"])
+            .strip()
+            .upper(),
+            verification_status=str(value["verification_status"]).strip().upper(),
+            default_for_poc3=value["default_for_poc3"],
+            source_tag=str(value["source_tag"]).strip(),
+            notes=str(value["notes"]).strip(),
+            display_order=order,
+        )
+        if not _MODEL_KEY.fullmatch(profile.model_key):
+            raise ValueError("model profile key is invalid")
+        if not profile.display_name or len(profile.display_name) > 128:
+            raise ValueError("model profile display name is invalid")
+        if profile.provider != "oci":
+            raise ValueError("unsupported model provider")
+        if not _MODEL_ID.fullmatch(profile.model_id):
+            raise ValueError("model id is invalid")
+        if not _ANSWER_MODEL_ID_ALIAS.fullmatch(profile.answer_model_id_alias):
+            raise ValueError("answer model id alias is invalid")
+        if not _OCI_REGION.fullmatch(profile.answer_model_region):
+            raise ValueError("answer model region is invalid")
+        if profile.answer_model_endpoint_mode != "OCI_REGIONAL_DEFAULT":
+            raise ValueError("answer model endpoint mode is invalid")
+        endpoint_match = _OCI_REGIONAL_ENDPOINT.fullmatch(
+            profile.answer_model_endpoint
+        )
+        if (
+            endpoint_match is None
+            or endpoint_match.group("region") != profile.answer_model_region
+        ):
+            raise ValueError("answer model endpoint is invalid")
+        if not _ORACLE_IDENTIFIER.fullmatch(profile.poc2_select_ai_profile):
+            raise ValueError("PoC_2 Select AI profile mapping is invalid")
+        if not _ORACLE_IDENTIFIER.fullmatch(profile.poc2_native_agent_team):
+            raise ValueError("PoC_2 Native Agent team mapping is invalid")
+        if profile.verification_status not in _VERIFICATION_STATUSES:
+            raise ValueError("model verification status is invalid")
+        if not _SOURCE_TAG.fullmatch(profile.source_tag):
+            raise ValueError("model profile source tag is invalid")
+        if not profile.notes or len(profile.notes) > 1_000:
+            raise ValueError("model profile notes are invalid")
+        return profile
+
+    def with_answer_route_overrides(
+        self,
+        environ: Mapping[str, str],
+    ) -> "ModelProfile":
+        """Apply only this profile's validated, non-secret OCI route settings."""
+
+        keys = _PROFILE_ROUTE_ENV_KEYS.get(self.model_key)
+        if keys is None:
+            raise ValueError("answer model route is not registered")
+        model_id = _env_override(environ, keys[0], self.model_id)
+        region = _env_override(environ, keys[1], self.answer_model_region).lower()
+        endpoint = _env_override(
+            environ,
+            keys[2],
+            _regional_endpoint(region),
+        )
+        if not _MODEL_ID.fullmatch(model_id):
+            raise ValueError("answer model route override is invalid")
+        if not _OCI_REGION.fullmatch(region):
+            raise ValueError("answer model route override is invalid")
+        endpoint_match = _OCI_REGIONAL_ENDPOINT.fullmatch(endpoint)
+        if endpoint_match is None or endpoint_match.group("region") != region:
+            raise ValueError("answer model route override is invalid")
+        return replace(
+            self,
+            model_id=model_id,
+            answer_model_region=region,
+            answer_model_endpoint=endpoint,
+        )
+
+    def public_metadata(self) -> dict[str, object]:
+        """System Details에 투영 가능한 비밀값 없는 metadata를 반환한다."""
+
+        return {
+            "model_key": self.model_key,
+            "display_name": self.display_name,
+            "provider": self.provider,
+            "answer_model_id_alias": self.answer_model_id_alias,
+            "answer_model_region": self.answer_model_region,
+            "answer_model_endpoint_mode": self.answer_model_endpoint_mode,
+            "answer_model_endpoint_host_alias": _endpoint_host_alias(
+                self.answer_model_region
+            ),
+            "poc2_select_ai_profile": self.poc2_select_ai_profile,
+            "poc2_native_agent_team": self.poc2_native_agent_team,
+            "verification_status": self.verification_status,
+            "default_for_poc3": self.default_for_poc3,
+            "source_tag": self.source_tag,
+            "notes": self.notes,
+        }
+
+
+@dataclass(frozen=True)
+class ModelProfileRegistry:
+    """검증된 PoC_3 model profile 집합."""
+
+    profiles: tuple[ModelProfile, ...]
+    default_model_profile: str
+    registry_name: str
+    source_commit: str
+    schema_version: int = 1
+
+    def by_key(self, model_key: object) -> ModelProfile:
+        candidate = str(model_key or "").strip().lower()
+        candidate = MODEL_PROFILE_ALIASES.get(candidate, candidate)
+        for profile in self.profiles:
+            if profile.model_key == candidate:
+                return profile
+        # 사용자 입력이나 환경변수 원문을 오류에 반사하지 않는다.
+        raise ValueError("model profile is not registered")
+
+    @property
+    def default_profile(self) -> ModelProfile:
+        return self.by_key(self.default_model_profile)
+
+    def selector_options(self) -> tuple[ModelProfile, ...]:
+        return tuple(sorted(self.profiles, key=lambda item: item.display_order))
+
+    def resolve(
+        self,
+        requested: object = None,
+        *,
+        environ: Optional[Mapping[str, str]] = None,
+    ) -> ModelProfile:
+        """명시 요청은 엄격히 검증하고 환경 기본값은 안전하게 fallback한다."""
+
+        source = os.environ if environ is None else environ
+        if requested is not None and str(requested).strip():
+            return self.by_key(requested).with_answer_route_overrides(source)
+
+        for name in (MODEL_PROFILE_ENV, MODEL_PROFILE_DEFAULT_ENV):
+            candidate = source.get(name)
+            if not candidate or not candidate.strip():
+                continue
+            try:
+                profile = self.by_key(candidate)
+            except ValueError:
+                continue
+            return profile.with_answer_route_overrides(source)
+        return self.default_profile.with_answer_route_overrides(source)
+
+
+def load_model_registry(path: Path = REGISTRY_PATH) -> ModelProfileRegistry:
+    """JSON registry를 매 호출마다 검증해 반환한다."""
+
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise ValueError("PoC_3 model profile registry cannot be loaded") from exc
+    if not isinstance(payload, Mapping):
+        raise ValueError("PoC_3 model profile registry must be an object")
+    raw_profiles = payload.get("profiles")
+    if not isinstance(raw_profiles, list) or not raw_profiles:
+        raise ValueError("PoC_3 model profile registry has no profiles")
+    profiles = tuple(
+        ModelProfile.from_mapping(item)
+        for item in raw_profiles
+        if isinstance(item, Mapping)
+    )
+    if len(profiles) != len(raw_profiles):
+        raise ValueError("PoC_3 model profile registry contains an invalid profile")
+    keys = tuple(item.model_key for item in profiles)
+    if len(set(keys)) != len(keys):
+        raise ValueError("PoC_3 model profile keys must be unique")
+    if len({item.display_name for item in profiles}) != len(profiles):
+        raise ValueError("PoC_3 model profile display names must be unique")
+    defaults = tuple(item.model_key for item in profiles if item.default_for_poc3)
+    configured_default = str(payload.get("default_model_profile") or "").strip().lower()
+    if defaults != (configured_default,):
+        raise ValueError("PoC_3 model profile default is inconsistent")
+    if configured_default != DEFAULT_MODEL_PROFILE_KEY:
+        raise ValueError("PoC_3 GPT-5.5 default contract is not satisfied")
+    if not set(EXISTING_MODEL_PROFILE_KEYS).issubset(keys):
+        raise ValueError("existing PoC_3 selector models are missing")
+    actual_answer_routes = {
+        item.model_key: (
+            item.model_id,
+            item.answer_model_region,
+            item.answer_model_id_alias,
+            item.answer_model_endpoint_mode,
+        )
+        for item in profiles
+    }
+    if actual_answer_routes != _EXPECTED_ANSWER_MODEL_ROUTES:
+        raise ValueError("PoC_3 answer model route mapping is inconsistent")
+    if str(payload.get("source_commit") or "").strip() != EXPECTED_SOURCE_COMMIT:
+        raise ValueError("PoC_2 source commit is inconsistent")
+    if any(item.source_tag != EXPECTED_SOURCE_TAG for item in profiles):
+        raise ValueError("PoC_2 source tag is inconsistent")
+    if payload.get("schema_version") != 1:
+        raise ValueError("unsupported PoC_3 model profile registry schema")
+    registry_name = str(payload.get("registry_name") or "").strip()
+    if not registry_name:
+        raise ValueError("PoC_3 model profile registry name is missing")
+    return ModelProfileRegistry(
+        profiles=profiles,
+        default_model_profile=configured_default,
+        registry_name=registry_name,
+        source_commit=EXPECTED_SOURCE_COMMIT,
+    )
+
+
+def resolve_model_profile(
+    requested: object = None,
+    *,
+    environ: Optional[Mapping[str, str]] = None,
+) -> ModelProfile:
+    return load_model_registry().resolve(requested, environ=environ)
+
+
+def resolve_model_profile_key(
+    requested: object = None,
+    *,
+    environ: Optional[Mapping[str, str]] = None,
+) -> str:
+    return resolve_model_profile(requested, environ=environ).model_key
+
+
+def is_registered_model_profile(value: object) -> bool:
+    try:
+        load_model_registry().by_key(value)
+    except ValueError:
+        return False
+    return True
+
+
+__all__ = [
+    "DEFAULT_MODEL_PROFILE_KEY",
+    "EXISTING_MODEL_PROFILE_KEYS",
+    "EXPECTED_SOURCE_COMMIT",
+    "EXPECTED_SOURCE_TAG",
+    "MODEL_PROFILE_DEFAULT_ENV",
+    "MODEL_PROFILE_ENV",
+    "MODEL_PROFILE_ALIASES",
+    "ModelProfile",
+    "ModelProfileRegistry",
+    "REGISTRY_PATH",
+    "is_registered_model_profile",
+    "load_model_registry",
+    "resolve_model_profile",
+    "resolve_model_profile_key",
+]

poc4_active_source_20260714/src/poc3/questions.py

@@ -0,0 +1,264 @@
+"""PoC_3 preset catalog와 현재 질문 기반의 결정적 intent router."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import re
+import unicodedata
+
+
+QUESTION_CATEGORIES = ("STRUCTURED", "RAG", "HYBRID")
+GENERIC_RAG_QUESTION_ID = "R0"
+MAX_QUESTION_CHARS = 2_000
+
+
+@dataclass(frozen=True)
+class DemoQuestion:
+    """ID와 분류가 고정된 데모 질문."""
+
+    question_id: str
+    category: str
+    text: str
+
+    def __post_init__(self) -> None:
+        if self.category not in QUESTION_CATEGORIES:
+            raise ValueError("unsupported demo question category")
+        if not self.question_id or not self.text.strip():
+            raise ValueError("demo question id/text is required")
+
+
+@dataclass(frozen=True)
+class ResolvedQuestionIntent:
+    """현재 질문만으로 결정된 실행 intent.
+
+    ``question_id``는 local route/fixture를 설명하는 분류 label이다. 8500 provider
+    입력으로 전달되지 않으며 UI에서 선택한 scenario ID도 이 모델에 들어오지
+    않는다.
+    """
+
+    question_id: str
+    category: str
+
+    def __post_init__(self) -> None:
+        if self.category not in QUESTION_CATEGORIES:
+            raise ValueError("unsupported resolved question category")
+        if self.question_id not in {
+            GENERIC_RAG_QUESTION_ID,
+            "S1",
+            "S2",
+            "S3",
+            "S4",
+            "S5",
+            "R1",
+            "R2",
+            "H1",
+            "H2",
+            "H3",
+            "H4",
+        }:
+            raise ValueError("unsupported resolved question id")
+
+
+COMMON_DEMO_QUESTIONS = (
+    DemoQuestion("S1", "STRUCTURED", "상품별 계약 건수를 보여줘."),
+    DemoQuestion("S2", "STRUCTURED", "총 지급보험금이 가장 큰 상품은?"),
+    DemoQuestion(
+        "S3",
+        "STRUCTURED",
+        "숫자로 계산 가능한 평균 보장금액이 높은 상품 10개를 보여줘.",
+    ),
+    DemoQuestion("S4", "STRUCTURED", "고객 등급별 평균 보험료를 보여줘."),
+    DemoQuestion("S5", "STRUCTURED", "이해관계자 역할별 인원 수를 보여줘."),
+    DemoQuestion("R1", "RAG", "자동차보험 약관의 면책 사항은?"),
+    DemoQuestion("R2", "RAG", "보험금 청구 시 필요한 서류는?"),
+    DemoQuestion(
+        "H1",
+        "HYBRID",
+        "보험금이 가장 큰 상품의 주요 면책 조항을 알려줘.",
+    ),
+    DemoQuestion(
+        "H2",
+        "HYBRID",
+        "청구가 많은 상품군의 보장 제외 조건을 알려줘.",
+    ),
+    DemoQuestion(
+        "H3",
+        "HYBRID",
+        "고객 등급별 보험료 수준을 보고, 관련 약관상 유의해야 할 보장 제외 조건도 함께 알려줘.",
+    ),
+)
+
+_QUESTION_BY_ID = {item.question_id: item for item in COMMON_DEMO_QUESTIONS}
+if len(_QUESTION_BY_ID) != len(COMMON_DEMO_QUESTIONS):
+    raise RuntimeError("duplicate PoC_3 demo question id")
+
+
+def question_by_id(question_id: str) -> DemoQuestion:
+    """정규화된 ID로 질문을 찾되, 알 수 없는 ID는 거부한다."""
+
+    normalized = str(question_id).strip().upper()
+    try:
+        return _QUESTION_BY_ID[normalized]
+    except KeyError:
+        raise ValueError("unknown PoC_3 demo question id") from None
+
+
+def normalize_scenario_id(value: object) -> str | None:
+    """선택적인 preset ID를 정규화하되 실행 routing에는 관여하지 않는다."""
+
+    if value is None:
+        return None
+    if not isinstance(value, str):
+        raise ValueError("scenario id must be a string")
+    normalized = value.strip().upper()
+    if not normalized:
+        return None
+    question_by_id(normalized)
+    return normalized
+
+
+def _normalized_question(question: object) -> tuple[str, str]:
+    if not isinstance(question, str):
+        raise ValueError("question must be a non-empty string")
+    if len(question) > MAX_QUESTION_CHARS:
+        raise ValueError("question exceeds the supported length")
+    normalized = re.sub(
+        r"\s+", " ", unicodedata.normalize("NFKC", question).strip().lower()
+    )
+    if not normalized:
+        raise ValueError("question must be a non-empty string")
+    return normalized, normalized.replace(" ", "")
+
+
+def _contains_any(text: str, candidates: tuple[str, ...]) -> bool:
+    return any(candidate in text for candidate in candidates)
+
+
+def resolve_question_intent(question: object) -> ResolvedQuestionIntent:
+    """현재 질문 텍스트만으로 S/R/H route intent를 결정한다.
+
+    명확한 structured/hybrid intent에 해당하지 않는 질문은 임의로 추측하지 않고
+    ``R0`` generic RAG 검색으로 보낸다. 이 함수는 scenario 또는
+    question ID hint를 받지 않으므로 preset metadata가 실행을 바꿀 수 없다.
+    """
+
+    normalized, compact = _normalized_question(question)
+
+    # Canonical preset은 기존 10문항 동작을 byte-for-byte 보존한다.
+    for item in COMMON_DEMO_QUESTIONS:
+        candidate, _ = _normalized_question(item.text)
+        if normalized == candidate:
+            return ResolvedQuestionIntent(item.question_id, item.category)
+
+    has_product = _contains_any(compact, ("상품", "상품군"))
+    has_exclusion = _contains_any(
+        compact,
+        ("면책", "보장제외", "제외조건", "보상하지않", "약관상유의"),
+    )
+    has_top = _contains_any(
+        compact, ("가장큰", "최대", "최고", "1위", "상위", "제일많", "높은")
+    )
+
+    # 자유 질문에서 자동차보험 보유 규모와 타사 대비 강점을 함께 요구하면
+    # generic structured 조회와 자유 evidence 검색을 로컬에서 합성하는 H4로
+    # 보낸다. Label은 routing metadata일 뿐 provider query ID가 아니다.
+    has_auto_insurance = "자동차보험" in compact
+    has_count = _contains_any(
+        compact,
+        (
+            "갯수",
+            "개수",
+            "건수",
+            "계약수",
+            "몇개",
+            "몇건",
+            "보유수",
+            "상품수",
+        ),
+    )
+    has_competitor_comparison = _contains_any(
+        compact, ("타사", "경쟁사", "다른회사", "타보험사")
+    ) and _contains_any(compact, ("강점", "장점", "차별", "우위", "비교"))
+    if has_auto_insurance and has_count and has_competitor_comparison:
+        return ResolvedQuestionIntent("H4", "HYBRID")
+
+    # Hybrid를 먼저 판별해 정형 키워드가 포함된 복합 질문이 S/R 단일 route로
+    # 축소되지 않도록 한다.
+    if (
+        _contains_any(compact, ("고객등급", "등급별"))
+        and "보험료" in compact
+        and has_exclusion
+    ):
+        return ResolvedQuestionIntent("H3", "HYBRID")
+    if (
+        "청구" in compact
+        and _contains_any(compact, ("많은", "빈도", "건수", "상위"))
+        and has_product
+        and has_exclusion
+    ):
+        return ResolvedQuestionIntent("H2", "HYBRID")
+    if (
+        has_product
+        and _contains_any(compact, ("지급보험금", "보험금"))
+        and has_top
+        and has_exclusion
+    ):
+        return ResolvedQuestionIntent("H1", "HYBRID")
+
+    if (
+        has_product
+        and "계약" in compact
+        and _contains_any(compact, ("건수", "계약수", "몇건", "집계"))
+    ):
+        return ResolvedQuestionIntent("S1", "STRUCTURED")
+    if (
+        has_product
+        and _contains_any(compact, ("지급보험금", "보험금총액", "총보험금"))
+        and has_top
+    ):
+        return ResolvedQuestionIntent("S2", "STRUCTURED")
+    if (
+        has_product
+        and _contains_any(compact, ("보장금액", "가입금액"))
+        and "평균" in compact
+        and has_top
+    ):
+        return ResolvedQuestionIntent("S3", "STRUCTURED")
+    if (
+        _contains_any(compact, ("고객등급", "등급별"))
+        and "보험료" in compact
+        and "평균" in compact
+    ):
+        return ResolvedQuestionIntent("S4", "STRUCTURED")
+    if (
+        "이해관계자" in compact
+        and "역할" in compact
+        and _contains_any(compact, ("인원", "사람수", "몇명", "명수", "수"))
+    ):
+        return ResolvedQuestionIntent("S5", "STRUCTURED")
+
+    if (
+        "자동차보험" in compact
+        and _contains_any(compact, ("면책", "보상하지않", "보장제외", "제외사항"))
+    ):
+        return ResolvedQuestionIntent("R1", "RAG")
+    if (
+        _contains_any(compact, ("보험금", "청구"))
+        and _contains_any(compact, ("서류", "문서", "증빙", "제출자료"))
+    ):
+        return ResolvedQuestionIntent("R2", "RAG")
+
+    return ResolvedQuestionIntent(GENERIC_RAG_QUESTION_ID, "RAG")
+
+
+__all__ = [
+    "COMMON_DEMO_QUESTIONS",
+    "DemoQuestion",
+    "GENERIC_RAG_QUESTION_ID",
+    "MAX_QUESTION_CHARS",
+    "QUESTION_CATEGORIES",
+    "ResolvedQuestionIntent",
+    "normalize_scenario_id",
+    "question_by_id",
+    "resolve_question_intent",
+]