shadow_brain_core/brain/core.py

import os
import json
import logging
import sys
import warnings
import time
import requests
import re
import base64
import subprocess
import hashlib
import hmac
import uuid
import secrets
import threading
import traceback
from dotenv import load_dotenv

try:
    from google import genai
    from google.genai import types
except ImportError:
    genai = None

from typing import Generator, Optional, List, Dict, Any
from .browser_controller import BrowserController
from .coding_tools import CodingToolkit, OPENAI_SCHEMAS as CODING_TOOL_SCHEMAS, CODING_AGENT_GUIDE
from .tool_runtime import ToolRuntime, RuntimeCtx
import sovereign_memory
import quota_manager
from quota_manager import ImperialQuotaManager, QuotaExceededError

# 🗂️ [Imperial Obsidian Directive] 옵시디언 의도 감지 시 시스템 지시에 주입.
# 약한 모델이 "곧 만들겠습니다"라고 말만 하고 obsidian_write를 호출하지 않던 문제를
# 강한 명령형 지시로 차단한다. (ask_user/coding 가이드와 동일 패턴)
_OBSIDIAN_DIRECTIVE = (
    "\n\n## 🗂️ Imperial Obsidian Directive (반드시 준수)\n"
    "사용자가 옵시디언에 노트/위키/문서/페이지를 만들거나(만들어/작성/정리/저장/기록) 찾아달라고 했다.\n"
    "1. **말로 약속하지 마라.** '곧 만들겠습니다/생성하겠습니다/작성하겠습니다'라고만 답하고 끝내는 것은 "
    "절대 금지 — 아무것도 만들어지지 않는 실패다. 반드시 **같은 턴에 도구를 실제로 호출**하라.\n"
    "2. 작성 요청이면 `obsidian_write(path, content, mode)`를 호출하라. content는 네가 직접 전체 "
    "마크다운(# 제목, [[백링크]], #태그, - 목록)을 써서 채운다. 플레이스홀더 금지.\n"
    "3. 여러 페이지 위키면 `obsidian_write`를 **여러 번** 호출하라(노트 1개당 1회). 인덱스 노트를 만들고 "
    "각 노트를 `[[ ]]`로 연결하라. 예: 'Wiki/태민게임즈/index.md' + 게임별 노트들.\n"
    "4. 기존 노트가 있는지 모르면 먼저 `obsidian_search`로 확인한 뒤, 있으면 mode='overwrite'/'append'.\n"
    "5. 도구 호출 후에만 '작성 완료'라고 보고하라. 호출 전에는 완료라고 말하지 마라.\n"
)

# 🔱 [Imperial v4.0] MCP Client dynamic import
try:
    from mcp_client import get_mcp_client
except ImportError:
    import sys
    import os
    sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
    try:
        from mcp_client import get_mcp_client
    except ImportError:
        get_mcp_client = None

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("JarvisBrain")


# ━━━━━━━━━━━━━━━━ 🙋 Ask User (턴 중간 선택지 질문) ━━━━━━━━━━━━━━━━
# 모델이 작업 도중 유저 결정을 기다리는 자비스(AskUserQuestion)식 메커니즘.
# 스트림에 ```imperial-ask``` 블록을 방출(헤임달이 버튼+타이머 렌더링)하고,
# threading.Event로 대기 → POST /api/shadow/ask/answer 가 해제.
# 타임아웃 시 자동 취소 + 사유 고지 + 레지스트리 정리 (좀비 스레드 방지).
_PENDING_ASKS = {}
_PENDING_ASKS_LOCK = threading.Lock()

def submit_ask_answer(ask_id: str, answer: str) -> bool:
    """헤임달 버튼 탭이 호출 — 대기 중인 ask를 깨운다. 없으면 False (만료/취소됨)."""
    with _PENDING_ASKS_LOCK:
        entry = _PENDING_ASKS.get(ask_id)
    if not entry:
        return False
    entry["answer"] = str(answer or "").strip()
    entry["event"].set()
    return True


def begin_ask(args: dict) -> dict:
    """ask 시작 — 대기 엔트리 등록 후 UI 페이로드 반환. (MCP/외부 에이전트 경로 공용)"""
    import uuid as _uuid
    ask_id = _uuid.uuid4().hex[:12]
    try:
        timeout = int(args.get("timeout_seconds", 120) or 120)
    except (TypeError, ValueError):
        timeout = 120
    timeout = max(30, min(timeout, 600))
    options = [str(o) for o in (args.get("options") or []) if str(o).strip()][:6]
    payload = {
        "ask_id": ask_id,
        "question": str(args.get("question", "")).strip(),
        "options": options,
        "timeout_seconds": timeout,
        "expires_at": time.time() + timeout,  # 헤임달 카운트다운 타이머용 (epoch)
    }
    with _PENDING_ASKS_LOCK:
        _PENDING_ASKS[ask_id] = {
            "event": threading.Event(), "answer": None,
            "created": time.time(), "payload": payload,  # pending 목록 노출용
        }
    return payload


def wait_ask(ask_id: str, timeout: int):
    """답변 대기. 답이 오면 문자열, 타임아웃이면 None. 어느 쪽이든 레지스트리는 반드시 정리."""
    with _PENDING_ASKS_LOCK:
        entry = _PENDING_ASKS.get(ask_id)
    if not entry:
        return None
    try:
        answered = entry["event"].wait(timeout)
        return entry["answer"] if answered else None
    finally:
        # 🧹 좀비 방지: 성공/타임아웃/예외 무관 항상 제거 → 늦게 온 답은 submit이 False 반환
        with _PENDING_ASKS_LOCK:
            _PENDING_ASKS.pop(ask_id, None)


def list_pending_asks() -> list:
    """현재 대기 중(미응답)인 ask 페이로드 목록 — 헤임달 out-of-band 카드 표시용."""
    now = time.time()
    with _PENDING_ASKS_LOCK:
        return [
            e["payload"] for e in _PENDING_ASKS.values()
            if e.get("payload") and not e["event"].is_set()
            and e["payload"].get("expires_at", 0) > now
        ]


# ━━━━━━━━━━━━━━━━ 🐞 Debug Mode (도구 로그 노출 제어) ━━━━━━━━━━━━━━━━
# 헤임달 '그림자 두뇌 설정'의 debugMode 토글. ON이면 ⚡ 도구 진행 로그를 채팅
# 스트림에 방출(=DB 저장/벡터화 포함), OFF면 서버 로그에만 남긴다.
# 도구 실행 자체는 설정과 무관하게 항상 동작한다.
_DEBUG_MODE_CACHE = {"value": False, "checked_at": 0.0}

def _is_debug_mode() -> bool:
    """JARVIS_SHADOW_SETTINGS.json의 debugMode를 핫리로드(5초 캐시)로 읽는다."""
    now = time.time()
    if now - _DEBUG_MODE_CACHE["checked_at"] < 5:
        return _DEBUG_MODE_CACHE["value"]
    value = False
    try:
        proj_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        settings_file = os.path.join(proj_root, "docker_data", "shared_workspace", "JarvisRun", "JARVIS_SHADOW_SETTINGS.json")
        if os.path.exists(settings_file):
            with open(settings_file, 'r', encoding='utf-8') as f:
                value = bool(json.load(f).get("debugMode", False))
    except Exception:
        pass
    _DEBUG_MODE_CACHE["value"] = value
    _DEBUG_MODE_CACHE["checked_at"] = now
    return value


# ━━━━━━━━━━━━━━━━ 🧠 Reasoning 노출 제어 ━━━━━━━━━━━━━━━━
# 일부 모델(예: minimax-m2.7)은 사고 과정을 reasoning_content 채널로 쏟아내 최종
# 답변에 메타/지시 메아리가 섞인다. show_reasoning(기본 False)이면 추론 채널을
# 스트림·최종 답변에서 숨긴다. (헤임달 '그림자 두뇌 설정'의 show_reasoning 토글)
_SHOW_REASONING_CACHE = {"value": False, "checked_at": 0.0}

def _show_reasoning() -> bool:
    """JARVIS_SHADOW_SETTINGS.json의 show_reasoning을 핫리로드(5초 캐시)로 읽는다."""
    now = time.time()
    if now - _SHOW_REASONING_CACHE["checked_at"] < 5:
        return _SHOW_REASONING_CACHE["value"]
    value = False
    try:
        proj_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        settings_file = os.path.join(proj_root, "docker_data", "shared_workspace", "JarvisRun", "JARVIS_SHADOW_SETTINGS.json")
        if os.path.exists(settings_file):
            with open(settings_file, 'r', encoding='utf-8') as f:
                value = bool(json.load(f).get("show_reasoning", False))
    except Exception:
        pass
    _SHOW_REASONING_CACHE["value"] = value
    _SHOW_REASONING_CACHE["checked_at"] = now
    return value


def _pick_nonreasoning_final(content_text, reasoning_text):
    """추론 숨김 모드에서 content가 비면 reasoning으로 폴백해 답변 유실을 막는다.

    대부분의 모델은 실제 답변을 content로 낸다 → content 우선. content가 완전히
    비어 있을 때만(추론 채널에만 답을 담는 모델 대비) reasoning을 폴백으로 쓴다.
    """
    if (content_text or "").strip():
        return content_text
    return reasoning_text or ""


def _summarize_tool_result(display_name: str, func_name: str, result) -> str:
    """🔱 [가독성] 도구 실행 결과를 채팅에 보여줄 한 줄 요약으로 변환한다.
    (원시 JSON 200자 덤프가 그대로 노출되던 문제 개선)"""
    text = str(result or '').strip()
    # 실패는 명확하게
    if text.startswith('❌') or text.lower().startswith('error'):
        one = ' '.join(text.split())[:160]
        return f"**{func_name}** 실패 — {one}"
    # JSON 결과면 핵심만 추출
    summary = None
    if text.startswith('{') or text.startswith('['):
        try:
            data = json.loads(text)
            if isinstance(data, dict):
                results = data.get('results')
                if isinstance(results, list):
                    titles = [str((r or {}).get('title', ''))[:40] for r in results[:3] if isinstance(r, dict)]
                    summary = f"검색 결과 {len(results)}건" + (f" — {' / '.join(t for t in titles if t)}" if titles else "")
                elif 'content' in data:
                    summary = f"본문 {len(str(data['content']))}자 수신"
        except Exception:
            pass
    if summary is None:
        one = ' '.join(text.split())
        summary = (one[:120] + '…') if len(one) > 120 else one
    return f"**{func_name}** 완료 — {summary}"

class JarvisBrain:
    # 🔱 [Imperial Fix] google-genai 2.10+ 의 generate_content는 config.model_copy(deep=True)로
    # 설정을 deep-copy 한다. self._tools에는 바운드 메서드(self.xxx)가 들어가는데, 그 deepcopy가
    # __self__(이 브레인 인스턴스 = genai.Client httpx 락 + threading.local 보유)까지 복사하려다
    # 'cannot pickle _thread.lock' 으로 실패한다(특히 Gemini AFC 비스트리밍 경로). 브레인을
    # deepcopy 불변(자기 자신 반환)으로 만들어 락 복사를 회피한다 — SDK는 브레인을 변형하지 않으므로 안전.
    def __deepcopy__(self, memo):
        memo[id(self)] = self
        return self

    def __init__(self):
        # 🔱 [Imperial] Force drop legacy environment variables to silence SDK warnings
        os.environ.pop("GOOGLE_API_KEY", None)
        
        self.api_key = os.environ.get("GEMINI_API_KEY")
        
        # Guard: Check if the key is a Bitwarden placeholder (Aegis format)
        if self.api_key and (self.api_key.startswith("BW_ID:") or self.api_key.startswith("Imperial:")):
            logger.info(f"Targeting Key ID: {self.api_key}")
            # [🔱 Imperial Fix] If it's a placeholder, force missing state to trigger recovery
            self.api_key = None
            
        # [IMPERIAL FIX] If API Key is still missing or was a placeholder, try to load from Bitwarden
        # [🔱 Imperial Centralized Secret Recovery]
        if not self.api_key:
            self.api_key = sovereign_memory.get_secret("GEMINI_API_KEY")
            if self.api_key:
                os.environ["GEMINI_API_KEY"] = self.api_key
                logger.info("[SUCCESS] GEMINI_API_KEY recovered from Aegis Oracle.")
            else:
                logger.warning("[SECURITY] Failed to recover GEMINI_API_KEY from Aegis Oracle.")


        self.brave_api_key = os.environ.get("BRAVE_API_KEY")
        if self.brave_api_key and (self.brave_api_key.startswith("BW_ID:") or self.brave_api_key.startswith("Imperial:")):
            self.brave_api_key = None
            
        if not self.brave_api_key:
            self.brave_api_key = sovereign_memory.get_secret("BRAVE_API_KEY")
            if self.brave_api_key:
                os.environ["BRAVE_API_KEY"] = self.brave_api_key
                logger.info("[SUCCESS] BRAVE_API_KEY recovered from Aegis Oracle.")

        if self.api_key and genai:
            # Mask key for safety in logs
            masked = self.api_key[:8] + "..." + self.api_key[-4:]
            logger.info(f"[STATUS] Imperial Brain Synchronized with key: {masked}")
            self.client = genai.Client(api_key=self.api_key)
        else:
            logger.error("Imperial Brain Initialization Failed: GEMINI_API_KEY is missing or invalid.")
            self.client = None
            
        if self.brave_api_key:
            masked_brave = self.brave_api_key[:8] + "..." + self.brave_api_key[-4:] if len(self.brave_api_key) > 8 else "..."
            logger.info(f"[STATUS] Imperial Brave Search Synchronized with key: {masked_brave}")
        
        self.current_sender = "마왕님" # [🔱 Identity SSOT]
        
        # 자동 검색은 Aegis common_retrieval_policy가 명시적으로 허용할 때만 동작한다.
        self.web_search_enabled = False
        
        # Initialize Hands (Tools)
        self.browser = BrowserController()
        # 🛠️ [Imperial Coding Arsenal] Claude Code식 에이전틱 코딩 도구
        self.coding = CodingToolkit()
        # 🛡️ [Tool Runtime] 3개 OpenAI식 대화 경로가 공유하는 단일 도구 실행 런타임
        self.runtime = ToolRuntime(self, _is_debug_mode, _summarize_tool_result,
                                   get_mcp_client=get_mcp_client)
        # [🧠 정직 게이트] 요청별 컨텍스트(aegis_url/token/session) — 도구가 호출 시점에 참조
        self._request_ctx = threading.local()
        self._tools = [
            # 코딩 도구를 최우선 배치 — 코드 질문에서 모델이 먼저 고려하게 함
            *self.coding.callables(),
            # [🧠 정직 게이트] 기억/기록 사실 조회 도구를 최상단 배치 — 기억 질문에 추측 대신 조회
            self.recall_memory,
            self.count_history,
            # [🧠 지식 베이스] 운영 사실 조회/기록 — 시스템 구성 질문에 추측 대신 조회
            self.search_knowledge,
            self.save_knowledge,
            self.web_search,  # 🌐 실시간 웹 검색 도구 (네이버/다음/구글 headless Playwright)
            # [🖼️ 미디어] 로열티-프리 사진/영상/GIF 검색 — 이미지 요청에 사이트 추천 대신 실제 검색
            self.search_media,
            # [🧊 3D 에셋] 로열티-프리 3D 에셋(Pixabay 3D) 검색 — 3D/GLB 요청에 사이트 추천 대신 실제 검색
            self.search_3d_model,
            # [🔭 관측 도구] 읽기 전용 시스템 관측 — 환경 지식의 현장 재검증
            self.check_port,
            self.find_process,
            # [📚 사서] 타 수호자 지식 제안 심사 (쉐도우 브레인 = 지식 사서)
            self.review_knowledge_proposals,
            self.judge_knowledge_proposal,
            # [🔱 Imperial Fix] CLI/파일 우선 도구를 앞쪽에 배치하여 경량 모델이 우선 고려하게 함.
            # CLI-First Tools
            self.jarvis_memory,
            self.jarvis_system,
            self.jarvis_skill,
            self.jarvis_soul,
            self.jarvis_gbb_post,
            self.generate_image,  # 🎨 Imperial Image Generation Tool
            self.sovereign_type_text,
            self.sovereign_press_key,
            self.sovereign_run_app, # 🚀 OS Application Launch Tool
            self.delegate_team_task,  # 🏛️ Team Autonomous Delegation Tool
            self.delegate_to_guardian,  # 🤝 동료 수호자 위임 (코덱스/미니맥스/자비스 등)
            self.dispatch_to_hydra,   # 🐲 Hydra Parallel Planning Dispatch Tool
            # [🔱 Imperial Fix v2.7] Hugin(브라우저 조작) 도구는 맨 뒤로 이동.
            # 모델이 파일 수정/이미지 분석 요청에서 습관적으로 hugin_screenshot을 먼저
            # 호출하던 문제를 막기 위함. (명시적 브라우저 조작 요청에서만 사용되도록 유도)
            self.browser.hugin_navigate,
            self.browser.hugin_screenshot,
        ]

        # 🗂️ [Obsidian] 옵시디언 및 Graphify 도구를 Gemini AFC 도구 주머니에 추가.
        try:
            import obsidian_router as _ov
            if _ov.is_any_enabled():
                self._tools += [self.obsidian_search, self.obsidian_read, self.obsidian_write, self.project_graph_query]
                logger.info("[🗂️ Obsidian] 옵시디언 도구 4종(지식 그래프 포함)을 Gemini 도구 주머니에 추가.")
        except Exception as _obs_err:
            logger.warning(f"[🗂️ Obsidian] 도구 추가 건너뜀: {_obs_err}")

        # 🔱 [Imperial v4.0] Dynamic MCP Tools injection for tool unification
        if get_mcp_client:
            try:
                mcp_client = get_mcp_client()
                if mcp_client:
                    mcp_client.wait_ready(timeout=15.0)  # [🔱 Imperial Fix] npx 다운로드 및 실행 시간을 고려하여 15초 대기
                    mcp_callables = mcp_client.get_gemini_callables()
                    if mcp_callables:
                        self._tools.extend(mcp_callables)
                        logger.info(f"[🔱 Imperial Core] MCP 도구 {len(mcp_callables)}개를 채팅 엔진 도구 주머니에 통합 완료.")
            except Exception as mcp_init_err:
                logger.warning(f"[🔱 Imperial Core] MCP 도구 주입 중 예외 발생: {mcp_init_err}")

        # AFC Tool Result Tracker (for empty response fallback)
        self._afc_tool_results = []

        # [🔱 Imperial Fix v2.7] MCP 도구 실행 결과를 _afc_tool_results로 흘려보내는 싱크 연결.
        # AFC가 mcp_filesystem_write_file 등을 호출해 성공해도 모델이 최종 텍스트를 안 만들면
        # 화면에 아무것도 안 뜨던 문제 해결 — MCP 결과도 화면에 표시되게 한다.
        if get_mcp_client:
            try:
                _mc = get_mcp_client()
                if _mc:
                    _mc._result_sink = self._afc_tool_results
            except Exception:
                pass

        # Copilot Cache
        self._copilot_token = None
        self._copilot_expires = 0
        self._copilot_url = "https://api.individual.githubcopilot.com"

        # [IMPERIAL EXTENSION] Configurable Primary/Fallback
        self.primary_provider = "google"
        self.primary_model = "gemini-3.1-flash-lite"
        self.fallback_provider = "github"
        self.fallback_model = "gpt-4o"
        self.search_grounding_model = "gemini-2.5-flash"
        self.ollama_host = os.environ.get("OLLAMA_HOST", "http://127.0.0.1:11434")
        
        # [🔱 OpenRouter] API Key (Bitwarden: Imperial:OPENROUTER_CLAUDE_TOKEN)
        self.openrouter_api_key = os.environ.get("OPENROUTER_API_KEY", "")
        if not self.openrouter_api_key:
            self.openrouter_api_key = sovereign_memory.get_secret("OPENROUTER_CLAUDE_TOKEN")
            if self.openrouter_api_key:
                logger.info("[OpenRouter] API Key recovered from Aegis Oracle.")

        # [🔱 NVIDIA] NVIDIA NIM API Keys.
        #   MiniMax-M3 uses its own key. Do not fall back to the M2.7/general key;
        #   NVIDIA accepts model-scoped keys and a wrong key returns auth/model errors.
        #   OpenAI 호환 엔드포인트: https://integrate.api.nvidia.com/v1
        #   ⚠️ env에 'Imperial:...' 플레이스홀더(미해결)가 들어오면 그대로 Bearer로 나가 401이 된다 → 반드시 걸러낼 것.
        def _normalize_nvidia_key(val):
            val = (val or "").strip().strip('"').strip("'")
            if val.lower().startswith("bearer "):
                val = val[7:].strip()
            return val

        def _resolve_nvidia_key(env_names):
            for env_name in env_names:
                val = _normalize_nvidia_key(os.environ.get(env_name))
                source = "environment"
                if val and val.startswith(("BW_ID:", "Imperial:")):
                    val = None
                if not val:
                    val = _normalize_nvidia_key(sovereign_memory.get_secret(env_name))
                    source = "Aegis Oracle"
                    if val and val.startswith(("BW_ID:", "Imperial:")):
                        val = None
                if not val:
                    for item_name in (f"Imperial:{env_name}", f"Imperial: {env_name}"):
                        try:
                            bw = subprocess.run(
                                ["bw", "get", "password", item_name, "--raw"],
                                capture_output=True,
                                text=True,
                                encoding="utf-8",
                                errors="replace",
                                timeout=10,
                            )
                            candidate = _normalize_nvidia_key(bw.stdout)
                            if bw.returncode == 0 and candidate and not candidate.startswith(("BW_ID:", "Imperial:")):
                                val = candidate
                                source = f"Bitwarden item {item_name}"
                                break
                        except Exception:
                            pass
                if val:
                    os.environ[env_name] = val
                    logger.info(f"[NVIDIA] API Key recovered from {source} (via {env_name}).")
                    return val
            return None

        self.nvidia_api_key = _resolve_nvidia_key(("NVIDIA_API_KEY",))
        self.nvidia_minimax_m3_api_key = None  # 통합: NVIDIA_API_KEY 단일 키 사용

        # [🔱 GLM / Z.ai] zai-org GLM 모델 (modal.direct OpenAI-compat 엔드포인트)
        glm_raw = (os.environ.get("GLM_API_KEY", "") or "").strip()
        if not glm_raw or glm_raw.startswith(("BW_ID:", "Imperial:")):
            glm_raw = _normalize_nvidia_key(sovereign_memory.get_secret("GLM_API_KEY")) or ""
        if not glm_raw or glm_raw.startswith(("BW_ID:", "Imperial:")):
            for _item in ("Imperial:GLM_API_KEY", "Imperial: GLM_API_KEY"):
                try:
                    _bw = subprocess.run(
                        ["bw", "get", "password", _item, "--raw"],
                        capture_output=True, text=True, encoding="utf-8",
                        errors="replace", timeout=10,
                    )
                    _c = _normalize_nvidia_key(_bw.stdout)
                    if _bw.returncode == 0 and _c and not _c.startswith(("BW_ID:", "Imperial:")):
                        glm_raw = _c
                        break
                except Exception:
                    pass
        if glm_raw:
            os.environ["GLM_API_KEY"] = glm_raw
            logger.info("[GLM] API Key resolved (Bitwarden/Aegis).")
        self.glm_api_key = glm_raw or None
        
        # [🔱 Imperial Atlas] Dynamic Port Discovery
        self.atlas_broker_url = os.environ.get("ATLAS_BROKER_URL", "http://127.0.0.1:18799")
        self._atlas_assigned_port = None
        
        self._load_shadow_settings()
        self.dynamic_config = self.get_dynamic_config()
        
        # [🔱 V7.0 Stateless Client Cache]
        self._clients = {} # {api_key: genai.Client}
        if self.api_key and genai and hasattr(self, 'client'):
            self._clients[self.api_key] = self.client

    def _classify_job_by_prompt(self, prompt: str) -> str:
        """사용자의 질문(prompt)을 분석하여 최적의 직업 ID를 자동 판별합니다.
        
        기본 내장된 직업 매핑:
        - obsidian / 노트 / 위키 / 일지 / 마크다운 작성 -> file_artisan (파일 장인 - 옵시디언 도구 포함)
        - 검색 / 구글 / 웹서핑 / 최신 정보 / 찾아줘 -> scout (정찰병)
        - 브라우저 / hugin / 스크린샷 / 캡쳐 / 사이트 방문 -> browser_pilot (브라우저 조종사)
        - 팀 / 히드라 / 위임 / 소환 -> commander (지휘관)
        - 그려줘 / 이미지 / 생성 / 그림 -> artist (창작가)
        """
        p = prompt.lower()
        
        # 1. 브라우저 조종사 의도 우선 (hugin, 브라우저 제어)
        if any(k in p for k in ["hugin", "브라우저", "스크린샷", "캡쳐", "캡처", "웹페이지 열", "사이트 방문", "플레이라이트", "playwright"]):
            return "browser_pilot"
            
        # 2. 창작가 의도 (그려줘, 이미지 생성)
        if any(k in p for k in ["그려줘", "이미지", "그림", "생성", "화폭", "단조", "generate_image"]):
            # 단, 단순 텍스트 '코드 생성' 등은 제외
            if "코드" not in p and "텍스트" not in p:
                return "artist"
                
        # 3. 지휘관 의도 (위임, 히드라, 팀)
        if any(k in p for k in ["히드라", "위임", "소환", "오케스트", "팀 태스크", "다른 수호자"]):
            return "commander"
            
        # 4. 파일 장인 (옵시디언, 로컬 파일 쓰기/읽기)
        if any(k in p for k in ["옵시디언", "obsidian", "노트", "위키", "일지", "일기", "글쓰기", "파일", "폴더", "수정", "저장", "기록"]):
            return "file_artisan"
            
        # 5. 정찰병 의도 (검색, 최신 정보)
        if any(k in p for k in ["검색", "구글", "구글링", "최신", "찾아", "정보 수집"]):
            return "scout"
            
        # 6. 기본값은 만능 수호자
        return ""

    def get_dynamic_config(self, token=None, override_config=None, aegis_url=None):
        """[🔱 Imperial] 주입된 설정 또는 DB/Aegis API로부터 설정을 동기화합니다."""
        # 1. 명시적으로 주입된 설정이 있다면 최우선 (Pure Intelligence Mode)
        if override_config:
            logger.info("[🧠 Config] Using injected override configuration.")
            return override_config

        # [🛡️] Default Config (Fallback)
        default_config = {
            "provider": "google",
            "model": "gemini-3.1-flash-lite",
            "temperature": 0.7,
            "max_iterations": 15
        }
        
        try:
            from db_util import get_db_connection
            conn = get_db_connection()
            
            # 2. DB 연결이 없는 경우 Aegis API 시도
            if not conn:
                target_aegis = aegis_url or os.environ.get("AEGIS_URL", "").rstrip("/")
                if not target_aegis:
                    return default_config # Zero-Config Fallback

                auth_token = token or os.environ.get("IMPERIAL_API_TOKEN")
                persona_id = sovereign_memory.get_secret("SHADOW_BRAIN_PERSONA_ID", default="a8e1a1af-f28b-47ab-b371-193082508638")
                api_url = f"{target_aegis}/api/chat/settings/{persona_id}"
                
                resp = requests.get(api_url, headers={"X-Imperial-Token": auth_token}, timeout=3)
                if resp.status_code == 200:
                    data = resp.json()
                    if data.get('status') == 'SUCCESS':
                        return data.get('config', default_config)
                return default_config
                
            # 3. DB 직접 조회 (Legacy Fallback)
            cur = conn.cursor()
            shadow_brain_id = sovereign_memory.get_secret("SHADOW_BRAIN_PERSONA_ID", default="a8e1a1af-f28b-47ab-b371-193082508638")
            cur.execute("""
                SELECT ai_provider, ai_model, temperature, capabilities 
                FROM taemingames.guardian_personas 
                WHERE id = %s
            """, (shadow_brain_id,))
            row = cur.fetchone()
            cur.close()
            conn.close()
            
            if row:
                capabilities = row[3] if isinstance(row[3], dict) else json.loads(row[3] or '{}')
                # [🎖️ Guardian Job M3] capabilities에 'job' 프리셋이 있으면 직업 설정을 병합.
                #   - 성능 토글(memory_recall 등)과 num_ctx는 '직업이 단일 진실'이므로
                #     allowed_tools 유무와 무관하게 항상 병합한다.
                #   - 도구/스킬/룰은 allowed_tools를 직접 지정하지 않은 경우에만 직업으로 덮어쓴다.
                if capabilities.get("job"):
                    try:
                        from guardian_jobs import get_job as _get_job
                        _job = _get_job(capabilities.get("job")) or {}
                        for _k in ("memory_recall", "proactive_context", "stage_timing", "inject_guidelines", "use_history", "use_persona", "inject_emotion", "inject_readability"):
                            capabilities[_k] = bool(_job.get(_k, True))
                        capabilities["obsidian_job_path"] = str(_job.get("obsidian_job_path") or "")
                        if _job:
                            capabilities["job_source"] = str(_job.get("source") or "vault").lower()
                        if _job.get("num_ctx"):
                            capabilities["num_ctx"] = int(_job.get("num_ctx") or 0)
                        if _job.get("num_predict"):
                            capabilities["num_predict"] = int(_job.get("num_predict") or 0)
                        if not capabilities.get("allowed_tools"):
                            # 🎮 [앱 세트] 장착 앱(equipped_apps)을 도구/스킬로 펼쳐 합집합.
                            try:
                                from guardian_jobs import resolve_effective_tools_skills as _resolve_ets
                                _eff_tools, _eff_skills = _resolve_ets(_job)
                            except Exception:
                                _eff_tools = list(_job.get("allowed_tools") or [])
                                _eff_skills = list(_job.get("equipped_skills") or [])
                            capabilities["allowed_tools"] = _eff_tools
                            capabilities["tool_mode"] = _job.get("tool_mode", "all")
                            capabilities["equipped_skills"] = _eff_skills
                            capabilities["job_rule"] = _job.get("rule", "")
                    except Exception as _job_err:
                        logger.warning(f"[Guardian Job M3] 직업 프리셋 적용 실패: {_job_err}")
                return {
                    "provider": row[0] or "google",
                    "model": row[1] or "gemini-3.1-flash-lite",
                    "temperature": row[2] or 0.7,
                    "max_iterations": capabilities.get("max_iterations", 10),
                    # [🔱 Guardian Job M1] 직업(도구/스킬 특화) 필드 통과
                    "allowed_tools": capabilities.get("allowed_tools"),
                    "tool_mode": capabilities.get("tool_mode", "all"),
                    "equipped_skills": capabilities.get("equipped_skills", []),
                    "job_rule": capabilities.get("job_rule", ""),  # 직업 기본 룰
                    # [🎖️ Guardian Job] Ollama 컨텍스트 윈도우(직업별 속도/정확도 튜닝). 0/미지정이면 자동.
                    "num_ctx": capabilities.get("num_ctx", 0),
                    "num_predict": capabilities.get("num_predict", 0),
                    # 🚀 성능 토글 (직업별)
                    "memory_recall": capabilities.get("memory_recall", True),
                    "proactive_context": capabilities.get("proactive_context", True),
                    "stage_timing": capabilities.get("stage_timing", True),
                    "inject_guidelines": capabilities.get("inject_guidelines", True),
                    "inject_emotion": capabilities.get("inject_emotion", True),
                    "inject_readability": capabilities.get("inject_readability", True),
                    "use_history": capabilities.get("use_history", True),
                    "use_persona": capabilities.get("use_persona", True),
                    "obsidian_job_path": capabilities.get("obsidian_job_path", ""),
                    "job_source": capabilities.get("job_source", "cloud"),
                }
        except Exception:
            pass
        
        return default_config

    def _load_shadow_settings(self):
        """Loads Shadow Brain configuration from the shared workspace."""
        try:
            proj_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
            settings_file = os.path.join(proj_root, "docker_data", "shared_workspace", "JarvisRun", "JARVIS_SHADOW_SETTINGS.json")
            if os.path.exists(settings_file):
                with open(settings_file, 'r', encoding='utf-8') as f:
                    settings = json.load(f)
                    self.primary_provider = settings.get("provider", "google")
                    self.primary_model = settings.get("model", "gemini-3.1-flash-lite")
                    self.fallback_provider = settings.get("fallbackProvider", "github")
                    self.fallback_model = settings.get("fallbackModel", "gpt-4o")
                    self.web_search_enabled = settings.get("webSearchEnabled", self.web_search_enabled)
                    self.search_grounding_model = settings.get("searchGroundingModel") or settings.get("search_grounding_model") or self.search_grounding_model
                    self.ollama_host = settings.get("ollamaHost") or settings.get("ollama_host") or self.ollama_host
                    
                    # 🔱 아틀라스 관제 관문(Broker Port) 연동
                    atlas_port = settings.get("atlasBrokerPort", 18799)
                    self.atlas_broker_url = f"http://localhost:{atlas_port}"
                    
                    logger.info(f"[STATUS] Imperial Brain Configured: Primary={self.primary_model}, Port={atlas_port}, WebSearch={self.web_search_enabled}")
        except Exception as e:
            logger.warning(f"Failed to load Shadow settings: {e}")

    def _execute_cli(self, script_name: str, args: List[str]) -> str:
        """Helper to execute external CLI scripts safely."""
        try:
            proj_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
            script_path = os.path.join(proj_root, "scripts", "cli", script_name)
            
            # [🛡️] Dynamic Token Delegation: Pass auth_token to CLI tools
            env = os.environ.copy()
            if hasattr(self, 'auth_token') and self.auth_token:
                env["IMPERIAL_API_TOKEN"] = self.auth_token
                env["X_IMPERIAL_TOKEN"] = self.auth_token
            
            # Capture raw bytes to handle Windows encoding (CP949) vs UTF-8 conflicts
            result = subprocess.run(cmd, capture_output=True, timeout=60, env=env)
            
            def _multi_decode(data: bytes) -> str:
                if not data: return ""
                for enc in ['utf-8', 'cp949', 'euc-kr']:
                    try:
                        return data.decode(enc)
                    except UnicodeDecodeError:
                        continue
                return data.decode('utf-8', errors='replace')

            stdout_str = _multi_decode(result.stdout)
            stderr_str = _multi_decode(result.stderr)

            if result.returncode == 0:
                self._afc_tool_results.append(stdout_str.strip() if stdout_str.strip() else f"✅ {script_name} 실행 완료")
                return stdout_str
            else:
                return f"Error ({result.returncode}): {stderr_str}"
        except Exception as e:
            logger.error(f"CLI Execution Failed ({script_name}): {e}")
            return f"Critical Error: {str(e)}"

    def _get_mcp_allowed_dirs(self):
        """mcp_servers.json의 filesystem 화이트리스트 폴더 목록을 읽는다 (실패 시 빈 목록)."""
        try:
            try:
                from mcp_client import resolve_mcp_config_path
                cfg_path = resolve_mcp_config_path()
            except Exception:
                proj_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
                cfg_path = os.path.join(proj_root, 'mcp_servers.json')
            with open(cfg_path, 'r', encoding='utf-8') as f:
                config = json.load(f)
            for server in config.get('servers', []):
                if server.get('name') == 'filesystem':
                    args = server.get('args', [])
                    if "@modelcontextprotocol/server-filesystem" in args:
                        idx = args.index("@modelcontextprotocol/server-filesystem")
                        return [str(d) for d in args[idx + 1:]]
        except Exception:
            pass
        return []

    def _resolve_aegis(self):
        """요청 컨텍스트(우선) 또는 환경변수에서 Aegis URL/토큰을 해석한다."""
        ctx = getattr(self, "_request_ctx", None)
        aegis_url = (getattr(ctx, "aegis_url", None) if ctx else None) or os.environ.get("AEGIS_URL", "")
        token = (getattr(ctx, "token", None) if ctx else None) or os.environ.get("IMPERIAL_API_TOKEN", "")
        return (aegis_url or "").rstrip("/"), token

    def search_knowledge(self, query: str, limit: int = 5) -> str:
        """
        Searches the Imperial Knowledge Base (curated operational facts: config paths, ports,
        system locations, architecture decisions). MUST be used before answering questions about
        system configuration, file locations, or how Imperial infrastructure works.
        Args:
            query: Keywords to search (e.g., 'OpenClaw 설정 경로', 'Hermes 프록시 포트')
            limit: Max facts to return (default 5)
        """
        try:
            aegis_url, token = self._resolve_aegis()
            if not aegis_url:
                return "지식 베이스 접근 불가: Aegis URL 미설정."
            import platform
            resp = requests.get(
                f"{aegis_url}/api/knowledge",
                params={"q": query, "limit": int(limit or 5), "env": platform.node()},
                headers={"X-Imperial-Token": token}, timeout=5,
            )
            data = resp.json() if resp.status_code == 200 else {}
            results = data.get("results", [])
            if not results:
                return ("운영 지식 베이스 검색 결과: 관련 지식 없음. "
                        "추측하지 말고 '등록된 지식이 없어 확인이 필요하다'고 답하라. "
                        "이후 사실을 확인하면 save_knowledge로 기록하라.")
            lines = []
            for i, r in enumerate(results, 1):
                verified = (r.get("verified_at") or "")[:10]
                scope_tag = "🌍보편" if r.get("scope") == "global" else f"💻이 환경({r.get('environment_id')})"
                line = f"{i}. [{r.get('topic')}|{scope_tag}] {r.get('fact')} (확인: {verified}, 등록: {r.get('registered_by')})"
                if r.get("verify_hint"):
                    line += f"\n   ↳ 재검증 방법: {r['verify_hint']}"
                lines.append(line)
            return ("운영 지식 베이스 검색 결과:\n" + "\n".join(lines) +
                    "\n규칙(신뢰 등급): 🌍보편 지식은 그대로 신뢰 가능. 💻환경 지식은 '기록상' 사실이므로, "
                    "재검증 방법이 있으면 check_port/find_process 등으로 지금 이 환경에서 확인한 뒤 단정하라. "
                    "검증 없이 답할 때는 '기록 기준(확인일 명시)'임을 밝혀라. 확인일이 오래됐으면 그 점도 알려라.")
        except Exception as e:
            return f"지식 베이스 검색 실패: {e}. 검색이 안 됐으므로 시스템 구성을 단정하지 마라."

    def search_media(self, query: str, media_type: str = "photo", provider: str = "pexels", limit: int = 8) -> str:
        """
        로열티-프리(무료) 미디어를 검색한다. Pexels/Pixabay 사진·영상·GIF를 키워드로 찾아
        원본/다운로드/출처 URL을 반환한다. 사용자가 이미지/사진/영상/GIF/배경화면/스톡 미디어를
        찾아달라고 하면 추측(사이트 추천)하지 말고 이 도구로 실제 검색하라.
        Args:
            query: 검색 키워드 (예: '고양이', 'fire')
            media_type: 'photo' | 'video' | 'gif' (기본 photo)
            provider: 'pexels' | 'pixabay' (photo/video용, 기본 pexels. gif는 무시)
            limit: 반환 개수 1-20 (기본 8)
        """
        try:
            aegis_url, token = self._resolve_aegis()
            if not aegis_url:
                aegis_url = os.environ.get("AEGIS_URL", "http://localhost:18701").rstrip("/")
            try:
                lim = max(1, min(20, int(limit or 8)))
            except (TypeError, ValueError):
                lim = 8
            resp = requests.get(
                f"{aegis_url}/api/wallpaper/search",
                params={"q": query, "type": media_type, "provider": provider},
                headers={"X-Imperial-Token": token} if token else {},
                timeout=25,
            )
            if resp.status_code != 200:
                return f"미디어 검색 실패: HTTP {resp.status_code}. 결과를 단정하지 마라."
            data = resp.json()
            if data.get("status") != "ok":
                return f"미디어 검색 실패: {data.get('message', 'unknown')}"
            hits = (data.get("hits") or [])[:lim]
            if not hits:
                return f"'{query}' ({media_type}) 검색 결과 없음. 다른 검색어를 제안하라."
            lines = [f"'{query}' {media_type} 로열티-프리 검색 결과 ({len(hits)}건, 모두 무료 사용 가능):"]
            for i, h in enumerate(hits, 1):
                dl = h.get("download", "")
                if isinstance(dl, str) and dl.startswith("/"):
                    dl = aegis_url + dl
                lines.append(
                    f"{i}. [{h.get('source')}|{h.get('type')}] {h.get('url')}"
                    f"\n   ⬇ 다운로드: {dl}\n   🔗 출처: {h.get('page')}"
                )
            lines.append("규칙: 실제 검색된 결과다. URL을 그대로 사용자에게 제시하라(사이트 추천으로 대체하지 말 것).")
            return "\n".join(lines)
        except Exception as e:
            return f"미디어 검색 실패: {e}. 결과를 단정하지 마라."

    def search_media_hits(self, query: str, media_type: str = "photo", provider: str = "pexels", limit: int = 8):
        try:
            aegis_url, token = self._resolve_aegis()
            if not aegis_url:
                aegis_url = os.environ.get("AEGIS_URL", "http://localhost:18701").rstrip("/")
            try:
                lim = max(1, min(20, int(limit or 8)))
            except (TypeError, ValueError):
                lim = 8
            resp = requests.get(
                f"{aegis_url}/api/wallpaper/search",
                params={"q": query, "type": media_type, "provider": provider},
                headers={"X-Imperial-Token": token} if token else {},
                timeout=25,
            )
            if resp.status_code != 200:
                return aegis_url, [], f"미디어 검색 실패: HTTP {resp.status_code}. 결과를 추정하지 마라."
            data = resp.json()
            if data.get("status") != "ok":
                return aegis_url, [], f"미디어 검색 실패: {data.get('message', 'unknown')}"
            return aegis_url, (data.get("hits") or [])[:lim], None
        except Exception as e:
            return "", [], f"미디어 검색 실패: {e}. 결과를 추정하지 마라."

    def search_3d_model(self, query: str, limit: int = 8, source: str = "both") -> list:
        """🧊 로열티-프리 3D 모델을 검색해 hit 리스트를 반환한다.

        source: 'pixabay' | 'sketchfab' | 'both'(기본).
        반환: [{id, name, thumbnail_url, href, source, ...}, ...] (실패/결과없음이면 []).
        Sketchfab hit은 embed_url/viewer_url/author를 추가로 담는다 (헤임달이
        source로 뷰어를 분기). 실제 채팅 카드(imperial-3d 블록) 방출은 ToolRuntime 담당.
        """
        try:
            aegis_url, token = self._resolve_aegis()
            if not aegis_url:
                aegis_url = os.environ.get("AEGIS_URL", "http://localhost:18701").rstrip("/")
            headers = {"X-Imperial-Token": token} if token else {}
            try:
                lim = max(1, min(20, int(limit or 8)))
            except (TypeError, ValueError):
                lim = 8
            src = (source or "both").strip().lower()
            if src not in ("pixabay", "sketchfab", "both"):
                src = "both"

            def _fetch(path: str) -> list:
                try:
                    resp = requests.get(
                        f"{aegis_url}{path}",
                        params={"q": query},
                        headers=headers,
                        timeout=25,
                    )
                    if resp.status_code != 200:
                        logger.warning(f"[3D Search] {path} HTTP {resp.status_code}")
                        return []
                    data = resp.json()
                    if data.get("status") != "ok":
                        logger.warning(f"[3D Search] {path} 실패: {data.get('message')}")
                        return []
                    return data.get("hits") or []
                except Exception as e:  # noqa: BLE001
                    logger.warning(f"[3D Search] {path} 오류: {e}")
                    return []

            out = []
            if src in ("sketchfab", "both"):
                for h in _fetch("/api/sketchfab/search"):
                    if not h.get("embed_url"):
                        continue
                    out.append({
                        "id": str(h.get("id") or ""),
                        "name": h.get("name") or "Sketchfab Model",
                        "thumbnail_url": h.get("thumbnail_url") or "",
                        "href": h.get("viewer_url") or "",
                        "embed_url": h.get("embed_url") or "",
                        "viewer_url": h.get("viewer_url") or "",
                        "author": h.get("author") or "",
                        "license": h.get("license") or "",
                        "source": "sketchfab",
                    })
            if src in ("pixabay", "both"):
                for h in _fetch("/api/3d-models/search"):
                    if not h.get("href"):
                        continue
                    out.append({
                        "id": str(h.get("id") or ""),
                        "name": h.get("name") or "Pixabay 3D Model",
                        "thumbnail_url": h.get("thumbnail_url") or "",
                        "href": h.get("href") or "",
                        "source": "pixabay",
                    })
            if src == "both" and out:
                # 두 소스를 번갈아 섞어 한쪽이 상위를 독점하지 않게 한다
                sf = [h for h in out if h["source"] == "sketchfab"]
                px = [h for h in out if h["source"] == "pixabay"]
                mixed = []
                for i in range(max(len(sf), len(px))):
                    if i < len(sf):
                        mixed.append(sf[i])
                    if i < len(px):
                        mixed.append(px[i])
                out = mixed
            return out[:lim]
        except Exception as e:  # noqa: BLE001
            logger.error(f"[3D Search] 오류: {e}")
            return []

    def web_search(self, query: str, engine: str = "naver") -> str:
        """🌐 실시간 웹검색(시사·인물·뉴스·날씨·시세 등).
        1순위로 Google 검색 그라운딩(Gemini, 매우 빠름)을 쓰고, 실패하면
        headless Python Playwright(네이버→다음→구글, 느림)로 폴백한다.
        사용자가 특정 엔진(naver/daum/google)을 명시하면 곧장 Playwright를 쓴다.
        (구글 직접 스크래핑은 데이터센터 IP 캡차로 막히므로 폴백은 네이버 우선.)"""
        # 🔱 [Imperial Turbo Search] 엔진 미지정(기본 naver)이면 Google 그라운딩 우선.
        # 명시적 엔진 선택(사용자가 ask_user에서 네이버/다음/구글 고른 경우)은 의도 존중해 Playwright 직행.
        if (engine or "naver").strip().lower() == "naver":
            try:
                aegis_url, token = self._resolve_aegis()
                if aegis_url:
                    grounded = self._search_grounding_context(query, aegis_url=aegis_url, token=token)
                    if grounded and grounded.strip():
                        return grounded
            except Exception as ge:  # noqa: BLE001
                logger.warning(f"[Web Search] Google 그라운딩 실패 → Playwright 폴백: {ge}")
        try:
            return self.browser.local_playwright_search(query, engine=engine)
        except Exception as e:  # noqa: BLE001
            logger.error(f"[Web Search] 오류: {e}")
            return f"웹검색 실패: {e}"

    def youtube_search(self, query: str, limit: int = 5) -> str:
        """🎥 실시간 유튜브(YouTube) 동영상 검색.
        특정 키워드/주제에 대한 실시간 유튜브 영상 정보를 검색하여 썸네일과 동영상 링크 목록을 획득한다.
        Args:
            query: 유튜브 검색 질의어 (예: '용인 여행 코스 추천')
            limit: 가져올 최대 동영상 개수 (기본 5)
        """
        try:
            aegis_url, token = self._resolve_aegis()
            if not aegis_url:
                return "유튜브 검색 실패: Aegis URL 미설정."
            
            url = f"{aegis_url.rstrip('/')}/api/youtube/search"
            headers = {}
            if token:
                headers['X-Imperial-Token'] = token
                headers['Authorization'] = f"Bearer {token}"
                
            resp = requests.get(url, params={'query': query}, headers=headers, timeout=6)
            if resp.status_code != 200:
                return f"유튜브 검색 실패: API 응답 코드 {resp.status_code}"
                
            data = resp.json()
            videos = data.get('videos', [])[:limit]
            if not videos:
                return "유튜브 검색 결과가 없습니다."
                
            video_blocks = []
            for v in videos:
                video_blocks.append(f"- **[{v['title']}]({v['video_url']})** (채널: {v['channel_title']})")
                
            json_block = {
                "query": query,
                "videos": videos
            }
            json_str = json.dumps(json_block, ensure_ascii=False, indent=2)
            
            result = (
                f"유튜브 검색 완료 (키워드: '{query}'):\n"
                + "\n".join(video_blocks)
                + f"\n\n```imperial-youtube\n{json_str}\n```"
            )
            return result
        except Exception as e:
            logger.error(f"[YouTube Search] 오류: {e}")
            return f"유튜브 검색 중 예외 발생: {e}"

    def save_knowledge(self, topic: str, fact: str, source: str = "", scope: str = "global", verify_hint: str = "") -> str:
        """
        Saves a verified operational fact to the Imperial Knowledge Base, shared across ALL
        guardians. MUST be called whenever you discover/verify a new system fact (config path
        changed, port changed, new service location). Same topic updates the existing fact.
        Args:
            topic: Short subject (e.g., 'OpenClaw 실설정 경로')
            fact: The verified fact in present tense
            source: How it was verified (e.g., '실행 프로세스 커맨드라인 확인')
            scope: 'global' if true on any PC (product/architecture), 'environment' if only true on this machine (paths, running ports)
            verify_hint: How to re-verify this fact in a new environment (e.g., 'check_port 18789')
        """
        try:
            aegis_url, token = self._resolve_aegis()
            if not aegis_url:
                return "지식 저장 불가: Aegis URL 미설정. 저장됐다고 말하지 마라."
            import platform
            scope = scope if scope in ("global", "environment") else "global"
            payload = {"topic": topic, "fact": fact, "source": source,
                       "registered_by": "그림자 두뇌 (Shadow Brain)",
                       "scope": scope, "verify_hint": verify_hint}
            if scope == "environment":
                payload["environment_id"] = platform.node()
            resp = requests.post(
                f"{aegis_url}/api/knowledge",
                json=payload,
                headers={"X-Imperial-Token": token}, timeout=5,
            )
            data = resp.json() if resp.status_code == 200 else {}
            if data.get("status") == "SUCCESS":
                action = "갱신" if data.get("action") == "updated" else "신규 등록"
                return f"지식 베이스 저장 완료 ({action}): [{topic}]"
            return f"지식 저장 실패 (HTTP {resp.status_code}): {resp.text[:120]}. 저장됐다고 말하지 마라."
        except Exception as e:
            return f"지식 저장 실패: {e}. 저장됐다고 말하지 마라."

    def obsidian_search(self, query: str, limit: int = 8, source: str = None) -> str:
        """
        Searches the Obsidian vault (curated/distilled knowledge in markdown notes) — a
        knowledge layer separate from conversation memory. Use to find hand-organized notes,
        design docs, and reference material that the user has written. Works in both Local
        and Cloud (R2 Live) environments based on routing.
        Args:
            query: Keywords to search in note titles and bodies
            limit: Max notes to return (default 8)
            source: 'cloud' to search the R2 cloud vault (PGVector) even while running
                    locally; omit/None for the active vault (local-first).
        """
        try:
            import obsidian_router as ov
            n = int(limit or 8)
            if (source or "").strip().lower() == "cloud":
                vault = ov.get_cloud_vault()
                if vault is None:
                    return "클라우드(R2) 옵시디언 볼트가 비활성입니다(자격증명 없음). 단정하지 마라."
                # PGVector 의미 검색 우선, 장애(None)거나 결과가 비면([]) R2 키워드/파일명
                # 검색으로 보강한다. (예: 파일명 'ttt' 같은 리터럴 조회는 시맨틱보다 파일명
                # 매칭이 정확 — 클라우드는 노트 수가 적어 R2 스캔 비용도 작다.)
                # 🔑 임베딩(OpenAI 키)은 Aegis 경유라 aegis_url/token이 있어야 sync·검색이
                #   동작한다. 없으면 PGVector가 비고(쓰기 실패) 검색도 None → R2 폴백만 된다.
                _ctx = getattr(self, "_request_ctx", None)
                _aegis = getattr(_ctx, "aegis_url", None) if _ctx else None
                _tok = getattr(_ctx, "token", None) if _ctx else None
                results = None
                try:
                    import cloud_obsidian_vector
                    results = cloud_obsidian_vector.search(
                        query, limit=n, aegis_url=_aegis, token=_tok
                    )
                except Exception:
                    results = None
                if not results:
                    results = vault.search_notes(query, limit=n)
            else:
                vault = ov.get_vault()
                if vault is None:
                    return "옵시디언 볼트가 비활성입니다. 이 환경에서는 사용할 수 없으니 단정하지 마라."
                results = vault.search_notes(query, limit=n)
            if not results:
                return f"옵시디언 검색 결과: '{query}' 관련 노트 없음. 추측하지 말고 노트가 없다고 답하라."
            lines = []
            for i, r in enumerate(results, 1):
                lines.append(f"{i}. [{r['title']}] ({r['path']})\n   ↳ {r['snippet']}")
            return ("옵시디언 볼트 검색 결과:\n" + "\n".join(lines) +
                    "\n전체 본문이 필요하면 obsidian_read에 path를 넘겨 읽어라.")
        except Exception as e:
            return f"옵시디언 검색 실패: {e}. 결과를 단정하지 마라."

    def obsidian_read(self, path: str, source: str = None) -> str:
        """
        Reads the full body of a single Obsidian note by its vault-relative path (as returned by
        obsidian_search). Works in both Local and Cloud (R2 Live) environments.
        Args:
            path: Vault-relative path, e.g. 'projects/jarvis.md'
            source: 'cloud' to read from the R2 cloud vault even while running locally;
                    omit/None for the active vault (local-first).
        """
        try:
            import obsidian_router as ov
            vault = (
                ov.get_cloud_vault()
                if (source or "").strip().lower() == "cloud"
                else ov.get_vault()
            )
            if vault is None:
                return "옵시디언 볼트가 비활성입니다. 이 환경에서는 사용할 수 없다."
            rel, content = vault.read_note(path)
            if len(content) > 8000:
                content = content[:8000] + "\n…(이하 생략)"
            return f"옵시디언 노트 [{rel}]:\n\n{content}"
        except FileNotFoundError as e:
            return f"노트 없음: {e}"
        except Exception as e:
            return f"옵시디언 읽기 실패: {e}."

    def obsidian_write(self, path: str, content: str, mode: str = "create") -> str:
        """
        Creates/updates a markdown note in the Obsidian vault. MUST be actually CALLED (not just
        promised in text) whenever the user asks to create/write/save/organize a note, wiki, document
        or page in Obsidian. YOU generate the full markdown body and pass it in `content`. For a
        multi-page wiki, call repeatedly (one note per call) and link with [[ ]]. Works in both Local
        and Cloud (R2 Live) environments.
        Args:
            path: Vault-relative path (e.g. 'Wiki/태민게임즈/index.md'); '.md' added if missing
            content: Full markdown body you write (headings/[[backlinks]]/#tags), not a placeholder
            mode: 'create' (fail if exists) | 'overwrite' | 'append'
        """
        try:
            import re as _re
            import obsidian_router as ov
            vault = ov.get_vault()
            if vault is None:
                return "옵시디언 볼트가 비활성입니다. 저장됐다고 말하지 마라."
            mode = mode if mode in ("create", "overwrite", "append") else "create"
            # [🔱 경로 정제] Windows 금지 문자 제거 — [Errno 22] 방지
            # 폴더 구분자(/)는 유지, 나머지 금지 문자만 제거
            def _sanitize_path(p: str) -> str:
                # Windows 파일명 금지 문자: < > : " \ | ? * 그리고 유니코드 따옴표류
                parts = p.replace("\\", "/").split("/")
                clean_parts = []
                for part in parts:
                    part = _re.sub(r'[<>:"|?*\x00-\x1f\u2018\u2019\u201c\u201d\u0060]', "", part)
                    part = part.strip(". ")  # 앞뒤 점·공백 제거 (Windows 금지)
                    if part:
                        clean_parts.append(part)
                return "/".join(clean_parts) if clean_parts else "ollama_note"
            path = _sanitize_path(path)
            if not path.endswith(".md"):
                path += ".md"
            # [🛡️ 쓰기 규율] AI가 새로 만드는/덮어쓰는 노트가 frontmatter 없이 들어오면
            # 출처 메타데이터를 자동으로 찍는다. 검색 주입 시 '미검증'으로 라벨링되어
            # 환각이 단정적 근거로 굳는 것을 막는다. (append/이미 frontmatter 있는 노트는 보존)
            if mode in ("create", "overwrite") and not (content or "").lstrip().startswith("---"):
                from datetime import datetime, timezone
                _who = ""
                try:
                    _who = str((self.dynamic_config or {}).get("display_name") or "").strip()
                except Exception:
                    _who = ""
                _stamp = (
                    "---\n"
                    "source_type: ai-generated\n"
                    "verified: false\n"
                    f"recorded_by: {_who or 'shadow-brain'}\n"
                    f"recorded_at: {datetime.now(timezone.utc).isoformat()}\n"
                    "---\n\n"
                )
                content = _stamp + (content or "")
            saved = vault.write_note(path, content, mode=mode)
            return f"옵시디언 노트 저장 완료 ({mode}): [{saved}]"
        except FileExistsError as e:
            return f"이미 존재하는 노트입니다: {e}. overwrite/append 중 하나로 다시 시도하라."
        except Exception as e:
            return f"옵시디언 저장 실패: {e}. 저장됐다고 말하지 마라."

    def project_graph_query(self, query: str, mode: str = "depends_on") -> str:
        """
        Queries the Graphify knowledge graph (project map) of the code and documentation structure.
        Helps understand file dependencies, class methods, and how documents relate to code,
        saving token usage by targeting specific files.
        Args:
            query: Name of the file, module, or class to query (e.g., 'web_server.py', 'CodingToolkit')
            mode: 'depends_on' (import relations) | 'details' (metadata/methods) | 'all' (summary of graph)
        """
        try:
            import json
            from pathlib import Path
            
            # 1단계: 로컬 static/graph/graph.json 경로
            core_dir = Path(__file__).parent.parent
            graph_path = core_dir / "static" / "graph" / "graph.json"
            
            graph_data = None
            
            # 2단계: 로컬에 없거나 클라우드 환경인 경우 R2 Live에서 로드 시도
            import obsidian_router as ov
            vault_type = ov.get_vault_type()
            
            if vault_type == "cloud" or not graph_path.exists():
                try:
                    from cloud_tool import R2_ACCOUNTS, _get_bw_credential
                    import boto3
                    from botocore.config import Config
                    
                    live_cfg = R2_ACCOUNTS.get("live")
                    if live_cfg:
                        account_id = _get_bw_credential(live_cfg["account_id"])
                        access_key = _get_bw_credential(live_cfg["access_key_id"])
                        secret_key = _get_bw_credential(live_cfg["secret_access_key"])
                        bucket_name = _get_bw_credential(live_cfg.get("bucket_name", "taemingames"))
                        
                        if all([account_id, access_key, secret_key]):
                            s3 = boto3.client(
                                "s3",
                                endpoint_url=f"https://{account_id}.r2.cloudflarestorage.com",
                                aws_access_key_id=access_key,
                                aws_secret_access_key=secret_key,
                                config=Config(signature_version="s3v4")
                            )
                            response = s3.get_object(Bucket=bucket_name, Key="cloud-obsidian/graph/graph.json")
                            graph_data = json.loads(response["Body"].read().decode("utf-8"))
                except Exception as r2_err:
                    logger.warning(f"Failed to load graph from R2: {r2_err}")
            
            # 로컬 경로 백업 로드
            if not graph_data and graph_path.exists():
                with open(graph_path, "r", encoding="utf-8") as f:
                    graph_data = json.load(f)
                    
            if not graph_data:
                return "지식 그래프 데이터(graph.json)를 로드할 수 없습니다. scripts/build_graph.py를 먼저 빌드하십시오."
                
            nodes = graph_data.get("nodes", [])
            links = graph_data.get("links", [])
            
            if mode == "all":
                types = {}
                for n in nodes:
                    t = n.get("type", "unknown")
                    types[t] = types.get(t, 0) + 1
                type_summary = ", ".join([f"{k}: {v}개" for k, v in types.items()])
                return f"🔱 제국 지식 그래프 전체 요약:\n- 총 노드: {len(nodes)}개 ({type_summary})\n- 총 연결(의존성): {len(links)}개"
                
            target = query.strip()
            found_node = None
            for n in nodes:
                if n["id"] == target or n["label"] == target or target in n["id"]:
                    found_node = n
                    break
                    
            if not found_node:
                return f"지식 그래프에서 '{target}'에 해당하는 노드를 찾을 수 없습니다. (모듈명/클래스명 오타 확인 필요)"
                
            node_id = found_node["id"]
            
            if mode == "details":
                details_str = json.dumps(found_node.get("details", {}), ensure_ascii=False, indent=2)
                return f"📊 노드 상세 정보 [{found_node['label']} ({found_node['type']})]:\nID: {node_id}\n상세:\n{details_str}"
                
            incoming = []
            outgoing = []
            for link in links:
                src = link["source"]
                tgt = link["target"]
                
                if isinstance(src, dict): src = src.get("id")
                if isinstance(tgt, dict): tgt = tgt.get("id")
                
                if src == node_id:
                    outgoing.append(f"-> [{tgt}] ({link.get('type', 'dependency')})")
                elif tgt == node_id:
                    incoming.append(f"<- [{src}] ({link.get('type', 'dependency')})")
                    
            out_str = "\n".join(outgoing) if outgoing else "없음"
            in_str = "\n".join(incoming) if incoming else "없음"
            
            return f"🔱 지식 그래프 의존 관계 [{found_node['label']} ({found_node['type']})]:\n\n[출력 의존성 (내가 참조하는 것)]\n{out_str}\n\n[입력 의존성 (나를 참조하는 것)]\n{in_str}"
            
        except Exception as e:
            return f"지식 그래프 쿼리 중 오류 발생: {e}"

    def review_knowledge_proposals(self) -> str:
        """
        [Curator] Lists pending knowledge proposals from other guardians awaiting your review.
        You are the Imperial Knowledge librarian: verify each proposal with observation tools
        (check_port/find_process) before approving via judge_knowledge_proposal.
        """
        try:
            aegis_url, token = self._resolve_aegis()
            if not aegis_url:
                return "제안 목록 조회 불가: Aegis URL 미설정."
            resp = requests.get(f"{aegis_url}/api/knowledge/propose",
                                headers={"X-Imperial-Token": token}, timeout=5)
            data = resp.json() if resp.status_code == 200 else {}
            results = data.get("results", [])
            if not results:
                return "대기 중인 지식 제안 없음."
            lines = []
            for r in results:
                lines.append(f"- id={r['id']} [{r.get('topic')}|{r.get('scope')}] {r.get('fact')[:150]} "
                             f"(제안: {r.get('proposed_by')}, 검증법: {r.get('verify_hint') or '없음'})")
            return ("대기 중인 지식 제안 목록:\n" + "\n".join(lines) +
                    "\n규칙: 각 제안을 관측 도구로 직접 검증한 뒤 judge_knowledge_proposal로 승인/기각하라. "
                    "검증 불가능한 제안은 승인하지 마라.")
        except Exception as e:
            return f"제안 목록 조회 실패: {e}"

    def judge_knowledge_proposal(self, proposal_id: str, approve: bool, note: str = "") -> str:
        """
        [Curator] Approves or rejects a pending knowledge proposal. ONLY approve after you have
        verified the fact yourself (observation tools or trusted source). Approval replaces any
        existing approved fact on the same topic.
        Args:
            proposal_id: The proposal id from review_knowledge_proposals
            approve: true to approve (becomes official knowledge), false to reject
            note: Verification basis or rejection reason
        """
        try:
            aegis_url, token = self._resolve_aegis()
            if not aegis_url:
                return "심사 불가: Aegis URL 미설정."
            resp = requests.post(f"{aegis_url}/api/knowledge/review",
                                 json={"id": proposal_id, "approve": bool(approve), "note": note},
                                 headers={"X-Imperial-Token": token}, timeout=5)
            data = resp.json() if resp.status_code == 200 else {}
            if data.get("status") == "SUCCESS":
                return f"심사 완료: {data.get('action')} (id={proposal_id})"
            return f"심사 실패 (HTTP {resp.status_code}): {resp.text[:120]}"
        except Exception as e:
            return f"심사 실패: {e}"

    def _begin_ask_user(self, args: dict) -> dict:
        """🙋 ask_user 도구 시작 — 모듈 함수 begin_ask 위임."""
        return begin_ask(args)

    def _store_web_recon(self, func_name: str, args: dict, result) -> None:
        """🕸️ hugin 정찰 결과를 web_recon 카테고리 벡터 메모리에 저장."""
        mem_text = f"Action: {func_name}, Args: {args}, Result: {str(result)[:500]}"
        sovereign_memory.memory_write_vector(mem_text, category="web_recon")

    def _wait_ask_user(self, ask_id: str, timeout: int):
        """답변 대기 — 모듈 함수 wait_ask 위임."""
        return wait_ask(ask_id, timeout)

    def check_port(self, port: int) -> str:
        """
        Read-only observation: checks if a local TCP port is listening RIGHT NOW and which
        process owns it. MUST be used to verify environment-scoped knowledge (ports, running
        services) before asserting it as current fact. Never changes system state.
        Args:
            port: TCP port number to check (e.g., 18789)
        """
        try:
            import psutil
            port = int(port)
            for conn in psutil.net_connections(kind="tcp"):
                if conn.laddr and conn.laddr.port == port and conn.status == psutil.CONN_LISTEN:
                    info = f"포트 {port}: LISTENING"
                    if conn.pid:
                        try:
                            p = psutil.Process(conn.pid)
                            cmd = " ".join(p.cmdline())[:300]
                            info += f" (PID {conn.pid}, 프로세스: {p.name()}, 커맨드: {cmd})"
                        except Exception:
                            info += f" (PID {conn.pid})"
                    return info + "\n→ 지금 이 환경에서 직접 관측한 사실이므로 단정해도 된다."
            return (f"포트 {port}: 현재 이 환경에서 수신 대기 중인 프로세스 없음. "
                    "해당 서비스가 꺼져 있거나 다른 포트를 쓴다. 켜져 있다고 단정하지 마라.")
        except Exception as e:
            return f"포트 관측 실패: {e}. 관측이 안 됐으므로 상태를 단정하지 마라."

    def find_process(self, keyword: str, limit: int = 5) -> str:
        """
        Read-only observation: finds running processes whose name or command line contains the
        keyword, and shows their command lines (reveals actual config paths in use). MUST be
        used to verify which config/install a running service actually uses. Never changes state.
        Args:
            keyword: Substring to match in process name or command line (e.g., 'openclaw')
            limit: Max processes to return (default 5)
        """
        try:
            import psutil
            kw = str(keyword or "").lower()
            if len(kw) < 2:
                return "관측 실패: 키워드는 2자 이상이어야 한다."
            hits = []
            for p in psutil.process_iter(["pid", "name", "cmdline"]):
                try:
                    name = (p.info.get("name") or "").lower()
                    cmd = " ".join(p.info.get("cmdline") or [])
                    if kw in name or kw in cmd.lower():
                        hits.append(f"- PID {p.info['pid']} [{p.info.get('name')}] {cmd[:300]}")
                        if len(hits) >= int(limit or 5):
                            break
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    continue
            if not hits:
                return (f"'{keyword}' 관련 실행 중 프로세스 없음 (지금 이 환경 기준). "
                        "실행 중이라고 단정하지 마라.")
            return (f"'{keyword}' 실행 중 프로세스 (지금 이 환경에서 직접 관측):\n" + "\n".join(hits) +
                    "\n→ 커맨드라인에 보이는 경로가 실제 사용 중인 설정/설치 경로다.")
        except Exception as e:
            return f"프로세스 관측 실패: {e}. 관측이 안 됐으므로 상태를 단정하지 마라."

    def recall_memory(self, query: str, limit: int = 5) -> str:
        """
        Searches the user's long-term conversation memory (Aegis vector store). MUST be used
        before answering any question about past conversations, user preferences, or shared history.
        If this returns no results, honestly say you don't remember instead of guessing.
        Args:
            query: What to search for in memory (e.g., '생일', '좋아하는 음식', '어제 한 약속')
            limit: Max number of memories to return (default 5)
        """
        try:
            ctx = getattr(self, "_request_ctx", None)
            aegis_url = getattr(ctx, "aegis_url", None) if ctx else None
            token = getattr(ctx, "token", None) if ctx else None
            session_id = getattr(ctx, "session_id", None) if ctx else None
            search_context = session_id if session_id and session_id != "GLOBAL" else None
            results = sovereign_memory.memory_search_vector(
                query, limit=int(limit or 5), context_id=search_context,
                aegis_url=aegis_url, token=token,
            )
            if results is None:
                # 🧠 [정직 게이트] 임베딩 장애 — "기억 없음"과 구분해 장애를 고지
                notice = sovereign_memory.memory_health_notice() or "기억 시스템 장애"
                return (
                    f"기억 검색 불능: {notice} "
                    "규칙: '기억에 없다'고 단정하지 말고, 현재 기억 기능에 장애가 있어 "
                    "저장도 검색도 되지 않는 상태임을 솔직하게 알려라. "
                    "절대 '기억하겠다/저장했다'고 약속하지 마라."
                )
            if not results:
                return "기억 저장소 검색 결과: 관련 기억 없음. 아는 척하지 말고 '기억에 없어요, 알려주시면 기억할게요'라고 답하라."
            lines = []
            for i, r in enumerate(results, 1):
                content = str(r.get("content", "")).strip()
                if len(content) > 300:
                    content = content[:300] + "…"
                lines.append(f"{i}. {content}")
            return "기억 저장소 검색 결과 (실제 저장된 기억):\n" + "\n".join(lines)
        except Exception as e:
            return f"기억 검색 실패: {e}. 검색이 안 됐으므로 과거 사실을 단정하지 마라."

    def count_history(self, query: str = "", message_type: int = -1) -> str:
        """
        Counts real chat history records in the Aegis database. MUST be used for questions like
        'how many times did I say X' or 'when did we first talk'. Never guess numbers.
        Args:
            query: Text to count occurrences of (e.g., '안녕'). Empty counts all messages.
            message_type: 0 = user messages only, 1 = AI messages only, -1 = all (default)
        """
        try:
            ctx = getattr(self, "_request_ctx", None)
            aegis_url = (getattr(ctx, "aegis_url", None) if ctx else None) or "http://localhost:18701"
            token = getattr(ctx, "token", None) if ctx else None
            session_id = getattr(ctx, "session_id", None) if ctx else None
            params = {}
            if query:
                params["q"] = query
            if session_id and session_id != "GLOBAL":
                params["context_id"] = session_id
            if message_type in (0, 1):
                params["message_type"] = message_type
            auth = token or os.environ.get("IMPERIAL_API_TOKEN") or os.environ.get("SHADOWBRAIN_TOKEN") or ""
            resp = requests.get(
                f"{aegis_url.rstrip('/')}/api/chat/history/stats",
                params=params,
                headers={"X-Imperial-Token": auth, "Authorization": f"Bearer {auth}"},
                timeout=10,
            )
            if resp.status_code != 200:
                return f"기록 집계 실패 (HTTP {resp.status_code}). 집계가 안 됐으므로 숫자를 지어내지 마라."
            data = resp.json()
            return (
                f"대화 기록 DB 집계 결과 (실제 데이터): 총 {data.get('count', 0)}건"
                + (f", 검색어 '{query}'" if query else "")
                + (f", 최초 {data.get('first_at')}" if data.get('first_at') else "")
                + (f", 최근 {data.get('last_at')}" if data.get('last_at') else "")
            )
        except Exception as e:
            return f"기록 집계 실패: {e}. 숫자를 지어내지 말고 확인 불가라고 답하라."

    def jarvis_memory(self, action: str, content: str = None, category: str = "general", key: str = None, query: str = None) -> str:
        """
        Manages Imperial Permanent Memory (Taemin Empire).
        Args:
            action: 'write', 'read', 'search', 'list', 'delete'
            content: Content to write
            category: Memory category (e.g., 'user_pref', 'system_log')
            key: Specific key for direct read/write
            query: Semantic search query
        """
        args = ["--action", action]
        if content: args += ["--content", content]
        if category: args += ["--category", category]
        if key: args += ["--key", key]
        if query: args += ["--query", query]
        return self._execute_cli("jarvis_memory.py", args)

    def jarvis_system(self, target: str = "all") -> str:
        """
        Retrieves Imperial System Information (Guardians, Status, etc.).
        Args:
            target: 'all', 'guardians', 'cpu', 'memory', 'disk', 'network'
        """
        return self._execute_cli("jarvis_system.py", ["--target", target])

    def jarvis_skill(self, action: str, name: str = None, content: str = None) -> str:
        """
        Manages Imperial Skills (Protocols).
        Args:
            action: 'list', 'read', 'create'
            name: Skill name
            content: Skill content (for create)
        """
        args = ["--action", action]
        if name: args += ["--name", name]
        if content: args += ["--content", content]
        return self._execute_cli("jarvis_skill.py", args)

    def jarvis_soul(self, status: str, result_summary: str) -> str:
        """
        Reports high-level task completion or soul-sync status.
        Args:
            status: 'success', 'failure', 'in_progress'
            result_summary: Brief summary of the result
        """
        return self._execute_cli("jarvis_soul.py", ["--status", status, "--summary", result_summary])

    def jarvis_gbb_post(self, content: str) -> str:
        """
        Posts a message to the Imperial Guardian Bulletin Board (GBB).
        Args:
            content: The message content to post.
        """
        try:
            from aegis_link import neural_link
            
            token = getattr(self, "auth_token", None) or os.environ.get("IMPERIAL_API_TOKEN")
            aegis_url = getattr(self, "aegis_url", None)
            
            logger.info(f"🔱 [GBB Post] Sending announcement via REST: {content[:30]}...")
            result_data = neural_link.post_announcement(content, aegis_url=aegis_url, token=token)
            
            if result_data:
                msg = f"✅ GBB 게시 성공 (ID: {result_data.get('id', 'unknown')})"
                self._afc_tool_results.append(msg)
                return msg
            else:
                return "❌ GBB 게시 실패 (Aegis Oracle 응답 없음)"
        except Exception as e:
            logger.error(f"[GBB Post] Error: {e}")
            return f"Error: GBB 게시 실패 - {str(e)}"

    def generate_image(self, prompt: str, model: str = "imagen-4.0-generate-001", aspect_ratio: str = "1:1") -> str:
        """
        마왕님의 요청에 따라 AI 이미지를 생성하고 채팅창에 표시할 마크다운 이미지 URL을 반환합니다.
        이미지 생성, 그림 그리기, 사진 만들기 등의 요청에 사용하세요.
        Args:
            prompt: 생성할 이미지의 상세한 영문 프롬프트 (예: 'cinematic portrait of a dragon knight')
            model: 사용할 모델 ID. 선택지: 'imagen-4.0-generate-001'(기본), 'nano-banana'(고속), 'z_image_base'(로컬/고품질)
            aspect_ratio: 이미지 비율. '1:1'(정사각형), '16:9'(가로), '9:16'(세로), '4:3', '3:4'
        """
        try:
            # [IMPERIAL INTELLIGENCE] 자동 모델 선택 로직 (프롬프트 키워드 기반)
            low_prompt = prompt.lower()
            if any(k in low_prompt for k in ["나노바나나", "nanobanana", "nano-banana", "nano banana"]):
                model = "nano-banana"
                logger.info("[ImageGen] '나노바나나' 키워드 감지로 모델을 'nano-banana'로 자동 전환합니다.")
            elif any(k in low_prompt for k in ["헤파이스토스", "hephaestus", "로컬엔진", "z-image", "z_image"]):
                model = "z_image_base"
                logger.info("[ImageGen] '헤파이스토스' 관련 키워드 감지로 모델을 'z_image_base'로 자동 전환합니다.")

            logger.info(f"[ImageGen] 이미지 생성 시작: prompt='{prompt[:60]}', model={model}, ratio={aspect_ratio}")
            proj_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
            # [🔱 Imperial Fix] 이미지 생성 엔드포인트는 같은 프로세스(web_server)에 등록돼 있다.
            # 따라서 로컬은 18700, 배포(HF/gunicorn)는 $PORT(예: 7860)로 자기 자신을 호출해야 한다.
            # 과거 18700 하드코딩 탓에 배포 환경에서 'Connection Refused'(통신 연결 거부) 환각이 발생했다.
            default_port = os.environ.get("PORT", "18700")
            backend_url = os.environ.get("JARVIS_BACKEND_URL", f"http://127.0.0.1:{default_port}")

            resp = requests.post(
                f"{backend_url}/api/studio/generate-image",
                json={"prompt": prompt, "model_id": model, "aspect_ratio": aspect_ratio, "engine": "gemini"},
                timeout=300,
                headers={"Content-Type": "application/json"}
            )
            if resp.status_code == 200:
                data = resp.json()
                filename = data.get("filename", "")
                if filename:
                    # [🔱 Imperial Fix] 배포 환경은 로컬 디스크가 휘발성이므로 엔드포인트가 R2 공개 URL을
                    # 반환한다. 절대 URL(http...)이면 그대로 쓰고, 상대경로면 backend_url을 붙인다.
                    returned_url = data.get("url", "")
                    if isinstance(returned_url, str) and returned_url.startswith("http"):
                        image_url = returned_url
                    else:
                        image_url = f"{backend_url}/api/studio/images/{filename}"
                    result = f"✅ 이미지 생성 완료!\n\n![생성된 이미지]({image_url})"
                    self._afc_tool_results.append(result)
                    logger.info(f"[ImageGen] 성공: {filename} → {image_url}")
                    return result
                result = f"⚠️ [ImageGen] 이미지 생성 성공했지만 파일명을 받지 못했습니다. Response: {data}"
                self._afc_tool_results.append(result)
                return result
            else:
                err = resp.json().get('error', resp.text[:200])
                result = f"❌ [ImageGen] 이미지 생성 실패 (HTTP {resp.status_code}): {err}"
                self._afc_tool_results.append(result)
                return result
        except requests.exceptions.Timeout:
            result = "❌ [ImageGen] 이미지 생성 타임아웃 (5분 초과). 더 짧은 프롬프트를 사용하거나 다시 시도하세요."
            self._afc_tool_results.append(result)
            return result
        except Exception as e:
            logger.error(f"[ImageGen] 예외 오류: {e}")
            result = f"❌ [ImageGen] 이미지 생성 중 오류 발생 - {str(e)}"
            self._afc_tool_results.append(result)
            return result

    def sovereign_type_text(self, text: str, press_enter: bool = True) -> str:
        """
        Types text into the Master's active window or focused chat input field.
        ⚠️ WARNING: DO NOT use this tool to launch applications (e.g. typing 'notepad').
        Use 'sovereign_run_app' for launching system applications instead.
        Args:
            text: The text to type (chat messages, text fields, etc.).
            press_enter: Whether to press 'enter' automatically after typing. Default is True.
        """
        try:
            from remote_vision_service import remote_type, remote_hotkey
            success = remote_type(text)
            if success and press_enter:
                import time
                time.sleep(0.1) # Small delay before pressing enter
                remote_hotkey('enter')
                result = f"✅ Typed text '{text[:20]}...' and pressed ENTER."
            else:
                result = f"✅ Typed text '{text[:20]}...' directly via OS (No Enter)." if success else f"❌ Failed to type text via OS."
            
            self._afc_tool_results.append(result)
            logger.info(f"[Sovereign Tool] {result}")
            return result
        except Exception as e:
            msg = f"❌ Sovereign Type Error: {e}"
            self._afc_tool_results.append(msg)
            return msg

    def sovereign_run_app(self, app_name: str) -> str:
        """
        Launches an OS application directly. 
        Use this for 'notepad', 'calc', 'chrome', 'cmd', etc.
        Args:
            app_name: The name or path of the application to run.
        """
        try:
            import platform
            from engines.windows import WindowsEngine
            from engines.macos import MacOSEngine
            
            engine = WindowsEngine() if platform.system() == "Windows" else MacOSEngine()
            success = engine.open_file_or_url(app_name)
            
            result = f"✅ Launched application: {app_name}" if success else f"❌ Failed to launch application: {app_name}"
            self._afc_tool_results.append(result)
            logger.info(f"[Sovereign Tool] {result}")
            return result
        except Exception as e:
            msg = f"❌ Sovereign RunApp Error: {e}"
            self._afc_tool_results.append(msg)
            return msg

    def sovereign_press_key(self, keys: str) -> str:
        """
        Presses a keyboard key directly in the Master's OS.
        Use 'enter' to submit a chat message after typing.
        Args:
            keys: The key to press (e.g., 'enter', 'esc').
        """
        try:
            from remote_vision_service import remote_hotkey
            raw_keys = [k.strip() for k in keys.split('+')] if '+' in keys else keys
            success = remote_hotkey(raw_keys)
            result = f"✅ Pressed '{keys}' directly via OS." if success else f"❌ Failed to press '{keys}' via OS."
            self._afc_tool_results.append(result)
            logger.info(f"[Sovereign Tool] {result}")
            return result
        except Exception as e:
            msg = f"❌ Sovereign Key Error: {e}"
            self._afc_tool_results.append(msg)
            return msg

    def delegate_team_task(self, title: str, assignee_guardian: str, description: str = "", priority: str = "MEDIUM", requester_id: str = None) -> str:
        """
        Creates and delegates a task to a specific team member (guardian 요원).
        Use this tool when you need to record a formal task for later execution.
        Args:
            title: Brief task title (e.g., 'Optimize R2 upload pipeline')
            assignee_guardian: The guardian name to assign (e.g., 'Acubens', 'Hermes', 'Hydra', 'Atlas')
            description: Detailed description of what needs to be done
            priority: Task priority - 'LOW', 'MEDIUM', 'HIGH', 'CRITICAL'
            requester_id: 명의자 (생략 시 현재 대화 상대방 사용)
        """
        try:
            backend_url = os.environ.get("JARVIS_BACKEND_URL", "http://127.0.0.1:18701")
            
            # [🔱 Identity Alignment]
            actual_requester = requester_id or self.current_sender or "shadow_brain"
            
            payload = {
                "task_content": f"[{title}] {description}",
                "executor_id": assignee_guardian,
                "priority": priority,
                "delegated_by": f"{actual_requester} (via Shadow Brain)",
            }
            
            # [🛡️] Use Dynamic Delegated Token for authentication
            headers = {"Content-Type": "application/json"}
            token_to_use = getattr(self, "auth_token", None) or os.environ.get("IMPERIAL_API_TOKEN")
            if not token_to_use:
                # [🔱 V7.3 Unified Auth]
                token_to_use = os.environ.get("IMPERIAL_API_TOKEN") or sovereign_memory.get_secret("SHADOWBRAIN_TOKEN")
            
            if token_to_use:
                headers["X-Guardian-Token"] = token_to_use
                headers["X-Imperial-Token"] = token_to_use
            
            resp = requests.post(
                f"{backend_url}/api/team/tasks", # [🔱 Fix] Correct Aegis V3 endpoint
                json=payload,
                headers=headers,
                timeout=30,
            )
            
            if resp.status_code in (200, 201):
                data = resp.json()
                task_id = data.get("task_id", data.get("id", "unknown"))
                result = f"✅ 제국 팀 테스크 기록 완료!\n• ID: {task_id}\n• 제목: {title}\n• 담당: {assignee_guardian}\n• 우선도: {priority}"
                self._afc_tool_results.append(result)
                logger.info(f"[Team Delegation] Task '{title}' registered for {assignee_guardian}")
                return result
            else:
                err = resp.text[:200]
                result = f"❌ 팀 테스크 하달 실패 (HTTP {resp.status_code}): {err}"
                self._afc_tool_results.append(result)
                return result
        except Exception as e:
            msg = f"❌ Team Delegation Error: {e}"
            self._afc_tool_results.append(msg)
            logger.error(f"[Team Delegation] {msg}")
            return msg

    def dispatch_to_hydra(self, mission_goal: str, requester_id: str = None) -> str:
        """
        Directly dispatches a mission to Hydra for autonomous parallel planning and execution.
        Use this tool when you need an immediate synthesized report from multiple agents (Acubens).
        Args:
            mission_goal: The high-level objective for Hydra to analyze and decompose.
            requester_id: 요청자 명의 (생략 시 '마왕님' 또는 현재 대화 상대)
        """
        try:
            logger.info(f"[🔱 Hydra Dispatch] Initializing mission: {mission_goal}")
            
            # [🔱 Identity Alignment]
            actual_requester = requester_id or self.current_sender or "shadow_brain"
            
            hydra_url = os.environ.get("HYDRA_API_URL", "http://localhost:8002/hydra/dispatch")
            
            # [🛡️] Propagate delegated token to Hydra
            headers = {"Content-Type": "application/json"}
            token_to_use = getattr(self, "auth_token", None) or os.environ.get("IMPERIAL_API_TOKEN")
            if token_to_use:
                headers["X-Imperial-Token"] = token_to_use

            payload = {
                "goal": mission_goal,
                "requester_id": actual_requester
            }

            resp = requests.post(hydra_url, json=payload, headers=headers, timeout=300) # 5분 타임아웃
            
            if resp.status_code == 200:
                data = resp.json()
                report = data.get("synthesis", "보고서 생성 실패")
                result = f"🐲 **[히드라 자율 기획 보고서]**\n\n{report}"
                self._afc_tool_results.append(result)
                return result
            elif resp.status_code == 503:
                return "⚠️ [히드라 연결 실패] 히드라 서버가 응답하지 않습니다. 수호자 히드라를 먼저 기동(RunHydra.bat)해 주십시오."
            else:
                err = resp.text[:300]
                return f"❌ [히드라 오류] HTTP {resp.status_code}: {err}"
        except requests.exceptions.ConnectionError:
            return "⚠️ [히드라 오프라인] 히드라 서버(Port 8002)에 연결할 수 없습니다. 수호자 히드라를 깨워주십시오."
        except Exception as e:
            logger.error(f"[Hydra Dispatch] Exception: {e}")
            return f"❌ [Hydra Dispatch] 내부 오류 발생: {str(e)}"

    def _emit_token_usage(self, provider: str, model_name: str, usage):
        """[🪙 Imperial Token Ledger] API 호출 1건의 토큰 사용량을 Aegis에 비동기 적재.

        usage: OpenAI 호환 usage dict 또는 정규화된 dict.
        지원 키(자동 인식):
          - prompt_tokens / input_tokens / prompt_token_count
          - completion_tokens / output_tokens / candidates_token_count
          - total_tokens / total_token_count
          - prompt_tokens_details.cached_tokens / cache_read_input_tokens / cached_content_token_count
        스트리밍 응답에서 공급자가 usage를 보고하지 않으면 조용히 무시한다(노이즈 방지).
        """
        try:
            if not usage:
                return
            if not isinstance(usage, dict):
                # google genai usage_metadata 등 객체 → dict 정규화
                usage = {
                    "prompt_token_count": getattr(usage, "prompt_token_count", None),
                    "candidates_token_count": getattr(usage, "candidates_token_count", None),
                    "total_token_count": getattr(usage, "total_token_count", None),
                    "cached_content_token_count": getattr(usage, "cached_content_token_count", None),
                }

            def _pick(*keys):
                for k in keys:
                    v = usage.get(k)
                    if v is not None:
                        try:
                            return int(v)
                        except (TypeError, ValueError):
                            continue
                return 0

            prompt_t = _pick("prompt_tokens", "input_tokens", "prompt_token_count")
            completion_t = _pick("completion_tokens", "output_tokens", "candidates_token_count")
            total_t = _pick("total_tokens", "total_token_count") or (prompt_t + completion_t)

            cached_t = 0
            details = usage.get("prompt_tokens_details")
            if isinstance(details, dict):
                cached_t = int(details.get("cached_tokens") or 0)
            if not cached_t:
                cached_t = _pick("cache_read_input_tokens", "cached_content_token_count")

            if total_t <= 0:
                return

            session_id = getattr(self._request_ctx, "session_id", None) or "GLOBAL"
            token = (getattr(self._request_ctx, "token", None)
                     or getattr(self, "auth_token", None)
                     or os.environ.get("IMPERIAL_API_TOKEN", ""))
            aegis_url = getattr(self._request_ctx, "aegis_url", None) or "http://localhost:18701"
            aegis_url = aegis_url.rstrip("/")

            provider_key = str(provider or "").strip().lower()
            model_key = str(model_name or "").strip().lower()
            is_billable = (
                provider_key not in {"ollama", "atlas", "local"}
                and not model_key.endswith(":free")
            )
            payload = {
                "context_id": session_id,
                "usage_type": "chat",
                "billable": is_billable,
                "provider": provider,
                "model": model_name,
                "prompt_tokens": prompt_t,
                "completion_tokens": completion_t,
                "total_tokens": total_t,
                "cached_tokens": cached_t,
            }

            def _post():
                try:
                    headers = {"Content-Type": "application/json"}
                    if token:
                        headers["Authorization"] = f"Bearer {token}"
                        headers["X-Imperial-Token"] = token
                    r = requests.post(
                        f"{aegis_url}/api/token-usage",
                        json=payload, headers=headers, timeout=10,
                    )
                    if r.status_code != 200:
                        logger.warning(
                            f"[🪙 TokenLedger] 적재 실패 HTTP {r.status_code}: "
                            f"{r.text[:200]} (url={aegis_url}/api/token-usage, has_token={bool(token)})"
                        )
                    else:
                        logger.info(f"[🪙 TokenLedger] 적재 성공: {r.text[:120]}")
                except Exception as _e:
                    logger.warning(f"[🪙 TokenLedger] 적재 예외(무시): {_e}")

            threading.Thread(target=_post, daemon=True).start()
            logger.info(
                f"[🪙 TokenLedger] {provider}/{model_name} "
                f"in={prompt_t} out={completion_t} total={total_t} cached={cached_t} ctx={session_id[:16]}"
            )
        except Exception as e:
            logger.debug(f"[TokenLedger] emit 실패(무시): {e}")

    def _resolve_shadow_brain_user_id(self, aegis_url=None, token=None) -> str | None:
        """[🔱 Stateless Oracle] Aegis API를 통해 Shadow Brain의 user_id를 확인합니다."""
        try:
            from aegis_link import neural_link
            shadow_persona_id = "a8e1a1af-f28b-47ab-b371-193082508638"
            
            p_data = neural_link.get_guardian_identity(shadow_persona_id, aegis_url=aegis_url, token=token)
            if p_data:
                owner_id = p_data.get("owner_user_id")
                if owner_id:
                    logger.info(f"[GBB] Shadow Brain user_id resolved: {owner_id[:8]}...")
                    return owner_id
            
            # Fallback to persona ID itself if owner lookup fails
            return shadow_persona_id
        except Exception as e:
            logger.error(f"[GBB] Shadow Brain identity resolution failed: {e}")
            return "a8e1a1af-f28b-47ab-b371-193082508638"

    def _get_vsc_machine_id(self) -> str:
        """Generates a stable, pseudo-random hex string for VS Code identity mimicry."""
        import socket
        try:
            device_seed = socket.gethostname() or "Imperial-Sentinel-Core"
        except:
            device_seed = "Imperial-Sentinel-Core"
        return hashlib.sha256(device_seed.encode()).hexdigest()

    def _get_copilot_token(self):
        """Exchanges GITHUB_TOKEN for a Copilot session token or reads local native session."""
        if self._copilot_token and self._copilot_expires > time.time() + 60:
            return self._copilot_token, self._copilot_url
            
        # [IMPERIAL EXTENSION] Read directly from Shared Relay, Aizen, or Gilgamesh credentials
        base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "docker_data"))
        token_paths = [
            os.path.join(base_dir, "shared-relay", "credentials", "github-copilot.token.json"),
            os.path.join(base_dir, "openclaw", "aizen", "credentials", "github-copilot.token.json"),
            os.path.join(base_dir, "openclaw", "gilgamesh", "credentials", "github-copilot.token.json")
        ]
        
        for tp in token_paths:
            if os.path.exists(tp):
                try:
                    with open(tp, 'r', encoding='utf-8') as f:
                        data = json.load(f)
                        tok = data.get("token")
                        exp = data.get("expiresAt", 0)
                        if exp > 10000000000:
                            exp /= 1000
                        if tok and exp > time.time() + 60:
                            self._copilot_token = tok
                            self._copilot_expires = exp
                            match = re.search(r"(?:^|;)\s*proxy-ep=([^;\s]+)", tok, re.IGNORECASE)
                            if match:
                                proxy_ep = match.group(1).strip()
                                host = proxy_ep.replace("https://", "").replace("http://", "").replace("proxy.", "api.")
                                self._copilot_url = f"https://{host}"
                            logger.info(f"Imperial Guardian successfully synchronized Copilot soul from: {os.path.basename(os.path.dirname(os.path.dirname(tp)))}")
                            return self._copilot_token, self._copilot_url
                except Exception as e:
                    logger.warning(f"Failed to read token from {tp}: {e}")

        # [IMPERIAL EXTENSION] 2. Prioritize PROJECT ENV GITHUB_TOKEN or read permanent OAuth from profiles
        _project_root_env = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".env"))
        from dotenv import dotenv_values
        env_dict = dotenv_values(_project_root_env)
        oauth_token = os.environ.get("GITHUB_TOKEN") or env_dict.get("GITHUB_TOKEN")
        
        if not oauth_token or "your_github" in oauth_token:
            profile_paths = [
                os.path.join(base_dir, "shared-relay", "auth-profiles.json"),
                os.path.join(os.path.expanduser("~"), ".openclaw", "auth-profiles.json")
            ]
            
            for pp in profile_paths:
                if os.path.exists(pp):
                    try:
                        with open(pp, 'r', encoding='utf-8') as f:
                            p_data = json.load(f)
                            oauth = p_data.get("profiles", {}).get("github-copilot:github", {}).get("token")
                            if oauth and (oauth.startswith("ghu_") or oauth.startswith("gho_")):
                                oauth_token = oauth
                                logger.info(f"Imperial Guardian acquired permanent OAuth token from profile: {pp}")
                                break
                    except Exception:
                        pass

        if not oauth_token:
            logger.info("No valid local Copilot session or profile found. Falling back to GITHUB_TOKEN api...")
            github_token = os.environ.get("GITHUB_TOKEN")
            if not github_token:
                return None, None
            oauth_token = github_token
            
        try:
            res = requests.get("https://api.github.com/copilot_internal/v2/token", headers={
                "Authorization": f"Bearer {oauth_token}",
                "Accept": "application/json",
                "Editor-Version": "vscode/1.93.0",
                "Editor-Plugin-Version": "copilot/1.222.0",
                "User-Agent": "GithubCopilot/1.222.0"
            }, timeout=10)
            res.raise_for_status()
            data = res.json()
            
            self._copilot_token = data.get("token")
            self._copilot_expires = data.get("expires_at", 0)
            if self._copilot_expires > 10000000000: # ms to s
                self._copilot_expires /= 1000
                
            match = re.search(r"(?:^|;)\s*proxy-ep=([^;\s]+)", self._copilot_token, re.IGNORECASE)
            if match:
                proxy_ep = match.group(1).strip()
                host = proxy_ep.replace("https://", "").replace("http://", "").replace("proxy.", "api.")
                self._copilot_url = f"https://{host}"
                
            logger.info("Successfully minted new Copilot Session Token using OAuth Device Token.")
            return self._copilot_token, self._copilot_url
        except Exception as e:
            logger.error(f"Copilot Token Error: {e}")
            return None, None

    def _get_client(self, aegis_url=None, token=None):
        """🔱 [Imperial] 요청 컨텍스트에 맞는 genai Client를 반환합니다."""
        # 1. 특정 Aegis URL이 주입된 경우, 해당 환경의 키를 수혈받음
        target_key = self.api_key
        if aegis_url:
            # sovereign_memory를 통해 해당 Aegis의 키를 가져옴
            from sovereign_memory import get_secret
            # [🛡️] 쉐도우 브레인의 자체 토큰이 있으면 우선 사용 (인증용)
            # brain_service에서 넘어온 token이 있다면 그것을 사용
            resolved_key = get_secret("GEMINI_API_KEY", aegis_url=aegis_url, token=token)
            if resolved_key and not resolved_key.startswith("Imperial:"):
                target_key = resolved_key

        if not target_key:
            logger.error(f"[Brain Core] ❌ GEMINI_API_KEY를 해결할 수 없습니다. (Aegis: {aegis_url})")
            return None

        # 2. 클라이언트 캐시 확인 및 생성
        if target_key not in self._clients:
            if genai:
                logger.info(f"[Brain Core] 🛰️ New environment detected. Initializing client for key: {target_key[:8]}...")
                self._clients[target_key] = genai.Client(api_key=target_key)
            else:
                return None
        
        return self._clients[target_key]


    def _think_fallback(self, prompt: str, system_instruction: str = None, skip_tools: bool = False, tools: list = None) -> Generator[str, None, None]:
        """[폴백 비활성화] 과거에는 1차 실패 시 다른 프로바이더(github/gpt-4o 등)로 자동 전환했으나,
        실제 에러를 가려 디버깅을 방해하므로 제거함. 이제 실패를 그대로 노출한다."""
        logger.warning(
            "[Fallback Disabled] 프로바이더 자동 전환이 비활성화되었습니다. "
            f"(primary={self.primary_provider}/{self.primary_model})"
        )
        yield (
            "❌ [브레인 실패] 1차 모델이 응답하지 못했고, 자동 폴백은 비활성화되어 있습니다.\n"
            f"(provider={self.primary_provider}, model={self.primary_model})\n"
            "프로바이더/모델/네트워크 상태를 직접 확인하세요."
        )

    def _get_tools_schema(self, prompt: str = ""):
        """Returns the tools in OpenAI-compatible format.

        prompt이 주어지면 의도별로 도구를 큐레이션해 목록을 작게 유지한다(헤르메스처럼 깔끔).
        prompt이 비면(레거시 호출) 전량 노출로 안전하게 동작.
        """
        # 🌐 MCP 도구(google-surf 검색, filesystem 등)도 OpenAI 경로에 노출
        mcp_tools = []
        if get_mcp_client:
            try:
                _mc = get_mcp_client()
                if _mc:
                    mcp_tools = _mc.get_openai_tools()
            except Exception as _mcp_err:
                logger.debug(f"[MCP] OpenAI 스키마 변환 실패(무시): {_mcp_err}")

        # 🧹 의도 감지 (prompt 없으면 전량 노출)
        if prompt:
            # 🗂️ 옵시디언 작성/검색 의도면 코딩 모드를 끈다 — 파일/터미널 도구가 목록을
            #    범람시켜 약한 모델이 obsidian_write를 못 고르던 문제 차단.
            wants_coding = self._wants_coding(prompt) and not self._wants_obsidian(prompt)
            flags = {"hugin": self._wants_hugin(prompt),
                     "os": self._wants_os_control(prompt),
                     "delegation": self._wants_delegation(prompt)}
        else:
            wants_coding = True
            flags = {"hugin": True, "os": True, "delegation": True}

        coding = list(CODING_TOOL_SCHEMAS) if wants_coding else []

        # 🔱 [통일] MCP 도구 큐레이션 — 네이티브 중복 제거 + 무거운 그룹 의도 게이트.
        _mcp_flags = {
            "hugin": flags.get("hugin", True),
            "search": (self._wants_web_search(prompt) if prompt else True),
            "coding": wants_coding,
        }
        mcp_tools = [t for t in mcp_tools
                     if self._mcp_name_kept(str(t.get("function", {}).get("name", "")), _mcp_flags)]

        # 🧰 네이티브 도구 스키마(의도별) + 직업 whitelist 상한 (Gemini 경로와 동일 규칙)
        native = self._apply_native_whitelist_openai(self.runtime.curated_native_schemas(flags))
        return coding + mcp_tools + native

    def _get_client(self, aegis_url=None, token=None):
        """🔱 [Imperial] 요청 컨텍스트에 맞는 genai Client를 반환합니다."""
        # 1. 특정 Aegis URL이 주입된 경우, 해당 환경의 키를 수혈받음
        target_key = self.api_key
        if aegis_url:
            from sovereign_memory import get_secret
            resolved_key = get_secret("GEMINI_API_KEY", aegis_url=aegis_url, token=token)
            if resolved_key and not resolved_key.startswith("Imperial:"):
                target_key = resolved_key

        if not target_key:
            logger.error(f"[Brain Core] ❌ GEMINI_API_KEY를 해결할 수 없습니다. (Aegis: {aegis_url})")
            return None

        # 2. 클라이언트 캐시 확인 및 생성
        if target_key not in self._clients:
            if genai:
                logger.info(f"[Brain Core] 🛰️ New environment detected. Initializing client for key: {target_key[:8]}...")
                self._clients[target_key] = genai.Client(api_key=target_key)
            else:
                return None
        
        return self._clients[target_key]

    def _think_copilot(
        self,
        prompt: str,
        model_name: str,
        system_instruction: str = None,
        show_self_healing_banner: bool = False,
        self_healing_provider: str = "github",
        skip_tools: bool = False,
        tools: list = None
    ) -> Generator[str, None, None]:
        """Fallback brain using GitHub Copilot API with Tool Use support."""
        try:
            actual_model = self._resolve_actual_model("github", model_name)
        except ValueError as ve:
            yield f"\n❌ [Neural Core Error] {str(ve)}"
            return

        token, base_url = self._get_copilot_token()
        if not token:
            yield f"[ERROR] Neural Core Failed: GitHub Copilot token unavailable for {actual_model}."
            return
            
        messages = [{"role": "user", "content": prompt}]
        if system_instruction:
            messages.insert(0, {"role": "system", "content": system_instruction})

        tools = self._get_tools_schema(prompt) if not skip_tools else None
        
        try:
            logger.info(f"Imperial Brain Switching to Fallback Core: {actual_model} (GitHub Copilot) with Tool Support")
            
            # OpenAI / Copilot Tool-Calling Loop
            max_iterations = self.dynamic_config.get("max_iterations", 15)
            # 🛡️ [Tool Runtime] 도구 루프 상태 일원화 (3경로 공통)
            ctx = RuntimeCtx(prompt, interactive=True,
                             debug=_is_debug_mode(), display_name="Imperial")
            for _ in range(max_iterations):
                json_data = {
                    "messages": messages,
                    "model": actual_model,
                    "stream": True,
                    "stream_options": {"include_usage": True},  # 🪙 토큰 사용량 집계
                }
                if tools and not ctx.force_no_tools:
                    json_data["tools"] = tools
                    json_data["tool_choice"] = "auto"

                res = requests.post(f"{base_url}/chat/completions", headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json",
                    "Editor-Version": "vscode/1.102.0",
                    "Editor-Plugin-Version": "copilot/1.300.0",
                    "User-Agent": "GithubCopilot/1.300.0",
                    "VSC-Machine-Id": self._get_vsc_machine_id(),
                    "VSC-Session-Id": str(uuid.uuid4()),
                    "Accept": "application/json"
                }, json=json_data, stream=True, timeout=120) 
                
                if not res.ok:
                    error_detail = res.text
                    yield f"\n❌ [Neural Core Error] GitHub Copilot 이 요청을 거절했습니다 ({res.status_code}).\n상세 사유: {error_detail}"
                    return
                
                full_content = ""
                tool_calls = []
                first_yield = True
                _usage_acc = None  # 🪙 토큰 사용량(최종 청크)

                for line in res.iter_lines():
                    if line:
                        l_str = line.decode('utf-8')
                        if l_str.startswith("data: "):
                            data_raw = l_str[6:]
                            if data_raw.strip() == "[DONE]": break
                            try:
                                chunk = json.loads(data_raw)
                                if chunk.get("usage"):
                                    _usage_acc = chunk["usage"]
                                if not chunk.get("choices"):
                                    continue
                                delta = chunk["choices"][0].get("delta", {})

                                # Handle normal content
                                content = delta.get("content", "")
                                if content:
                                    full_content += content
                                    if first_yield:
                                        if show_self_healing_banner:
                                            yield f"[SYSTEM] [Neural Self-Healing: {self_healing_provider.upper()} ({actual_model}) Activated]\n\n"
                                        first_yield = False
                                    if not ctx.suppress_stream:
                                        yield content
                                
                                # Handle tool calls
                                tc_delta = delta.get("tool_calls", [])
                                for tc in tc_delta:
                                    index = tc.get("index", 0)
                                    while len(tool_calls) <= index:
                                        tool_calls.append({"id": "", "type": "function", "function": {"name": "", "arguments": ""}})
                                    
                                    if "id" in tc: tool_calls[index]["id"] += tc["id"]
                                    if "function" in tc:
                                        f = tc["function"]
                                        if "name" in f: tool_calls[index]["function"]["name"] += f["name"]
                                        if "arguments" in f: tool_calls[index]["function"]["arguments"] += f["arguments"]
                            except: pass

                self._emit_token_usage("github", actual_model, _usage_acc)
                if not tool_calls:
                    # 🛡️ [Tool Runtime] 도구 없는 응답 — 구제 처리 일원화
                    _action = yield from self.runtime.handle_no_tool_calls(full_content, messages, ctx)
                    if _action == "continue":
                        continue
                    break

                # 🛡️ [Tool Runtime] 도구 호출 일괄 실행 (검증·보정·ask 가드·서킷브레이커 공통)
                yield from self.runtime.run_tool_calls(tool_calls, full_content, messages, ctx)

        except Exception as e:
            logger.error(f"Copilot Core Error: {e}")
            yield f"\n❌ [Copilot] Neural Core Malfunction: {str(e)}"

    def _assign_atlas_port(self) -> Optional[int]:
        """Queries the Atlas Broker for a dynamic port assignment."""
        try:
            assign_url = f"{self.atlas_broker_url}/assign?agent=shadow_brain"
            logger.info(f"[🔱 Atlas] Requesting port assignment from {assign_url}")
            resp = requests.get(assign_url, timeout=5)
            if resp.status_code == 200:
                data = resp.json()
                port = data.get("port")
                if port:
                    logger.info(f"[🔱 Atlas] Received intelligence port: {port}")
                    return port
            logger.error(f"[🔱 Atlas] Broker assignment failed (HTTP {resp.status_code}): {resp.text}")
        except Exception as e:
            logger.error(f"[🔱 Atlas] Broker connection error: {e}")
        return None

    def _think_atlas(
        self,
        prompt: str,
        model_name: str,
        system_instruction: str = None,
        skip_tools: bool = False,
        tools: list = None
    ) -> Generator[str, None, None]:
        """Local Brain using Atlas Orchestrator (llama-server) with Tool Support."""
        port = self._assign_atlas_port()
        if not port:
            yield f"[ERROR] 🔱 Atlas Broker Neural Assignment Failed. Check if RunAtlas.bat is active."
            return

        # 🔱 아틀라스 사령부(Broker)를 관문으로 사용하여 전 모드 호환성 확보
        base_url = f"{self.atlas_broker_url}/v1"
        messages = [{"role": "user", "content": prompt}]
        if system_instruction:
            messages.insert(0, {"role": "system", "content": system_instruction})

        # Atlas currently uses standard OpenAI-compatible tool schema if supported by the model
        atlas_tools = self._get_tools_schema(prompt) if not skip_tools else None
        
        try:
            logger.info(f"Imperial Brain engaging LOCAL ATLAS Shield ({model_name}) on Port {port}")
            
            # Streaming Loop (similar to Copilot/OpenAI but local)
            max_iterations = self.dynamic_config.get("max_iterations", 15)
            # 🛡️ [Tool Runtime] 도구 루프 상태 일원화 (3경로 공통)
            ctx = RuntimeCtx(prompt, interactive=True,
                             debug=_is_debug_mode(), display_name="Atlas")
            for _ in range(max_iterations):
                json_data = {
                    "messages": messages,
                    "model": model_name,
                    "stream": True,
                    "temperature": 0.7,
                    "stream_options": {"include_usage": True},  # 🪙 토큰 사용량 집계
                }
                if atlas_tools and not ctx.force_no_tools:
                    json_data["tools"] = atlas_tools
                    json_data["tool_choice"] = "auto"

                res = requests.post(f"{base_url}/chat/completions", json=json_data, stream=True, timeout=120)
                res.raise_for_status()

                full_content = ""
                tool_calls = []
                _usage_acc = None  # 🪙 토큰 사용량(최종 청크)

                for line in res.iter_lines():
                    if line:
                        l_str = line.decode('utf-8')
                        if l_str.startswith("data: "):
                            data_raw = l_str[6:]
                            if data_raw.strip() == "[DONE]": break
                            try:
                                chunk = json.loads(data_raw)
                                if chunk.get("usage"):
                                    _usage_acc = chunk["usage"]
                                if not chunk.get("choices"):
                                    continue
                                delta = chunk["choices"][0].get("delta", {})

                                # Handle normal content
                                content = delta.get("content", "")
                                if content:
                                    full_content += content
                                    if not ctx.suppress_stream:
                                        yield content
                                
                                # Handle tool calls
                                tc_delta = delta.get("tool_calls", [])
                                for tc in tc_delta:
                                    index = tc.get("index", 0)
                                    while len(tool_calls) <= index:
                                        tool_calls.append({"id": "", "type": "function", "function": {"name": "", "arguments": ""}})
                                    
                                    if "id" in tc: tool_calls[index]["id"] += tc["id"]
                                    if "function" in tc:
                                        f = tc["function"]
                                        if "name" in f: tool_calls[index]["function"]["name"] += f["name"]
                                        if "arguments" in f: tool_calls[index]["function"]["arguments"] += f["arguments"]
                            except: pass

                self._emit_token_usage("atlas", model_name, _usage_acc)
                if not tool_calls:
                    # 🛡️ [Tool Runtime] 도구 없는 응답 — 구제 처리 일원화
                    _action = yield from self.runtime.handle_no_tool_calls(full_content, messages, ctx)
                    if _action == "continue":
                        continue
                    break

                # 🛡️ [Tool Runtime] 도구 호출 일괄 실행 (검증·보정·ask 가드·서킷브레이커 공통)
                yield from self.runtime.run_tool_calls(tool_calls, full_content, messages, ctx)

        except Exception as e:
            logger.error(f"Atlas Local Session Error: {e}")
            yield f"\n❌ Local Atlas Malfunction: {str(e)}"

    def _wants_coding(self, prompt: str) -> bool:
        """
        🛠️ [Imperial Coding Gate] 코딩/파일 작업 의도 감지.
        True면 에이전틱 코딩 수칙(CODING_AGENT_GUIDE)을 시스템 지시에 주입하고,
        검색 그라운딩 모드보다 코딩 모드를 우선한다.
        """
        if not prompt:
            return False
        p = prompt.lower()
        coding_triggers = [
            "코드", "코딩", "버그", "에러", "오류", "수정해", "고쳐", "구현", "리팩토링", "함수",
            "파일", "스크립트", "디버그", "분석해", "검토", "테스트", "빌드", "컴파일",
            "code", "bug", "error", "fix", "implement", "refactor", "debug", "build",
            ".py", ".dart", ".js", ".ts", ".json", ".yaml", ".md", "api", "backend", "frontend",
        ]
        return any(t in p for t in coding_triggers)

    def _wants_obsidian(self, prompt: str) -> bool:
        """🗂️ [Imperial Obsidian Gate] 옵시디언 노트/위키 작성·검색 의도 감지.
        True면 코딩 모드(파일/터미널 도구 범람)를 끄고 obsidian_search/write 도구를 부각시킨다.
        ('위키 만들어줘'가 코딩 의도로 빨려가 obsidian_write 대신 말로만 때우던 문제 차단)
        로컬 볼트가 비활성이면 굳이 게이트할 필요 없으므로 False.
        """
        if not prompt:
            return False
        try:
            import obsidian_vault as _ov
            if not _ov.is_enabled():
                return False
        except Exception:
            return False
        p = prompt.lower()
        return ("옵시디언" in p) or ("obsidian" in p) or ("볼트" in p and "vault" in p)

    def _wants_obsidian_write(self, prompt: str) -> bool:
        """🗂️ 옵시디언 의도 중 '쓰기/작성/저장' 의도인지 구분한다.
        검색('찾아/검색/알려')은 쓰기가 아니라 그라운딩(컨텍스트 주입)으로 답해야 하므로,
        쓰기 의도일 때만 obsidian_write 도구와 작성-재촉(nudge)을 켠다.
        """
        if not prompt:
            return False
        p = prompt.lower()
        _write_kw = (
            "만들", "작성", "저장", "정리", "기록", "써줘", "써 줘", "적어", "추가",
            "노트로", "메모", "업데이트", "수정", "갱신", "write", "create", "save", "append",
        )
        return any(k in p for k in _write_kw)

    def _obsidian_query(self, prompt: str) -> str:
        """옵시디언 검색용 핵심어 추출. 앞쪽 인사·출처어와 뒤쪽 검색 명령어를 떼어낸다.
        붙여 쓴 경우도 처리: '하이 클라우드 옵시디언에서 ttt찾아주세요' → 'ttt'.
        (명령어가 검색어에 붙어 시맨틱/키워드가 엉뚱한 노트로 쏠리는 것을 방지.)
        앵커(^/$) 방식이라 문장 중간의 같은 글자(예: '검색엔진')는 건드리지 않는다.
        """
        if not prompt:
            return prompt
        q = prompt.strip()
        _lead = (
            r"^(하이|안녕하세요|안녕|헤이|hi|hello|"
            r"클라우드|cloud|로컬|local|"
            r"옵시디언에서|옵시디언에|옵시디언|obsidian|볼트에서|볼트|vault|"
            r"노트에서|노트|메모에서|메모)\s*"
        )
        _tail = (
            r"\s*(에서|에)?\s*(찾아주세요|찾아줘|찾아 줘|찾아|검색해주세요|검색해줘|검색해|검색|"
            r"알려주세요|알려줘|알려|보여주세요|보여줘|말해줘|읽어줘|해주세요|해줘|주세요|줘|좀)$"
        )
        # 🇰🇷 한국어는 조사가 띄어쓰기 없이 붙는다('abcd에대해'). 명령어 제거 후 남는
        #   연결어(에 대해/관련/대해/관해 등)를 떼어 핵심어만 남긴다. (단일 조사는
        #   검색엔진의 _strip_particle이 키워드별로 따로 처리하므로 여기서 안 건드림.)
        _conn = (
            r"\s*(에\s*대해서|에\s*대해|에\s*대한|에\s*관해서|에\s*관해|에\s*관한|"
            r"에관해|에대해|관련해서|관련된|관련해|관련|대해서|대해|관해서|관해)$"
        )
        prev = None
        while q and q != prev:
            prev = q
            q = re.sub(_lead, "", q)
            q = re.sub(_tail, "", q)
            q = re.sub(_conn, "", q)
            q = q.strip()
        return q or prompt.strip()

    # 🔧 [Tool Policy] 내장 폴백 기본값 — DB 정책(tool_policy 섹션)이 없거나 조회 실패 시 사용.
    #   운영 조정은 코드가 아니라 이지스 '자동 기억·검색 정책' JSON의 tool_policy에서 한다.
    _TOOL_POLICY_FALLBACK = {
        "intents": {
            "media": {
                "nouns": ["이미지", "사진", "그림", "영상", "동영상", "gif", "움짤", "짤",
                          "배경화면", "월페이퍼", "image", "photo", "picture", "video", "wallpaper",
                          "3d", "3d모델", "3d 모델", "glb", "vrm", "에셋", "3d 에셋",
                          "유튜브", "youtube"],
                "actions": ["찾아", "검색", "구해", "다운", "받아", "보여",
                            "search", "find", "download", "get"],
                "tools": ["search_media", "search_3d_model", "search_youtube"],
                "skip_grounding": True,
            },
            "delegate": {
                "nouns": ["코덱스", "미니맥스", "헤르메스", "자비스", "수호자", "동료",
                          "codex", "minimax", "mavis", "hermes", "claude", "lm studio", "lmstudio"],
                "actions": ["시켜", "위임", "맡겨", "부탁", "물어", "분담", "협업", "지시",
                            "delegate", "ask", "assign"],
                "tools": ["delegate_to_guardian"],
                "skip_grounding": True,
            },
        },
        "tool_defaults": {
            "search_3d_model": {"limit": 8, "max": 20},
            "search_media": {"limit": 8, "max": 20},
            "search_youtube": {"limit": 6, "max": 20},
        },
        "search_engines": {"gemini_grounding": True, "serpapi": True, "playwright": True},
    }

    def _tool_policy(self) -> dict:
        """DB 검색 정책의 tool_policy 섹션 (TTL 캐시). 실패/누락 시 내장 폴백."""
        try:
            aegis_url, _ = self._resolve_aegis()
            if not aegis_url:
                aegis_url = os.environ.get("AEGIS_URL", "").rstrip("/") or None
            from utils.emotion import get_dynamic_retrieval_policy
            tp = (get_dynamic_retrieval_policy(aegis_url) or {}).get("tool_policy")
            if isinstance(tp, dict) and tp:
                return tp
        except Exception as e:  # noqa: BLE001
            logger.debug(f"[ToolPolicy] 조회 실패 → 내장 폴백: {e}")
        return self._TOOL_POLICY_FALLBACK

    def _intent_cfg(self, name: str) -> dict:
        """의도별 설정(nouns/actions/tools/skip_grounding). 정책 우선, 폴백 병합."""
        fb = (self._TOOL_POLICY_FALLBACK.get("intents") or {}).get(name) or {}
        cfg = ((self._tool_policy().get("intents") or {}).get(name)) or {}
        merged = dict(fb)
        merged.update({k: v for k, v in cfg.items() if v is not None})
        return merged

    def _wants_delegate(self, prompt: str) -> bool:
        """🤝 [Imperial Orchestration Gate] 동료 수호자 위임 의도 감지.
        키워드는 DB 정책 tool_policy.intents.delegate 에서 온다 (내장 폴백 병합)."""
        if not prompt:
            return False
        p = prompt.lower()
        cfg = self._intent_cfg("delegate")
        nouns = cfg.get("nouns") or []
        actions = cfg.get("actions") or []
        return any(n in p for n in nouns) and any(a in p for a in actions)

    # 🤝 [Imperial Orchestration] 수호자 위임 라우트 — 태미니 CLI/헤임달과 동일 중계 재사용
    _DELEGATE_ROUTES = {
        "codex":    ("/api/shadow/codex/send", "context_id"),
        "minimax":  ("/api/shadow/minimax-code/send", "sessionKey"),
        "hermes":   ("/api/shadow/hermes/send", "sessionKey"),
        "claude":   ("/api/shadow/claude-code/send", "context_id"),
        "lmstudio": ("/api/shadow/lmstudio/send", "sessionKey"),
    }
    _DELEGATE_ALIASES = {
        "코덱스": "codex", "미니맥스": "minimax", "mavis": "minimax",
        "헤르메스": "hermes", "자비스": "claude", "claude-code": "claude",
        "jarvis": "claude", "lm": "lmstudio", "lms": "lmstudio", "lm studio": "lmstudio",
    }

    def delegate_to_guardian(self, guardian: str = "", task: str = "", context: str = "") -> str:
        """🤝 동료 수호자(코덱스·미니맥스·헤르메스·자비스·LM Studio)에게 작업을 위임하고
        그 수행 결과를 돌려받는다. 기획→코딩→테스트→검증 오케스트레이션의 기본 단위."""
        g = self._DELEGATE_ALIASES.get((guardian or "").strip().lower(),
                                       (guardian or "").strip().lower())
        if g not in self._DELEGATE_ROUTES:
            return (f"오류: 알 수 없는 수호자 '{guardian}'. "
                    "사용 가능: codex(코딩), minimax(코딩/테스트), hermes, claude(기획/검증), lmstudio")
        task = (task or "").strip()
        if not task:
            return "오류: 위임할 작업(task) 설명이 비었습니다."
        path, key_field = self._DELEGATE_ROUTES[g]
        message = f"[쉐도우 브레인 위임 작업]\n{task}"
        if (context or "").strip():
            message += f"\n\n[참고 컨텍스트]\n{context.strip()[:4000]}"
        payload = {"message": message, key_field: f"orchestrate-{g}"}
        if g in ("codex", "minimax"):
            try:
                from brain.coding_tools import _ROOT_OVERRIDE
                cwd = getattr(_ROOT_OVERRIDE, "value", None)
            except Exception:
                cwd = None
            if cwd:
                payload["cwd"] = cwd
        port = (os.environ.get("SHADOW_PORT") or "18700").strip()
        try:
            import requests as _rq
            logger.info(f"🤝 [Delegate] {g} ← {task[:80]}")
            r = _rq.post(f"http://127.0.0.1:{port}{path}", json=payload, timeout=600)
            data = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
        except Exception as e:
            return f"위임 실패({g}): {e}"
        if not data or data.get("status") == "ERROR":
            return f"{g} 위임 오류: {(data or {}).get('message') or f'HTTP {r.status_code}'}"
        content = (data.get("content") or "").strip() or "(빈 응답)"
        return f"[{g} 수행 결과]\n{content[:12000]}"

    def _wants_media(self, prompt: str) -> bool:
        """🖼️ [Imperial Media Gate] 로열티-프리 미디어 및 3D 에셋 검색 의도 감지.
        True면 검색 그라운딩 대신 도구(search_media, search_3d_model)를 쓰도록 한다.
        키워드는 DB 정책 tool_policy.intents.media 에서 온다 (코드 수정 없이 조정 가능).
        """
        if not prompt:
            return False
        p = prompt.lower()
        cfg = self._intent_cfg("media")
        nouns = cfg.get("nouns") or []
        actions = cfg.get("actions") or []
        return any(n in p for n in nouns) and any(a in p for a in actions)

    def _wants_realtime_search(self, prompt: str) -> bool:
        """🌐 최신/실시간 정보 질문 감지 (Gemini 경로의 SEARCH_KEYWORDS와 동일 취지)."""
        if not prompt:
            return False
        p = prompt.lower()
        keywords = [
            "search", "find", "검색", "찾아", "구글", "google", "웹검색", "웹 검색",
            "뉴스", "news", "날씨", "weather", "기온", "온도", "미세먼지", "속보", "이슈", "사건", "사고",
            "최신", "실시간", "오늘", "지금", "현재", "최근", "요즘", "어제",
            "주가", "주식", "환율", "코인", "비트코인", "가격", "시세", "얼마",
            "대통령", "수상", "총리", "장관", "시장", "국회의원", "의원", "누구야", "누구지", "누구인", "누가",
        ]
        return any(k in p for k in keywords)

    def _wants_hugin(self, prompt: str) -> bool:
        """
        🔱 [Imperial Hugin Gate] 마왕님이 명시적으로 Hugin(브라우저 제어)을 요청했는지 감지합니다.
        True일 때만 hugin_navigate/screenshot/click/type 도구가 모델에 노출됩니다.
        기본은 제외 — 모델이 습관적으로 브라우저 스크린샷을 호출하는 것을 막기 위함입니다.
        """
        if not prompt:
            return False
        p = prompt.lower()
        hugin_triggers = [
            # 도구 이름 직접 언급
            "hugin", "후긴", "휴긴",
            # 브라우저 제어 의도
            "브라우저", "browser", "크롬", "chrome", "웹페이지", "웹 페이지", "사이트 열어",
            "페이지 열어", "url 열어", "주소 열어", "navigate",
        ]
        return any(t in p for t in hugin_triggers)

    def _wants_web_search(self, prompt: str) -> bool:
        """
        🔱 [Imperial Search Gate] 웹 검색 의도 감지.
        Ollama 복합 의도 처리 시 obsidian_write + web_search 조합 결정에 사용.
        """
        if not prompt:
            return False
        p = prompt.lower()
        search_triggers = [
            "검색", "찾아", "알아봐", "찾아봐", "조사해", "검색해", "인터넷에서",
            "웹에서", "구글", "google", "search", "찾아줘", "알려줘",
            "최신", "뉴스", "정보 찾", "정보를 찾",
        ]
        return any(t in p for t in search_triggers)

    def _wants_delegation(self, prompt: str) -> bool:
        """
        🔱 [Imperial Delegation Gate] 마왕님이 명시적으로 팀 위임/히드라 디스패치를
        요청했는지 감지합니다. True일 때만 delegate_team_task/dispatch_to_hydra 도구가
        모델에 노출됩니다. 기본은 제외 — 파일 만들기 같은 일반 작업에 모델이 습관적으로
        팀 위임/히드라를 호출해 실패하던 문제를 막기 위함입니다.
        """
        if not prompt:
            return False
        p = prompt.lower()
        delegation_triggers = [
            # 팀 테스크 / 위임 의도
            "팀 테스크", "팀테스크", "테스크 만들", "테스크 생성", "task 만들", "task 생성",
            "위임", "delegate", "맡겨", "맡아", "배정", "할당해",
            # 수호자 지명 / 히드라 / 병렬 기획
            "히드라", "hydra", "아쿠벤스", "acubens", "에르메스", "hermes",
            "수호자에게", "수호자한테", "팀에게", "팀한테", "병렬", "기획 보고서",
            "dispatch",
        ]
        return any(t in p for t in delegation_triggers)

    def _wants_os_control(self, prompt: str) -> bool:
        """
        🔱 [Imperial OS Control Gate] 마왕님이 명시적으로 OS 앱 실행/키입력/타이핑을
        요청했는지 감지한다. True일 때만 sovereign_run_app/type_text/press_key 도구가 노출된다.
        기본 제외 — '파일 만들어줘' 같은 작업에 모델이 notepad/cmd를 실행하려다 실패하며
        화면을 어지럽히던 문제를 막기 위함이다. (파일은 MCP 파일도구로 충분)
        """
        if not prompt:
            return False
        p = prompt.lower()
        os_triggers = [
            # 앱 실행 의도
            "실행해", "실행시켜", "켜줘", "열어줘", "구동", "launch", "run app", "open app",
            "메모장", "notepad", "계산기", "calc", "cmd", "터미널", "파워셸", "powershell",
            "프로그램 실행", "앱 실행", "응용프로그램",
            # 타이핑/키입력 의도 ('쳐줘'는 '고쳐줘'에 오매칭되어 제외, '타이핑 쳐줘'로 한정)
            "타이핑", "타이핑 쳐", "키 눌러", "키를 눌러", "단축키", "엔터 눌러", "type ", "press key",
        ]
        return any(t in p for t in os_triggers)

    @staticmethod
    def _tool_allowed(tool_name: str, allowed_list: list) -> bool:
        """
        🎖️ [Guardian Job M1] 도구 화이트리스트 매칭.
        allowed_list의 각 항목과 tool_name을 비교한다. 와일드카드(*) 지원.
        예) 'mcp_filesystem_*' 는 'mcp_filesystem_write_file' 에 매칭.
        """
        if not allowed_list:
            return False
        import fnmatch
        for pattern in allowed_list:
            if not pattern:
                continue
            if pattern == tool_name or fnmatch.fnmatch(tool_name, pattern):
                return True
        return False

    # 🔱 [도구 노출 통일] 3개 브레인 경로(OpenAI/Gemini/ollama)가 같은 규칙을 쓰도록 공통 헬퍼.
    _WHITELIST_BASE_TOOLS = {
        "jarvis_memory", "recall_memory", "count_history",
        "search_knowledge", "save_knowledge", "check_port", "find_process", "ask_user",
    }

    def _mcp_name_kept(self, name: str, flags: dict) -> bool:
        """MCP 도구를 이번 요청에 실을지 판정 (네이티브 중복 제거 + 무거운 그룹 의도 게이트).
        name은 'mcp_<server>_<tool>' 형식. 네이티브/코딩 도구는 여기서 거르지 않는다(True).
        flags: {hugin(브라우징), search(웹검색), coding(코딩)}.
        """
        n = name or ""
        if not n.startswith("mcp_"):
            return True
        # 1) 네이티브 옵시디언과 중복되는 MCP 제거 — 네이티브가 더 빠르고(같은 프로세스) 토큰 절약.
        if n.endswith("_search_obsidian") or n.endswith("_read_obsidian") or n.endswith("_write_obsidian"):
            return False
        # 2) 무거운 그룹은 의도가 맞을 때만 노출 (playwright 23개가 평소 토큰을 먹던 주범).
        if n.startswith("mcp_playwright"):
            return bool(flags.get("hugin"))
        if n.startswith("mcp_google-surf") or n.startswith("mcp_google_surf"):
            return bool(flags.get("search"))
        if n.startswith("mcp_filesystem"):
            return bool(flags.get("coding"))
        # imperial(taeminix) 등 작고 고유한 도구는 유지.
        return True

    def _apply_native_whitelist_openai(self, native_schemas: list) -> list:
        """직업 whitelist(tool_mode=='whitelist')를 OpenAI 스키마 네이티브 목록에 적용.
        Gemini 경로(unified_function_declarations)와 동일 규칙 — 경로 간 일관성 확보.
        """
        cfg = getattr(self, "dynamic_config", None) or {}
        if (cfg.get("tool_mode") or "all").lower() != "whitelist":
            return native_schemas
        allowed = cfg.get("allowed_tools") or []
        if not allowed:
            return native_schemas
        out = []
        for t in native_schemas:
            nm = (t.get("function", {}) or {}).get("name", "")
            if nm in self._WHITELIST_BASE_TOOLS or self._tool_allowed(nm, allowed):
                out.append(t)
        return out

    def _load_equipped_skills(self, skill_names: list, max_chars_each: int = 1500) -> str:
        """
        🎖️ [Guardian Job M2] 장착된 스킬 문서(.agent/skills/<name>/SKILL.md)를 읽어
        시스템 프롬프트에 주입할 '전문 지식' 텍스트로 조립한다.
        - 토큰 절약: 스킬당 max_chars_each 까지만 (앞부분 = 핵심 지침).
        - 캐싱: 파일 내용을 self._skill_cache 에 보관해 매 요청마다 디스크 안 읽음.
        """
        if not skill_names:
            return ""
        if not hasattr(self, "_skill_cache"):
            self._skill_cache = {}

        try:
            proj_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        except Exception:
            return ""
        skills_dir = os.path.join(proj_root, ".agent", "skills")

        blocks = []
        for name in skill_names:
            if not name or not isinstance(name, str):
                continue
            if name in self._skill_cache:
                blocks.append(self._skill_cache[name])
                continue
            # 경로 탈출 방지(이름에 슬래시/.. 차단)
            safe = os.path.basename(name.strip())
            path = os.path.join(skills_dir, safe, "SKILL.md")
            if not os.path.exists(path):
                logger.warning(f"[Guardian Job M2] 스킬 문서 없음: {safe}")
                continue
            try:
                with open(path, "r", encoding="utf-8") as f:
                    text = f.read().strip()
                if len(text) > max_chars_each:
                    text = text[:max_chars_each] + "\n...(이하 생략)"
                block = f"### 📘 장착 스킬: {safe}\n{text}"
                self._skill_cache[name] = block
                blocks.append(block)
                logger.info(f"[Guardian Job M2] 스킬 주입: {safe} ({len(text)}자)")
            except Exception as e:
                logger.warning(f"[Guardian Job M2] 스킬 읽기 실패 ({safe}): {e}")

        if not blocks:
            return ""
        return (
            "\n\n## 🎖️ 전문 지식 (장착된 스킬)\n"
            "아래는 당신의 직업에 부여된 전문 지식이다. 작업 시 이 지침을 우선 적용하라.\n\n"
            + "\n\n".join(blocks)
        )

    def _is_task_request(self, prompt: str, images: list = None) -> bool:
        """
        🔱 [Imperial Task Detector] 작업형 요청 여부를 감지합니다.
        파일/코드/UI/버그 관련 키워드가 있으면 True를 반환합니다.
        True인 경우 Gemini에게 도구 호출을 강제(mode=ANY)합니다.

        [🔱 Imperial Fix v2.7] 이미지가 첨부됐다는 사실만으로 도구를 강제하지 않는다.
        '이 이미지 뭐가 보여?' 같은 순수 분석/감상 질문은 도구 없이 멀티모달로
        그냥 보고 답해야 하기 때문이다. 이미지가 있어도 아래 작업 키워드가
        함께 있을 때만(예: '이미지 보고 CSS 고쳐줘') 도구를 강제한다.
        """
        # [🔱 Imperial Fix v2.7] '첨부', '스크린샷', '이미지를', '사진을' 등은
        # 이미지가 첨부됐다는 사실을 가리킬 뿐 작업 의도가 아니므로 작업 키워드에서 제외.
        # ('첨부한 이미지가 뭔지 보이나요?' 같은 순수 감상 질문이 도구 강제로 빠지지 않게 함)
        task_keywords = [
            # 한국어 작업 키워드 (실제 변경/조작 의도)
            "수정", "고쳐", "고치", "변경", "바꿔", "바꾸", "추가", "삭제", "제거", "구현",
            "버그", "오류", "에러", "error", "고장", "안 됨", "안됨", "문제", "이상",
            "파일", "코드", "css", "ui", "레이아웃",
            "검토", "분석", "찾아줘", "디버그",
            "여전히", "아직도",
            # 영어 작업 키워드
            "fix", "modify", "change", "update", "implement", "debug",
            "read file", "edit file", "write file", "inspect",
        ]
        prompt_lower = prompt.lower()
        return any(kw in prompt_lower for kw in task_keywords)

    def _should_use_google_grounding(self, prompt: str, aegis_url: str = None) -> bool:
        text = (prompt or "").strip().lower()
        if not text:
            return False
        from utils.emotion import get_dynamic_retrieval_policy
        policy = get_dynamic_retrieval_policy(aegis_url)
        if policy.get("grounding_enabled") is not True:
            return False
        for pattern in policy.get("grounding_exclude_patterns") or []:
            try:
                if re.search(str(pattern), text, re.IGNORECASE):
                    return False
            except re.error:
                logger.warning("[RetrievalPolicy] 잘못된 제외 정규식: %s", pattern)
        keywords = [str(k).lower() for k in (policy.get("grounding_keywords") or []) if str(k).strip()]
        return any(keyword in text for keyword in keywords)

    def _observe(self, event, reason="", severity="warning", meta=None, aegis_url=None, token=None):
        """🔭 관측 이벤트를 이지스 이벤트 저장소로 best-effort 전송 (단일 관측 지점 통합).
        절대 예외를 던지지 않는다 — 관측이 본 기능을 깨선 안 된다."""
        try:
            logger.warning("[OBSERVE:%s] %s | %s | %s", severity, event, str(reason)[:200], meta or {})
        except Exception:
            pass
        try:
            api = self._aegis_api_url(
                aegis_url or getattr(self._request_ctx, "aegis_url", None)
                or os.environ.get("AEGIS_URL"), "/system/events")
            if not api:
                return
            tok = token or os.environ.get("GUARDIAN_SERVICE_TOKEN") or os.environ.get("IMPERIAL_API_TOKEN")
            headers = {"Content-Type": "application/json"}
            if tok:
                headers["Authorization"] = f"Bearer {tok}"
                headers["X-Imperial-Token"] = tok
            requests.post(api, json={
                "event": event, "reason": str(reason)[:2000], "severity": severity,
                "source": "shadow_brain", "meta": meta or {},
            }, headers=headers, timeout=4)
        except Exception as e:  # noqa: BLE001
            logger.debug("[OBSERVE] 이지스 전송 실패(무시): %s", e)

    def _aegis_api_url(self, aegis_url: str, path: str) -> str:
        base = (aegis_url or "").strip().rstrip("/")
        if not base:
            return ""
        if not path.startswith("/"):
            path = f"/{path}"
        if base.endswith("/api"):
            return f"{base}{path}"
        return f"{base}/api{path}"

    def _search_grounding_context(self, query: str, aegis_url: str = None, token: str = None,
                                  context_id: str = None) -> str:
        api_url = self._aegis_api_url(aegis_url, "/chat/search-grounding")
        if not api_url:
            raise ValueError("X-Aegis-URL header is required for Ollama search grounding.")

        model = getattr(self, "search_grounding_model", None) or "gemini-2.5-flash"
        headers = {"Content-Type": "application/json"}
        if token:
            headers["Authorization"] = f"Bearer {token}"
            headers["X-Imperial-Token"] = token
            headers["X-Guardian-Token"] = token

        response = requests.post(
            api_url,
            json={
                "query": query,
                "model": model,
                "context_id": context_id
                or getattr(self._request_ctx, "session_id", None)
                or "GLOBAL",
            },
            headers=headers,
            # 🔱 Django 그라운딩 뷰(Gemini 호출 60s) 보다 길게 둬야 502/잘림 없이 결과를 받는다.
            timeout=70,
        )
        if response.status_code != 200:
            try:
                detail = response.json().get("message") or response.text
            except Exception:
                detail = response.text
            raise RuntimeError(f"Search grounding failed ({response.status_code}): {detail}")

        data = response.json()
        if data.get("status") not in ("SUCCESS", "OK", None):
            raise RuntimeError(data.get("message") or "Search grounding returned an error.")

        text = str(data.get("text") or "").strip()
        if not text:
            raise RuntimeError("Search grounding returned empty text.")

        # 🔎 실제 사용된 검색 엔진 (Aegis가 SerpAPI 폴백 시 provider='serpapi')
        _provider = str(data.get("provider") or "")
        self._last_grounding_engine = (
            "SerpAPI (구글 검색 결과)" if _provider == "serpapi"
            else f"Google Gemini 그라운딩 ({data.get('model') or model})"
        )

        sources = data.get("sources") or []
        lines = [
            f"[Web Search Grounding Context — {self._last_grounding_engine}]",
            "Use this live web search grounding context before relying on model memory.",
            "",
            text,
        ]
        if isinstance(sources, list) and sources:
            lines.extend(["", "[Sources]"])
            for source in sources[:6]:
                if isinstance(source, dict):
                    title = str(source.get("title") or "").strip()
                    uri = str(source.get("uri") or "").strip()
                    if title or uri:
                        lines.append(f"- {title} {uri}".strip())
        return "\n".join(lines)

    def _build_ollama_prompt(self, prompt: str, system_instruction: str = None, history: list = None, search_context: str = None) -> str:
        blocks = []
        if system_instruction:
            blocks.append(f"[System]\n{system_instruction}")
        if history and isinstance(history, list):
            history_lines = []
            for msg in history[-10:]:
                role = msg.get("role", "user")
                content = msg.get("content", "")
                if content:
                    history_lines.append(f"{role}: {content}")
            if history_lines:
                blocks.append("[Recent conversation]\n" + "\n".join(history_lines))
        if search_context:
            blocks.append(
                f"{search_context}\n\n"
                "[Important rules]\n"
                "- The Google Search Grounding context above has priority over model memory.\n"
                "- For current people, offices, schedules, prices, laws, and dates, answer from the newest search context.\n"
                "- If search context conflicts with model memory, follow the search context.\n"
                "- Answer in Korean unless the user asks otherwise."
            )
        blocks.append(f"[User question]\n{prompt}")
        return "\n\n".join(blocks)

    def _think_ollama(
        self,
        prompt: str,
        model_name: str,
        history: list = None,
        system_instruction: str = None,
        images: list = None,
        token: str = None,
        aegis_url: str = None,
        skip_tools: bool = False,
        tools: list = None,
        skip_grounding: bool = False,
        stream_tokens: bool = False,
    ) -> Generator[str, None, None]:
        """Local Ollama provider with optional Aegis Google Search Grounding and Tool-Calling support."""
        _op_t0 = time.time()
        _op_lap = [time.time()]

        def _oplap(name):
            now = time.time()
            if now - _op_lap[0] > 0.15:
                logger.info(f"⏱️ [ollama-prep] {name}: {(now - _op_lap[0])*1000:.0f}ms")
            _op_lap[0] = now
        try:
            import dotenv
            # 쉐도우 브레인 루트 및 태민게임즈 프로젝트 루트의 .env를 정밀하게 재부착
            _proj_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
            _env_path = os.path.join(_proj_root, ".env")
            if os.path.exists(_env_path):
                dotenv.load_dotenv(_env_path, override=False)
        except Exception as dotenv_err:
            logger.debug(f"[Ollama Core] Dotenv dynamic reload failed: {dotenv_err}")

        self._load_shadow_settings()

        # 🖼️ [Multimodal] 이미지가 첨부되면 Ollama 네이티브 형식(message.images = [base64])으로
        #   user 메시지에 실어 보낸다. gemma3/4(12b 등) 비전 모델이면 이미지를 보고 분석한다.
        #   (미지원 모델이면 Ollama가 무시/에러 — 더 이상 'text-only'로 하드 차단하지 않는다.)
        _ollama_images = []
        if images:
            for _img in images:
                _data = (_img.get("data") if isinstance(_img, dict) else _img) or ""
                if isinstance(_data, str) and "," in _data and _data.strip().startswith("data:"):
                    _data = _data.split(",", 1)[1]
                if _data:
                    _ollama_images.append(_data)
            if _ollama_images:
                logger.info(f"🖼️ [Ollama] 멀티모달 입력 — 이미지 {len(_ollama_images)}장을 모델 '{model_name}'에 전달.")

        search_context = None
        # 🗂️ [옵시디언 그라운딩] "옵시디언에서 ~ 찾아줘"(검색 의도)는 약한 로컬 모델의 불안정한
        #   도구호출/작성-재촉 루프(무한 '생각중') 대신, 서버가 직접 옵시디언을 검색해 결과를
        #   컨텍스트로 주입하고 모델은 한 번에 답한다(구글 그라운딩과 동일 패턴). 쓰기 의도는 제외.
        _oplap("dotenv-settings")
        _obs_search_intent = self._wants_obsidian(prompt) and not self._wants_obsidian_write(prompt)
        if _obs_search_intent:
            try:
                # ⚠️ '🔎 옵시디언에서 검색합니다...' 안내는 yield하지 않는다. yield하면 그게
                #    스트리밍 응답 본문에 섞여 버블에 남고(재접속해야 사라짐) 저장/벡터화까지 탄다.
                #    '생각중' 버블이 이미 진행 표시를 하므로 별도 안내 없이 조용히 검색한다.
                _pl = prompt.lower()
                _obs_src = "cloud" if ("클라우드" in _pl or "cloud" in _pl) else None
                _obs_ctx = self.obsidian_search(self._obsidian_query(prompt), limit=5, source=_obs_src)
                if _obs_ctx and "없음" not in _obs_ctx[:60] and "비활성" not in _obs_ctx[:60]:
                    search_context = _obs_ctx
            except Exception as _obs_err:
                logger.warning(f"[Obsidian Grounding] 실패: {_obs_err}")

        _wants_med = self._wants_media(prompt)
        _med_skip_grounding = bool(self._intent_cfg("media").get("skip_grounding", True))
        if (
            not skip_grounding
            and search_context is None
            and not (_wants_med and _med_skip_grounding)
            and self._should_use_google_grounding(prompt, aegis_url=aegis_url)
        ):
            # 🧠 [제국 지식 우선 규약] 웹 검색 전에 지식 베이스를 먼저 조회한다.
            #   제국 고유 명사(루나 게이트 등)를 웹이 동명이인(NASA Lunar Gateway 등)으로
            #   오답하는 사고 방지 — 옵시디언 그라운딩과 동일한 서버측 주입 패턴.
            #   히트가 있으면 그 사실을 컨텍스트로 쓰고 웹 검색은 생략한다.
            try:
                _kb = self.search_knowledge(prompt[:120], limit=3)
                if _kb and _kb.startswith("운영 지식 베이스 검색 결과:") and "관련 지식 없음" not in _kb[:40]:
                    search_context = _kb
                    logger.info("🧠 [KB Grounding] 지식 베이스 히트 — 웹 검색 생략")
            except Exception as _kb_err:
                logger.warning(f"[KB Grounding] 지식 베이스 프리패스 실패(무시): {_kb_err}")
        if (
            not skip_grounding
            and search_context is None
            and not (_wants_med and _med_skip_grounding)
            and self._should_use_google_grounding(prompt, aegis_url=aegis_url)
        ):
            try:
                if aegis_url:
                    yield "🔎 웹 검색을 실행합니다...\n\n"
                    search_context = self._search_grounding_context(prompt, aegis_url=aegis_url, token=token)
                    _engine = getattr(self, "_last_grounding_engine", "")
                    if _engine:
                        yield f"🔎 검색 엔진: {_engine}\n\n"
                else:
                    raise ValueError("Aegis URL 미설정 — Playwright 폴백")
            except Exception as search_err:
                _pw_on = ((self._tool_policy().get("search_engines") or {}).get("playwright")) is not False
                if not _pw_on:
                    logger.warning(f"[Ollama Grounding] 그라운딩 실패, Playwright는 정책상 비활성: {search_err}")
                    yield f"⚠️ 검색 실패 (폴백 엔진 비활성): {search_err}"
                    return
                logger.warning(f"[Ollama Grounding] 그라운딩 실패 → Playwright 폴백: {search_err}")
                self._observe('grounding_fallback', str(search_err), severity='warning',
                              meta={'from': 'aegis_grounding', 'to': 'playwright'}, aegis_url=aegis_url, token=token)
                try:
                    yield "🔎 로컬 Playwright 브라우저 검색을 실행합니다...\n\n"
                    search_context = self.browser.local_playwright_search(prompt)
                except Exception as pw_err:
                    logger.error(f"[Ollama Grounding] Playwright 폴백도 실패: {pw_err}")
                    yield f"⚠️ 검색 실패: {pw_err}"
                    return

        _oplap("obsidian+grounding-check")
        # 🎖 [Guardian Job M2] 장착 스킬 주입 — 예전엔 Gemini 경로에만 있어
        # 로컬 태미니(Ollama)는 직업에 스킬을 장착해도 전혀 반영되지 않았다(2026-07-10 발견).
        _equipped = (self.dynamic_config or {}).get("equipped_skills") or []
        if _equipped:
            _skill_text = self._load_equipped_skills(_equipped)
            if _skill_text:
                system_instruction = (system_instruction or "") + _skill_text
                # 📚 CLI/헤임달 '스킬 장착' 표시용 마커 (STAGE 파이프라인)
                yield f"[STAGE:📚 스킬 장착: {' · '.join(str(x) for x in _equipped[:4])}]\n"

        # 1. messages 포맷으로 구성 (/api/chat 지원)
        messages = []
        
        # [Defensive Instruction for tools]
        defensive_instruction = (
            "\n\n[⚠️ 중요 규약: 도구 사용 제한]\n"
            "1. 일반적인 채팅 응답이나 인사는 반드시 텍스트(Message)로만 수행하십시오.\n"
            "2. 'sovereign_type_text'나 'sovereign_press_key'와 같은 OS 제어 도구는 사용자가 명시적으로 "
            "시스템 조작이나 타이핑을 요청한 경우에만 사용하십시오.\n"
            "3. 절대로 채팅 응답을 타이핑 도구로 대신하지 마십시오.\n"
            "4. [중요] Playwright 브라우저 도구(mcp_playwright_browser_navigate)를 호출할 때, "
            "url 인자값을 절대로 비워두거나 빈 JSON ({})으로 보내지 마십시오. 또한 url 인자에 검색어(예: '대한민국 대통령')나 일반 질문을 넣지 마십시오. "
            "반드시 'url': 'https://www.google.com' 또는 'https://www.naver.com'과 같은 실제 http/https URL 주소를 넣어 먼저 호출하십시오.\n"
            "5. 브라우저 페이지에 들어간 후 검색창 등 입력 양식에 타이핑을 할 때는 `mcp_playwright_browser_fill_form` 이나 `mcp_playwright_browser_type`을 사용하고, "
            "반드시 selector 혹은 element 식별자가 올바른지 스냅샷(`mcp_playwright_browser_snapshot`)을 먼저 얻어 확인하십시오. Google의 경우 텍스트 입력 후 'enter' key press를 추가로 호출해야 검색이 실행됩니다."
        )
        # 🚀 방어 지침은 '도구 사용 규율'이라 도구 없는 잡담엔 불필요한 군더더기.
        # inject_guidelines(제국 가이드라인) 꺼지면 함께 빼서 로컬 Ollama 채널과
        # 동일하게 프롬프트를 최소화한다(시스템 프롬프트 0에 가깝게).
        full_system = (system_instruction or "")
        if (self.dynamic_config or {}).get("inject_guidelines", True):
            full_system += defensive_instruction
        if full_system.strip():
            messages.append({"role": "system", "content": full_system})

        if history and isinstance(history, list):
            for msg in history[-10:]:
                role = msg.get("role", "user")
                content = msg.get("content", "")
                if content:
                    messages.append({"role": role, "content": content})

        # 최종 user 프롬프트와 검색 그라운딩 컨텍스트 결합
        user_content = ""
        if search_context:
            # [🔱 Ollama 전용] 검색 결과 최대 2000자로 제한
            # 너무 길면 gemma4가 검색 결과와 시스템 프롬프트를 혼동하여 잘못된 내용을 저장
            _ctx_limit = 2000
            _ctx_trimmed = search_context[:_ctx_limit]
            if len(search_context) > _ctx_limit:
                _ctx_trimmed += "\n...(검색 결과 일부 생략)"
            user_content += f"{_ctx_trimmed}\n\n"
        user_content += prompt
        _user_msg = {"role": "user", "content": user_content}
        if _ollama_images:
            _user_msg["images"] = _ollama_images  # 🖼️ Ollama 네이티브 멀티모달
        messages.append(_user_msg)

        # 2. 도구 획득
        # [🔱 Ollama 전용] 약한 로컬 모델 오버로드 방지:
        # 의도별 필요 도구만 최소 전달 (최대 3개) — gemma4 과부하 방지
        # 🗂️ 옵시디언 '쓰기' 의도일 때만 obsidian_write 도구를 붙인다. '검색' 의도는 위에서
        #   이미 그라운딩으로 컨텍스트를 주입했으므로 도구(=불안정한 루프) 없이 답하게 한다.
        _wants_obs = not skip_tools and self._wants_obsidian(prompt) and self._wants_obsidian_write(prompt)
        _wants_search = not skip_tools and self._wants_web_search(prompt)
        _wants_hug = not skip_tools and self._wants_hugin(prompt)
        # 🛠️ 코딩 에이전트 모드 — Imperial Coding Protocol이 시스템 지시에 주입된 경우
        # (로컬 태미니 '태미니 코드' cwd 지정 등) code_* 도구 7종을 부착한다.
        # 일반 잡담에 '코드' 단어가 스쳤다고 붙이면 약한 로컬 모델이 과부하되므로,
        # 프로토콜 마커가 명시된 요청에만 한정한다.
        _wants_code = not skip_tools and (
            "Imperial Coding Protocol" in (system_instruction or "")
        )

        _wants_del = not skip_tools and self._wants_delegate(prompt)
        if _wants_del:
            # 🤝 [위임 강제 재촉] 약한 로컬 모델은 '위임해줘'를 무시하고 직접 code_* 로
            # 수행해버린다(실관측) — 위임 명시 요청에서는 delegate 도구 사용을 시스템
            # 지시로 못박는다.
            messages.append({
                "role": "system",
                "content": ("사용자가 특정 동료 수호자(코덱스/미니맥스/헤르메스/자비스/LM Studio)에게 "
                            "작업 '위임'을 명시적으로 요청했다. 그 작업을 네가 직접 수행하지 말고, "
                            "반드시 delegate_to_guardian(guardian, task) 도구를 호출해 위임하라. "
                            "task에는 대화를 못 보는 수호자도 이해할 완결된 지시문을 담아라. "
                            "도구가 돌려준 수행 결과를 그대로 정직하게 보고하라."),
            })
        if _wants_code:
            from .coding_tools import OPENAI_SCHEMAS as _CODE_SCHEMAS
            ollama_tools = list(_CODE_SCHEMAS)
            if _wants_del:
                # 🤝 코딩 콘솔에서 "코덱스에게 시켜줘" — 위임 도구를 함께 부착
                ollama_tools += self.runtime.curated_native_schemas(
                    {"only_tools": ["delegate_to_guardian"]})
        elif _wants_obs and _wants_search:
            # 복합 의도: 검색 후 옵시디언 저장 — 2개 도구
            ollama_tools = self.runtime.curated_native_schemas({"obsidian_with_search": True})
        elif _wants_obs:
            # 옵시디언만: 1개 도구
            ollama_tools = self.runtime.curated_native_schemas({"obsidian_only": True})
        elif _wants_hug:
            # 브라우저: hugin 도구만
            ollama_tools = self.runtime.curated_native_schemas({"hugin": True})
        elif not skip_tools and _wants_med:
            # 🖼️/🧊 미디어·3D·유튜브 검색: 정책이 정한 도구 묶음만 (gemma4 환각 답변 방지)
            ollama_tools = self.runtime.curated_native_schemas(
                {"only_tools": self._intent_cfg("media").get("tools")})
        elif _wants_del:
            # 🤝 위임 의도만: delegate 도구 1종 (약한 모델 과부하 방지)
            ollama_tools = self.runtime.curated_native_schemas(
                {"only_tools": self._intent_cfg("delegate").get("tools")})
        else:
            ollama_tools = None

        host = (getattr(self, "ollama_host", None) or "http://127.0.0.1:11434").rstrip("/")
        if host.startswith("0.0.0.0"):
            host = host.replace("0.0.0.0", "127.0.0.1")
        if not host.startswith("http://") and not host.startswith("https://"):
            host = f"http://{host}"
        if ":" not in host[8:]:
            host = f"{host}:11434"

        model = model_name or self.primary_model
        # 🦙 [모델 태그 보정] Ollama는 '정확한 설치 태그'만 받는다. 태그를 뺀 'gemma4'는
        #   'gemma4:latest'로 해석되는데 그 태그가 없으면 404("model not found")가 나고,
        #   provider만 ollama로 바뀌고 모델은 다른 프로바이더 값(예: 'gemini-3.1-flash-lite')이
        #   그대로 넘어와도 404다. 릴레이(_resolve_installed_model)와 동일하게 설치 목록과
        #   대조해, 없거나 태그가 어긋나면 첫 설치 모델로 폴백한다 → 404 대신 자동 정합.
        try:
            from blueprints.relay_common import _resolve_installed_model
            _resolved_model = _resolve_installed_model(
                host, model, "/api/tags", "models", "name", model
            )
            if _resolved_model and _resolved_model != model:
                logger.info(
                    f"[Ollama] 모델 태그 보정: '{model}' → '{_resolved_model}' (설치 목록 기준)"
                )
                model = _resolved_model
        except Exception as _resolve_err:
            logger.debug(f"[Ollama] 모델 태그 보정 생략: {_resolve_err}")
        # [🎖️ Guardian Job] 직업(capabilities)에 num_ctx가 지정되면 최우선 적용.
        # 미지정(0/None)이면 검색 컨텍스트 유무에 따라 합리적 기본값(8k/4k)을 쓴다.
        try:
            _cfg_ctx = int((self.dynamic_config or {}).get("num_ctx") or 0)
        except Exception:
            _cfg_ctx = 0
        try:
            _num_ctx = _cfg_ctx or int(getattr(self, "ollama_num_ctx", 0) or (8192 if search_context else 4096))
        except Exception:
            _num_ctx = 4096
        # 🛠️ 코딩 에이전트 모드: 파일 내용 + 도구 결과가 컨텍스트를 크게 차지하므로
        # 최소 16k를 보장한다 (4k면 시스템 프롬프트+코드 몇 줄에 바로 잘림).
        if _wants_code:
            _num_ctx = max(_num_ctx, 16384)
        # [🎖️ Guardian Job] num_predict = 응답 토큰 상한. 0/미지정이면 무제한(-1).
        # gemma 등이 짧은 인사에도 수백~수천 토큰을 생성해 느려지는 문제를 직업별로
        # 캡한다. 실측: 생성 속도는 고정(~tok/s)이라 토큰 수가 곧 응답 시간이다.
        try:
            _num_predict = int((self.dynamic_config or {}).get("num_predict") or 0)
        except Exception:
            _num_predict = 0

        _oplap("messages+tools-build")
        logger.info(
            f"[Ollama] model={model} messages_count={len(messages)} "
            f"has_search_ctx={bool(search_context)} num_ctx={_num_ctx} "
            f"num_predict={_num_predict or '∞'} has_tools={bool(ollama_tools)}"
        )

        try:
            max_iterations = self.dynamic_config.get("max_iterations", 15)
            ctx = RuntimeCtx(prompt, interactive=True, debug=_is_debug_mode(), display_name="Ollama")
            ctx.expect_obsidian_write = self._wants_obsidian(prompt) and self._wants_obsidian_write(prompt)
            # 🔇 [코딩 모드] 중간 턴(도구 호출 전 계획 발화, 재촉당한 턴)을 스트림에 내보내지
            # 않는다 — 재촉 루프가 돌 때마다 장황한 문단이 응답에 누적되던 문제 차단.
            # 최종 턴은 handle_no_tool_calls의 flush가 한 번만 방출한다.
            if _wants_code and not stream_tokens:
                # 🌊 stream_tokens(태미니 CLI send-stream)에서는 억류하지 않는다 —
                # 자비스처럼 도구 호출 사이 나레이션과 토큰이 실시간으로 흐르게.
                ctx.suppress_stream = True

            for _ in range(max_iterations):
                _opts = {"num_ctx": _num_ctx}
                if _num_predict > 0:
                    _opts["num_predict"] = _num_predict
                json_data = {
                    "model": model,
                    "messages": messages,
                    "stream": True,
                    "options": _opts,
                    # 🦙 모델을 VRAM에 30분 상주시켜 매 요청 콜드 로딩(실측 load=7.3s)을 제거.
                    # 로컬 채널이 빠른 이유 중 하나가 모델이 따뜻하게 떠 있기 때문.
                    "keep_alive": (os.environ.get("OLLAMA_KEEP_ALIVE") or "30m"),
                    # 🧠 [Thinking OFF] gemma4 등 추론(thinking) 모델은 답변 전에 수백 토큰을
                    # 추론에 쓰는데, 그 토큰은 content가 아니라 message.thinking으로 가 버려진다.
                    # → 시간만 잡아먹고(실측 답변 4.3s vs 12s+), num_predict가 낮으면 추론 단계에서
                    # 잘려 답이 빈다(content_len=0). 채팅 속도 우선이므로 thinking을 끈다.
                    # 비추론 모델은 이 필드를 무시한다. OLLAMA_THINK=1로 재활성화 가능.
                    "think": (os.environ.get("OLLAMA_THINK", "").lower() in ("1", "true", "yes")),
                }
                if ollama_tools and not ctx.force_no_tools:
                    json_data["tools"] = ollama_tools

                _ollama_post_at = time.time()
                logger.info(f"⏱️ [Ollama] POST /api/chat 시작 (진입후 {(time.time() - _op_t0)*1000:.0f}ms, msgs={len(messages)}, sys_len={sum(len(m.get('content') or '') for m in messages if m.get('role')=='system')})")
                response = requests.post(
                    f"{host}/api/chat",
                    json=json_data,
                    stream=True,
                    timeout=180,
                )
                response.raise_for_status()

                full_content = ""
                tool_calls = []
                first_yield = True
                _usage_acc = None  # 🪙 토큰 사용량(done 청크)
                # 🛡️ [런어웨이 가드] 일부 깨진 모델 태그(예: gemma4:12b-mlx)는 EOS 대신
                #   특수토큰 <pad>/<eos>/<unusedNN>을 무한 생성한다. num_predict 무제한이면
                #   180s 타임아웃까지 매달린다 → 연속 특수토큰이 임계 초과면 조기 중단·관측.
                _junk_run = 0
                _runaway_hit = False
                _JUNK_RE = re.compile(r"<pad>|<eos>|<unused\d*>")

                def _is_junk(txt: str) -> bool:
                    # 원문은 비어있지 않은데, 특수토큰(<pad>/<eos>/<unusedNN>)을 다 지우면
                    # 아무것도 안 남는 청크 = 런어웨이성 쓰레기 출력.
                    if not (txt or "").strip():
                        return False
                    return _JUNK_RE.sub("", txt).strip() == ""

                for line in response.iter_lines():
                    if not line:
                        continue
                    try:
                        data = json.loads(line.decode("utf-8"))
                    except Exception:
                        continue

                    message_chunk = data.get("message", {})
                    content = message_chunk.get("content", "")
                    if content:
                        # 특수토큰만으로 된 청크가 연속되면 런어웨이로 판정
                        if _is_junk(content):
                            _junk_run += 1
                            if _junk_run >= 12:
                                _runaway_hit = True
                                logger.error("[Ollama] 런어웨이 감지(<pad> 등 특수토큰 반복) — 모델=%s 중단", model)
                                self._observe(
                                    'model_runaway',
                                    f"{model}이 특수토큰(<pad> 등)을 반복 생성 — 조기 중단",
                                    severity='error',
                                    meta={'model': model, 'junk_run': _junk_run, 'chars': len(full_content)},
                                    aegis_url=aegis_url, token=token)
                                try:
                                    response.close()
                                except Exception:
                                    pass
                                break
                            continue  # junk 청크는 누적/방출하지 않음
                        _junk_run = 0
                        full_content += content
                        if first_yield:
                            first_yield = False
                            logger.info(f"⏱️ [Ollama] 첫 콘텐츠 청크 — prompt eval {(time.time() - _ollama_post_at)*1000:.0f}ms")
                        if not ctx.suppress_stream:
                            yield content

                    tc_chunk = message_chunk.get("tool_calls", [])
                    for tc in tc_chunk:
                        # Ollama tool call formats: {"function": {"name": "...", "arguments": {...}}}
                        fn = tc.get("function") or {}
                        name = fn.get("name", "")
                        args = fn.get("arguments", "")
                        if isinstance(args, dict):
                            args = json.dumps(args, ensure_ascii=False)
                        
                        tc_id = tc.get("id") or f"call_ollama_{len(tool_calls)}"
                        # 중복 수집 방지 (Ollama 스트림 청크 누적 대응)
                        if not any(x["id"] == tc_id for x in tool_calls):
                            tool_calls.append({
                                "id": tc_id,
                                "type": "function",
                                "function": {
                                    "name": name,
                                    "arguments": args or "{}",
                                }
                            })

                    if data.get("done") is True:
                        _usage_acc = {
                            "prompt_tokens": data.get("prompt_eval_count"),
                            "completion_tokens": data.get("eval_count"),
                        }
                        # 🦙 [Ollama 시간 분해] load(모델 로딩)/prompt_eval(입력 처리)/eval(토큰 생성)
                        # 실제 소요시간(ns)을 초로 환산해 진단 마커로 노출. 콜드 로딩 vs 입력 처리 vs
                        # 토큰 생성 중 무엇이 병목인지 한눈에 갈린다. stage_timing 켜졌을 때만.
                        try:
                            _ld = (data.get("load_duration") or 0) / 1e9
                            _pe = (data.get("prompt_eval_duration") or 0) / 1e9
                            _ev = (data.get("eval_duration") or 0) / 1e9
                            _pc = data.get("prompt_eval_count") or 0
                            _ec = data.get("eval_count") or 0
                            logger.info(
                                f"[Ollama 시간분해] model={model} load={_ld:.2f}s "
                                f"prompt_eval={_pe:.2f}s({_pc}tok) eval={_ev:.2f}s({_ec}tok) "
                                f"full_content_len={len(full_content)} suppress={getattr(ctx, 'suppress_stream', None)} "
                                f"content_preview={full_content[:60]!r}"
                            )
                            if (self.dynamic_config or {}).get("stage_timing", True):
                                # ⚠️ 반드시 '[STAGE:'로 시작해야 한다(앞에 \n 붙이면 수신측
                                # startswith 검사를 통과 못해 응답 본문에 섞여 오염됨).
                                yield (
                                    f"[STAGE:🦙 로드 {_ld:.1f}s · 입력 {_pe:.1f}s({_pc}tok) · "
                                    f"생성 {_ev:.1f}s({_ec}tok)|0]"
                                )
                        except Exception:
                            pass
                        break

                # 🛡️ 런어웨이로 중단됐고 실제 본문이 없으면, 빈 '생각중' 대신 안내를 내보낸다.
                if _runaway_hit and not full_content.strip() and not tool_calls:
                    yield (f"⚠️ 선택된 모델(`{model}`)이 비정상 출력(<pad> 등 특수토큰 반복)을 냈습니다. "
                           f"이 모델 태그가 깨진 것으로 보입니다 — 다른 모델(예: gemma4:31b-mlx)로 바꿔 주세요.")
                    return

                self._emit_token_usage("ollama", model, _usage_acc)
                if not tool_calls:
                    # 🛠️ [코딩 재촉] ① 말로만 계획하고 도구를 안 부르는 경우,
                    # ② 'code_edit_file(...)' 같은 도구 호출을 텍스트로 흉내만 낸 경우(가짜 호출),
                    # ③ 도구를 쓰다가 "수정하겠습니다" 같은 약속만 하고 멈춘 경우(약속-미이행)
                    # 실제 code_* 호출을 재촉한다 (옵시디언 nudge와 동일 패턴, 최대 3회).
                    # '완료' 보고가 포함된 텍스트는 최종 보고로 보고 재촉하지 않는다.
                    _fake_code_call = _wants_code and bool(re.search(
                        r"code_(?:list_dir|glob|grep|read_file|write_file|edit_file|run_command)\s*\(",
                        full_content or "",
                    ))
                    _code_promise = _wants_code and "완료" not in (full_content or "") and any(
                        ph in (full_content or "") for ph in (
                            "하겠습니다", "할게요", "수정하겠", "진행하겠", "실행하겠",
                            "검증하겠", "분석하겠", "시작하겠", "고치겠", "작성하겠",
                        )
                    )
                    if (_wants_code and full_content.strip()
                            and getattr(ctx, "coding_nudges", 0) < 3
                            and (not ctx.used_tools or _fake_code_call or _code_promise)):
                        ctx.coding_nudges = getattr(ctx, "coding_nudges", 0) + 1
                        messages.append({"role": "assistant", "content": full_content})
                        if _fake_code_call:
                            _nudge_msg = (
                                "방금 출력한 'code_...(...)' 표기는 텍스트일 뿐이며 아무것도 "
                                "실행되지 않았다. 반드시 실제 함수 호출(tool call)로 다시 실행하라. "
                                "호출 전에는 절대 '완료'라고 말하지 마라."
                            )
                        elif _code_promise and ctx.used_tools:
                            _nudge_msg = (
                                "'하겠습니다'라고 약속만 하고 멈추지 마라. 말한 작업을 지금 즉시 "
                                "code_edit_file / code_run_command 등 실제 도구 호출로 수행하고, "
                                "검증까지 끝난 뒤에 최종 보고와 taemini-todo 블록을 출력하라."
                            )
                        else:
                            _nudge_msg = (
                                "말로만 계획하지 말고 지금 즉시 code_read_file / code_grep / "
                                "code_list_dir 등 code_* 도구를 실제로 호출해 작업을 시작하라. "
                                "도구를 호출하지 않으면 파일은 읽히지도 수정되지도 않는다. "
                                "이미 작업을 끝냈고 검증까지 통과했다면 최종 보고와 taemini-todo "
                                "블록만 출력하라."
                            )
                        messages.append({"role": "user", "content": _nudge_msg})
                        logger.info(
                            f"🛠️ [Coding Nudge] "
                            f"{'가짜 텍스트 호출' if _fake_code_call else ('약속-미이행' if _code_promise and ctx.used_tools else '도구 미호출')} "
                            f"감지 — code_* 강제 재촉 ({ctx.coding_nudges}/3)"
                        )
                        continue

                    # [🗂️ obsidian 직접 폴백] tool_calls 없는데 obsidian 의도이면
                    # 텍스트 응답에서 직접 노트 작성 실행 (모델이 tool calling 못 할 때)
                    if ctx.expect_obsidian_write and not ctx.obsidian_written and full_content.strip():
                        _obsidian_result = yield from self._ollama_direct_obsidian_write(prompt, full_content, ctx)
                        if _obsidian_result == "continue":
                            continue
                        break

                    # 🛡️ [Tool Runtime] 도구 없는 응답 — 구제 처리 일원화
                    _action = yield from self.runtime.handle_no_tool_calls(full_content, messages, ctx)
                    if _action == "continue":
                        continue
                    # 🗂️ obsidian 재촉 (약속만 하고 도구 미호출)
                    _nudge = self.runtime.nudge_obsidian_write(full_content, messages, ctx)
                    if _nudge == "continue":
                        continue
                    break

                # 도구 호출 실행
                _n_before = len(messages)
                yield from self.runtime.run_tool_calls(tool_calls, full_content, messages, ctx)

                if _wants_code:
                    # 🛠️ [코딩 에이전트 루프] 읽기→수정→검증→반복이 가능하도록 루프를 계속한다.
                    # gemma4 등이 tool role 메시지 포함 follow-up을 400으로 거부하므로,
                    # 방금 붙은 OpenAI식 tool 메시지를 일반 텍스트 턴(assistant/user)으로
                    # 변환해 어떤 로컬 모델에서도 후속 추론이 가능하게 한다.
                    # ⚠️ assistant 턴에 '[도구 호출: ...]' 같은 텍스트 표기를 남기면 약한
                    # 모델이 그 포맷을 흉내 내 '가짜(텍스트) 도구 호출'을 출력한다 — 도구
                    # 이름/인자는 user 턴의 결과 블록에만 남긴다.
                    _appended = messages[_n_before:]
                    del messages[_n_before:]
                    _tool_lines = []
                    for _m in _appended:
                        if _m.get("role") == "assistant":
                            _atext = str(_m.get("content") or "").strip()
                            if _atext:
                                messages.append({"role": "assistant", "content": _atext})
                        elif _m.get("role") == "tool":
                            _tool_lines.append(
                                f"### {_m.get('name')} 결과\n{str(_m.get('content', ''))[:6000]}"
                            )
                    if _tool_lines:
                        messages.append({
                            "role": "user",
                            "content": (
                                "[도구 실행 결과]\n" + "\n\n".join(_tool_lines) +
                                "\n\n위 결과를 반영해 다음 단계를 계속 진행하라 (필요하면 code_* "
                                "도구를 다시 호출). 모든 수정과 검증(실행/테스트)이 끝났을 때만 "
                                "최종 보고와 taemini-todo 블록을 출력하고 마쳐라."
                            ),
                        })
                    continue

                # [🔱 Ollama 전용] 도구 실행 완료 후 루프 종료
                # gemma4 등이 tool role 메시지 포함 follow-up을 400으로 거부하므로
                # 도구만 실행하고 바로 종료한다 (요약 없어도 충분)
                break

        except requests.exceptions.HTTPError as http_err:
            _status = getattr(getattr(http_err, "response", None), "status_code", 0)
            if _status == 400 and ctx.used_tools:
                # [✅ 정상] 도구 실행 후 follow-up 400 — 이미 완료됨
                logger.info("[Ollama] 도구 실행 후 follow-up 400 (정상 동작) — 종료")
            elif _status == 400:
                # [⚠️ tools 미지원] 도구 없이 재시도
                logger.warning("[Ollama] 400 Bad Request — tools 미지원 모델. 도구 없이 재시도...")
                yield from self._think_ollama(
                    prompt, model_name, history=history,
                    system_instruction=system_instruction, images=images,
                    token=token, aegis_url=aegis_url, skip_tools=True,
                    skip_grounding=skip_grounding,
                )
            elif _status == 404:
                # [⚠️ 모델 없음] Ollama는 '정확한 설치 태그'만 받는다. 요청 모델이
                #   설치돼 있지 않으면 404("model not found"). 설치 목록을 '지금 다시'
                #   조회해 유효한 모델로 '자동 재시도'한다 → 유저가 수동으로 동기화/선택
                #   하지 않아도 대부분 그대로 성공. 재조회해도 바뀔 게 없으면(설치 모델이
                #   아예 없음) 무한루프 없이 안내로 폴백한다.
                logger.warning(f"[Ollama] 404 모델 없음: 요청='{model}' — 목록 재조회 후 자동 재시도")
                _retry_model = None
                try:
                    from blueprints.relay_common import _resolve_installed_model
                    _cand = _resolve_installed_model(
                        host, model, "/api/tags", "models", "name", model
                    )
                    if _cand and _cand != model:
                        _retry_model = _cand
                except Exception as _re_err:
                    logger.debug(f"[Ollama] 404 재조회 실패: {_re_err}")

                if _retry_model:
                    logger.info(f"[Ollama] 404 자동 복구: '{model}' → '{_retry_model}' 재시도")
                    yield from self._think_ollama(
                        prompt, _retry_model, history=history,
                        system_instruction=system_instruction, images=images,
                        token=token, aegis_url=aegis_url, skip_tools=skip_tools,
                        tools=tools, skip_grounding=True,  # 그라운딩은 첫 패스에서 완료 — 중복 방지
                    )
                else:
                    # 재조회해도 유효 모델이 없음 = 설치된 모델이 아예 없는 경우 → 안내.
                    yield (
                        f"❌ 로컬 Ollama에 설치된 모델이 없어 응답할 수 없습니다(요청: `{model}`).\n\n"
                        f"👉 터미널에서 `ollama pull <모델명>` 으로 모델을 먼저 받은 뒤, "
                        f"**브레인 설정 → '🔄 로컬 Ollama 모델 동기화'** 를 눌러 목록을 갱신해 주세요."
                    )
            else:
                logger.error(f"[Ollama] HTTP 오류: {http_err}")
                yield f"❌ 로컬 Ollama 통신 오류: {http_err}"
        except Exception as e:
            logger.error(f"[Ollama] Local chat request failed: {e}")
            yield f"❌ 로컬 Ollama 통신 오류: {e}"

    # ──────────────────────────────────────────────────────────────
    # [🗂️ Ollama 폴백] gemma4 등 tool_calls 미지원 모델 전용
    # ──────────────────────────────────────────────────────────────
    _GEMMA_JSON_TOOL_RE = re.compile(
        r"```(?:json)?\s*(\{[\s\S]*?\})\s*```", re.IGNORECASE
    )

    def _parse_gemma_tool_call(self, text: str):
        """gemma4가 ```json {...} ``` 형식으로 출력하는 도구 호출을 파싱한다.
        성공 시 (func_name, args_dict) 반환, 실패 시 (None, None)."""
        for m in self._GEMMA_JSON_TOOL_RE.finditer(text):
            try:
                obj = json.loads(m.group(1))
                # {"name": "obsidian_write", "parameters": {...}} 또는
                # {"tool": "obsidian_write", "arguments": {...}} 등 다양한 키 지원
                name = obj.get("name") or obj.get("tool") or obj.get("function") or ""
                args = (obj.get("parameters") or obj.get("arguments")
                        or obj.get("args") or {})
                if name and isinstance(args, dict):
                    return name, args
            except Exception:
                continue
        return None, None

    def _ollama_direct_obsidian_write(self, user_prompt: str, ai_response: str, ctx):
        """[🗂️ Ollama 폴백] 모델이 tool_calls API를 발화 못할 때 직접 obsidian_write 실행.
        1순위: gemma4 ```json 툴콜 텍스트 파싱 → 실제 인자로 실행
        2순위: AI 텍스트 응답을 그대로 노트 내용으로 저장"""
        import threading

        # --- 1단계: gemma4 ```json 파싱 시도 ---
        parsed_name, parsed_args = self._parse_gemma_tool_call(ai_response)
        if parsed_name == "obsidian_write" and parsed_args:
            logger.info(f"[🗂️ Ollama] gemma4 ```json 툴콜 파싱 성공 → {parsed_args}")
            write_args = parsed_args
            note_path = parsed_args.get("path") or parsed_args.get("title", "올라마 노트")
        else:
            # --- 2단계: 텍스트 직접 저장 (파싱 실패 시) ---
            # 응답이 ```json ... ``` 뿐이면 실제 내용 없음 → 사용자 프롬프트로 제목 설정
            clean_text = self._GEMMA_JSON_TOOL_RE.sub("", ai_response).strip()
            if not clean_text:
                # 모델이 JSON만 뱉고 실제 내용이 없음 → 기본 내용 생성
                clean_text = f"# {user_prompt}\n\n(Ollama 자동 생성 노트)"
            note_path = user_prompt.strip()[:60].replace("/", "-").replace("\\", "-")
            write_args = {"path": note_path, "content": clean_text, "mode": "overwrite"}

        logger.info(f"[🗂️ Ollama Direct Obsidian] 저장 시도: '{note_path}'")

        # --- 3단계: 타임아웃 스레드로 블로킹 방지 (최대 10초) ---
        result_holder = [None]
        error_holder = [None]

        def _write():
            try:
                result_holder[0] = self.runtime.dispatch("obsidian_write", write_args, ctx)
            except Exception as exc:
                error_holder[0] = str(exc)

        t = threading.Thread(target=_write, daemon=True)
        t.start()
        t.join(timeout=10)

        if t.is_alive():
            logger.error("[🗂️ Ollama Direct Obsidian] 타임아웃(10s) — Obsidian REST API 응답 없음")
            yield f"\n\n⚠️ Obsidian 저장 시도했으나 응답 없음(10초 초과). Obsidian이 실행 중인지 확인해주세요."
            return "break"

        ctx.obsidian_written = True
        result = result_holder[0] or error_holder[0] or "알 수 없는 오류"

        if error_holder[0] or "실패" in str(result) or "Error" in str(result):
            logger.error(f"[🗂️ Ollama Direct Obsidian] 저장 실패: {result}")
            yield f"\n\n⚠️ Obsidian 저장 실패: {result}"
        else:
            yield f"\n\n✅ Obsidian에 노트를 저장했습니다: **{note_path}**"
        return "break"

    def _ensure_hermes_proxy_running(self):
        """🪽 hermes proxy(8645)가 꺼져 있으면 백그라운드로 자동 기동한다."""
        import socket
        try:
            with socket.create_connection(('127.0.0.1', 8645), timeout=1):
                return True
        except OSError:
            pass
        try:
            home = os.path.expanduser('~')
            exe = os.path.join(home, 'AppData', 'Local', 'hermes', 'hermes-agent',
                               'venv', 'Scripts', 'hermes.exe')
            if not os.path.exists(exe):
                logger.warning('[HermesProxy] hermes.exe 미설치 — 프록시 자동 기동 불가')
                return False
            flags = subprocess.CREATE_NO_WINDOW if os.name == 'nt' else 0
            subprocess.Popen([exe, 'proxy', 'start'],
                             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                             creationflags=flags)
            logger.info('[HermesProxy] hermes proxy 자동 기동 (포트 8645)...')
            for _ in range(15):
                time.sleep(1)
                try:
                    with socket.create_connection(('127.0.0.1', 8645), timeout=1):
                        return True
                except OSError:
                    continue
            logger.warning('[HermesProxy] 15초 내 기동 확인 실패')
            return False
        except Exception as exc:
            logger.warning('[HermesProxy] 자동 기동 실패: %s', exc)
            return False

    def _think_openrouter(
        self,
        prompt: str,
        model_name: str,
        system_instruction: str = None,
        skip_tools: bool = False,
        tools: list = None,
        session_id: str = None,
        provider: str = "openrouter",
        interactive: bool = True,
        images: list = None
    ) -> Generator[str, None, None]:
        """OpenRouter 프로바이더 및 OpenClaw 등 OpenAI 호환 API를 호출합니다."""
        
        if provider == "openclaw":
            api_key = os.environ.get("OPENCLAW_GATEWAY_TOKEN", "imperial1234")
            base_url = "http://127.0.0.1:18789/v1"
            display_name = "OpenClaw"
        elif provider in ("nous", "hermes_proxy"):
            # 🪽 [Hermes Proxy] OAuth 인증된 Nous Portal을 hermes proxy(8645)가
            # OpenAI 호환으로 중계 — API 키 불필요 (아무 Bearer나 허용, 실제 자격증명은 프록시가 부착)
            api_key = "imperial-local"
            base_url = os.environ.get("HERMES_PROXY_URL", "http://127.0.0.1:8645").rstrip("/") + "/v1"
            display_name = "Nous Portal (Hermes Proxy)"
            self._ensure_hermes_proxy_running()
        elif provider == "relay":
            api_key = "imperial-local-bypass"
            base_url = "http://127.0.0.1:18710/api/v1" # Rust Brain for relay? or something else
            display_name = "Relay"
        elif provider == "nvidia":
            # [🔱 NVIDIA NIM] MiniMax-M3/M2.7 (OpenAI 호환, non-streaming)
            # M3·M2.7 모두 동일 키로 통합 — NVIDIA_MINIMAX_M3_API_KEY 우선, 없으면 NVIDIA_API_KEY
            api_key = self.nvidia_minimax_m3_api_key or self.nvidia_api_key
            base_url = "https://integrate.api.nvidia.com/v1"
            display_name = "NVIDIA NIM"
            if not api_key:
                yield "❌ [NVIDIA] API Key가 없습니다. Bitwarden에 'Imperial:NVIDIA_MINIMAX_M3_API_KEY'를 등록해주세요."
                return
        elif provider == "glm":
            # [🔱 GLM / Z.ai] zai-org GLM-5.1/5.2 (modal.direct OpenAI 호환)
            api_key = self.glm_api_key
            base_url = "https://api.us-west-2.modal.direct/v1"
            display_name = "Z.ai GLM"
            if not api_key:
                yield "❌ [GLM] API Key가 없습니다. Bitwarden에 'Imperial:GLM_API_KEY'를 등록해주세요."
                return
        else:
            api_key = self.openrouter_api_key
            base_url = "https://openrouter.ai/api/v1"
            display_name = "OpenRouter"
            if not api_key:
                yield "❌ [OpenRouter] API Key가 없습니다. Bitwarden에 'Imperial:OPENROUTER_CLAUDE_TOKEN'을 등록해주세요."
                return

        # 🖼️ [Multimodal] 이미지가 첨부되면 OpenAI 호환 멀티모달 형식으로 user 메시지를 구성한다.
        # (선택한 모델이 비전 지원이면 이 이미지를 보고 분석한다. 미지원 모델이면 무시/에러)
        if images:
            _content = [{"type": "text", "text": prompt}]
            for _img in images:
                _data = _img.get("data") or ""
                if isinstance(_data, str) and "," in _data and _data.strip().startswith("data:"):
                    _data = _data.split(",", 1)[1]
                _mime = _img.get("mimeType") or "image/jpeg"
                if _data:
                    _content.append({
                        "type": "image_url",
                        "image_url": {"url": f"data:{_mime};base64,{_data}"},
                    })
            messages = [{"role": "user", "content": _content}]
            logger.info(f"🖼️ [{display_name}] 멀티모달 입력 — 이미지 {len(images)}장을 모델 '{model_name}'에 전달.")
        else:
            messages = [{"role": "user", "content": prompt}]

        # [🔱 Defensive Instruction] OpenRouter 모델들이 채팅 응답 대신 OS 제어 도구를 남용하는 것 방어 및 Playwright 오용 방지
        defensive_instruction = (
            "\n\n[⚠️ 중요 규약: 도구 사용 제한]\n"
            "1. 일반적인 채팅 응답이나 인사는 반드시 텍스트(Message)로만 수행하십시오.\n"
            "2. 'sovereign_type_text'나 'sovereign_press_key'와 같은 OS 제어 도구는 사용자가 명시적으로 "
            "시스템 조작이나 타이핑을 요청한 경우에만 사용하십시오.\n"
            "3. 절대로 채팅 응답을 타이핑 도구로 대신하지 마십시오.\n"
            "4. [중요] Playwright 브라우저 도구(mcp_playwright_browser_navigate)를 호출할 때, "
            "url 인자값을 절대로 비워두거나 빈 JSON ({})으로 보내지 마십시오. 또한 url 인자에 검색어(예: '대한민국 대통령')나 일반 질문을 넣지 마십시오. "
            "반드시 'url': 'https://www.google.com' 또는 'https://www.naver.com'과 같은 실제 http/https URL 주소를 넣어 먼저 호출하십시오.\n"
            "5. 브라우저 페이지에 들어간 후 검색창 등 입력 양식에 타이핑을 할 때는 `mcp_playwright_browser_fill_form` 이나 `mcp_playwright_browser_type`을 사용하고, "
            "반드시 selector 혹은 element 식별자가 올바른지 스냅샷(`mcp_playwright_browser_snapshot`)을 먼저 얻어 확인하십시오. Google의 경우 텍스트 입력 후 'enter' key press를 추가로 호출해야 검색이 실행됩니다."
        )
        
        full_system_instruction = (system_instruction or "") + defensive_instruction
        messages.insert(0, {"role": "system", "content": full_system_instruction})

        # openclaw은 자체 플러그인 도구를 보유 — 섀도우브레인 도구를 함께 보내면 이름 충돌 발생
        or_tools = self._get_tools_schema(prompt) if (not skip_tools and provider != "openclaw") else None

        try:
            logger.info(f"[{display_name}] 모델 '{model_name}' 호출 시작")
            max_iterations = self.dynamic_config.get("max_iterations", 15)
            first_yield = True
            def _cfg_int(name, default, min_value=1, max_value=8192):
                try:
                    value = int(self.dynamic_config.get(name, default))
                except (TypeError, ValueError):
                    value = default
                return max(min_value, min(max_value, value))

            m3_tool_max_tokens = _cfg_int("nvidia_minimax_m3_tool_max_tokens", 768, 128, 4096)
            m3_final_max_tokens = _cfg_int("nvidia_minimax_m3_final_max_tokens", 1024, 128, 4096)
            m3_chat_max_tokens = _cfg_int("nvidia_minimax_m3_max_tokens", 2048, 256, 8192)
            try:
                m3_temperature = float(self.dynamic_config.get("nvidia_minimax_m3_temperature", 0.4))
            except (TypeError, ValueError):
                m3_temperature = 0.4
            try:
                m3_top_p = float(self.dynamic_config.get("nvidia_minimax_m3_top_p", 0.9))
            except (TypeError, ValueError):
                m3_top_p = 0.9
            # 🛡️ [Tool Runtime] 도구 루프 상태를 ctx로 일원화 (3경로 공통)
            ctx = RuntimeCtx(prompt, interactive=interactive,
                             debug=_is_debug_mode(), display_name=display_name)
            # 🗂️ 옵시디언 작성 의도면, 약속만 하고 obsidian_write를 안 부르면 재촉하도록 표시
            ctx.expect_obsidian_write = self._wants_obsidian(prompt) and self._wants_obsidian_write(prompt)

            for _ in range(max_iterations):
                # [🔱 Relay Pinning] 마왕님의 요청에 따라 특정 채널(Relay)로 강제 고정
                # session_id가 제공된 경우(DM 등) 이를 우선적으로 사용합니다.
                effective_session = session_id or "mawang_jarvis_relay"

                # NVIDIA NIM: nemotron은 streaming, 나머지(MiniMax/DeepSeek/StepFun)는 non-streaming
                _nvidia_streaming_models = {"nvidia/nemotron-3-ultra-550b-a55b", "z-ai/glm-5.1"}
                is_nvidia_streaming = (
                    provider == "nvidia"
                    and str(model_name or "").lower() in _nvidia_streaming_models
                )
                is_nvidia_minimax_m3 = provider == "nvidia" and not is_nvidia_streaming

                json_data = {
                    "model": model_name,
                    "messages": messages,
                    "stream": False if is_nvidia_minimax_m3 else True,
                }
                if is_nvidia_minimax_m3:
                    # NVIDIA MiniMax-M3 is sensitive to the generic OpenRouter-style
                    # streaming/tool loop. Match NVIDIA's reference request shape.
                    m3_has_tools = bool(or_tools and not ctx.force_no_tools)
                    json_data.update({
                        "max_tokens": (
                            m3_tool_max_tokens
                            if m3_has_tools
                            else m3_final_max_tokens
                            if ctx.force_no_tools
                            else m3_chat_max_tokens
                        ),
                        "temperature": m3_temperature,
                        "top_p": m3_top_p,
                    })
                else:
                    json_data["user"] = effective_session
                    json_data["stream_options"] = {"include_usage": True}  # 🪙 토큰 사용량 집계
                if or_tools and not ctx.force_no_tools:
                    json_data["tools"] = or_tools
                    json_data["tool_choice"] = "auto"

                res = requests.post(
                    f"{base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {api_key}",
                        "Content-Type": "application/json",
                        "Accept": "application/json" if is_nvidia_minimax_m3 else "text/event-stream",
                        **({} if provider == "nvidia" else {
                            "HTTP-Referer": "https://jarvis-imperial.com",
                            "X-Title": "Imperial Shadow Brain",
                        }),
                    },
                    json=json_data,
                    stream=not is_nvidia_minimax_m3,
                    timeout=300
                )

                if res.status_code == 429:
                    yield f"⚠️ [{display_name}] Rate Limit 초과 (무료 모델 한도). 잠시 후 다시 시도하거나 다른 모델을 선택하세요."
                    return
                res.raise_for_status()

                if is_nvidia_minimax_m3:
                    data = res.json()
                    self._emit_token_usage(provider, model_name, data.get("usage"))  # 🪙
                    message = ((data.get("choices") or [{}])[0].get("message") or {})
                    reasoning = message.get("reasoning_content") or ""
                    content = message.get("content") or ""
                    # 🧠 [Reasoning 게이트] show_reasoning이 꺼져 있으면(기본) 추론 채널을
                    # 최종 답변에서 제외한다 — minimax 계열의 메타/지시 메아리 누출 차단.
                    # content가 비면 reasoning으로 폴백(답변 유실 방지).
                    if _show_reasoning():
                        full_content = f"{reasoning}{content}".strip()
                    else:
                        full_content = _pick_nonreasoning_final(content, reasoning).strip()
                    tool_calls = message.get("tool_calls") or []

                    normalized_tool_calls = []
                    for i, tc in enumerate(tool_calls):
                        fn = (tc.get("function") or {}) if isinstance(tc, dict) else {}
                        args = fn.get("arguments", "")
                        if isinstance(args, (dict, list)):
                            args = json.dumps(args, ensure_ascii=False)
                        normalized_tool_calls.append({
                            "id": tc.get("id") or f"call_m3_{i}",
                            "type": tc.get("type") or "function",
                            "function": {
                                "name": fn.get("name") or "",
                                "arguments": args or "",
                            },
                        })

                    if not normalized_tool_calls:
                        if full_content:
                            # 🗂️ 옵시디언 작성 약속만 하고 도구 미호출이면 한 번 더 강제 재촉
                            _nudge = self.runtime.nudge_obsidian_write(full_content, messages, ctx)
                            if _nudge == "continue":
                                continue
                            yield full_content
                        else:
                            _action = yield from self.runtime.handle_no_tool_calls(full_content, messages, ctx)
                            if _action == "continue":
                                continue
                        break

                    yield from self.runtime.run_tool_calls(normalized_tool_calls, full_content, messages, ctx)

                    m3_tool_names = [
                        (tc.get("function") or {}).get("name", "")
                        for tc in normalized_tool_calls
                    ]
                    browser_mutated = any(name in {
                        "mcp_playwright_browser_type",
                        "mcp_playwright_browser_fill_form",
                        "mcp_playwright_browser_click",
                        "mcp_playwright_browser_press_key",
                    } for name in m3_tool_names)
                    asked_user = "ask_user" in m3_tool_names
                    if browser_mutated and not getattr(ctx, "m3_final_snapshot_added", False):
                        try:
                            snapshot = self.runtime.dispatch(
                                "mcp_playwright_browser_snapshot", {}, ctx
                            )
                        except Exception as snap_err:
                            snapshot = f"ERROR: automatic browser snapshot failed: {snap_err}"
                        ctx.m3_final_snapshot_added = True
                        ctx.force_no_tools = True
                        # MiniMax-M3 can hang on OpenAI-style role="tool" history.
                        # For the final answer, compact the browser evidence into a
                        # clean user message and disable tools.
                        messages[:] = [
                            {"role": "system", "content": full_system_instruction},
                            {
                                "role": "user",
                                "content": (
                                    "원래 질문:\n"
                                    f"{prompt}\n\n"
                                    "자동 브라우저 스냅샷 결과입니다. 이 내용을 근거로 한국어로 "
                                    "간결하게 최종 답변하십시오. 도구를 다시 호출하지 마십시오.\n\n"
                                    f"{str(snapshot)[:6000]}"
                                ),
                            },
                        ]
                    elif asked_user and not getattr(ctx, "m3_ask_final_compacted", False):
                        ask_result = ""
                        for msg in reversed(messages):
                            if msg.get("role") == "tool" and msg.get("name") == "ask_user":
                                ask_result = str(msg.get("content") or "")
                                break
                        ctx.m3_ask_final_compacted = True
                        ctx.force_no_tools = True
                        messages[:] = [
                            {"role": "system", "content": full_system_instruction},
                            {
                                "role": "user",
                                "content": (
                                    "원래 질문:\n"
                                    f"{prompt}\n\n"
                                    "사용자가 선택지 카드에서 다음처럼 선택했습니다. "
                                    "이 선택을 반영하여 한국어로 간결하게 최종 답변하십시오. "
                                    "도구를 다시 호출하지 마십시오.\n\n"
                                    f"{ask_result[:2000]}"
                                ),
                            },
                        ]
                    continue

                full_content = ""
                tool_calls = []
                reasoning_buffer = ""
                emit_reasoning = _show_reasoning()
                _usage_acc = None  # 🪙 토큰 사용량(최종 청크)

                for line in res.iter_lines():
                    if not line:
                        continue
                    l_str = line.decode('utf-8')
                    if not l_str.startswith("data: "):
                        continue
                    data_raw = l_str[6:]
                    if data_raw.strip() == "[DONE]":
                        break
                    try:
                        chunk = json.loads(data_raw)
                        if chunk.get("usage"):
                            _usage_acc = chunk["usage"]
                        if not chunk.get("choices"):
                            continue
                        delta = chunk["choices"][0].get("delta", {})

                        content = delta.get("content", "")
                        reasoning = delta.get("reasoning_content", "")
                        
                        if reasoning:
                            # 🧠 [Reasoning 게이트] show_reasoning이 꺼져 있으면(기본) 추론
                            # 채널을 스트림·최종 답변에서 제외한다. 메타/지시 메아리 누출 차단.
                            if emit_reasoning:
                                full_content += reasoning
                                if first_yield:
                                    first_yield = False
                                if not ctx.suppress_stream:
                                    yield reasoning # 추론 과정도 실시간으로 출력
                            else:
                                reasoning_buffer += reasoning  # 폴백용으로만 보관(미출력)

                        if content:
                            full_content += content
                            if first_yield:
                                first_yield = False
                            if not ctx.suppress_stream:
                                yield content

                        tc_delta = delta.get("tool_calls", [])
                        for tc in tc_delta:
                            index = tc.get("index", 0)
                            while len(tool_calls) <= index:
                                tool_calls.append({"id": "", "type": "function", "function": {"name": "", "arguments": ""}})
                            if "id" in tc:
                                tool_calls[index]["id"] += tc["id"]
                            if "function" in tc:
                                f = tc["function"]
                                if "name" in f:
                                    tool_calls[index]["function"]["name"] += f["name"]
                                if "arguments" in f:
                                    tool_calls[index]["function"]["arguments"] += f["arguments"]
                    except Exception:
                        pass

                self._emit_token_usage(provider, model_name, _usage_acc)  # 🪙
                if not tool_calls:
                    # 🧠 [Reasoning 폴백] 추론 숨김 모드에서 content가 비고 추론만 온 경우
                    # 답변 유실 방지 — reasoning을 최종 답변으로 승격해 스트림에 1회 방출.
                    if not emit_reasoning and not full_content.strip() and reasoning_buffer.strip():
                        full_content = _pick_nonreasoning_final(full_content, reasoning_buffer)
                        if first_yield:
                            first_yield = False
                        if not ctx.suppress_stream and full_content:
                            yield full_content
                    # 🛡️ [Tool Runtime] 도구 없는 응답 — Empty-Final / Fake-Ask / Markdown-Ask 구제 일원화
                    _action = yield from self.runtime.handle_no_tool_calls(full_content, messages, ctx)
                    if _action == "continue":
                        continue
                    break

                # 🛡️ [Tool Runtime] 도구 호출 일괄 실행 (검증·보정·ask 가드·서킷브레이커 공통)
                yield from self.runtime.run_tool_calls(tool_calls, full_content, messages, ctx)

        except requests.exceptions.HTTPError as e:
            yield f"❌ [{display_name}] HTTP 오류: {e}"
        except Exception as e:
            logger.error(f"[{display_name}] 오류: {e}")
            yield f"❌ [{display_name}] 오류: {str(e)}"

    def _load_google_cli_creds(self) -> Optional[Dict[str, Any]]:
        """Loads OAuth credentials from the Gemini CLI storage."""
        home = os.path.expanduser("~")
        old_creds_path = os.path.join(home, ".gemini", "oauth_creds.json")
        
        if os.path.exists(old_creds_path):
            try:
                with open(old_creds_path, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception as e:
                logger.error(f"Failed to load Gemini CLI credentials: {e}")
        return None

    def _resolve_actual_model(self, provider: str, model_name: str) -> str:
        """[🔱 Imperial] 마왕님의 지시: DB에서 넘어온 값을 임의로 변환하지 않고 그대로 반환합니다."""
        target = model_name or self.primary_model
        if not target or target == "자동":
            return "gemini-2.5-flash-lite"
            
        # [🔱 OpenRouter Aliases]
        t_lower = target.lower()
        if t_lower in ["nemotron", "nano-omni", "nemotron-free"]:
            return "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free"

        return target

    def _gemini_manual_stream_loop(self, client, model_name, contents, callables,
                                   system_instruction, tool_config=None, max_iters=8):
        """🌊 Gemini 수동 함수호출 + 토큰 스트리밍 루프 (stream_tokens 옵트인 전용).

        AFC(자동 함수호출)는 비스트리밍이 강제라 응답이 한 덩어리로 나온다.
        여기서는 AFC를 끄고 스트림에서 텍스트 델타는 즉시 yield, function_call은
        수집해 직접 실행한 뒤 대화에 붙여 반복한다(자비스식 타자기 효과 + 도구 사용).
        예외는 호출부(think)가 잡아 기존 AFC 경로로 폴백한다.
        """
        tool_map = {}
        for fn in callables:
            if callable(fn) and hasattr(fn, "__name__"):
                tool_map[fn.__name__] = fn
        contents = list(contents)
        safety = [
            types.SafetySetting(category="HARM_CATEGORY_HATE_SPEECH", threshold="BLOCK_NONE"),
            types.SafetySetting(category="HARM_CATEGORY_HARASSMENT", threshold="BLOCK_NONE"),
            types.SafetySetting(category="HARM_CATEGORY_SEXUALLY_EXPLICIT", threshold="BLOCK_NONE"),
            types.SafetySetting(category="HARM_CATEGORY_DANGEROUS_CONTENT", threshold="BLOCK_NONE"),
            types.SafetySetting(category="HARM_CATEGORY_CIVIC_INTEGRITY", threshold="BLOCK_NONE"),
        ]
        for it in range(max_iters):
            cfg_args = dict(
                system_instruction=system_instruction,
                tools=callables,
                temperature=0.0,
                automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True),
                safety_settings=safety,
            )
            # Task Mode의 도구 강제는 첫 턴에만 — 매 턴 강제하면 최종 답변을 못 낸다
            if tool_config is not None and it == 0:
                cfg_args["tool_config"] = tool_config
            stream = client.models.generate_content_stream(
                model=model_name, contents=contents,
                config=types.GenerateContentConfig(**cfg_args),
            )
            fcalls = []
            text_parts = []
            usage = None
            for chunk in stream:
                um = getattr(chunk, "usage_metadata", None)
                if um is not None:
                    usage = um
                for cand in (getattr(chunk, "candidates", None) or [])[:1]:
                    content = getattr(cand, "content", None)
                    for part in (getattr(content, "parts", None) or []):
                        fc = getattr(part, "function_call", None)
                        if fc is not None and getattr(fc, "name", None):
                            fcalls.append(fc)
                        pt = getattr(part, "text", None)
                        if pt:
                            text_parts.append(pt)
                            yield pt  # 🌊 토큰 즉시 방출
            if usage is not None:
                self._emit_token_usage("google", model_name, usage)  # 🪙
            if not fcalls:
                return  # 최종 답변 완료
            # 모델 턴(텍스트+함수호출) 재구성 후 도구 실행 결과를 붙여 다음 턴으로
            model_parts = []
            if text_parts:
                model_parts.append(types.Part(text="".join(text_parts)))
            model_parts.extend(types.Part(function_call=fc) for fc in fcalls)
            contents.append(types.Content(role="model", parts=model_parts))
            resp_parts = []
            for fc in fcalls:
                name = getattr(fc, "name", "") or ""
                args = dict(getattr(fc, "args", None) or {})
                logger.info(f"🌊 [TokenStream] 도구 실행: {name}({list(args.keys())})")
                fn = tool_map.get(name)
                if fn is None:
                    result = f"오류: '{name}' 도구를 찾을 수 없습니다."
                else:
                    try:
                        result = fn(**args)
                    except Exception as tool_err:
                        result = f"도구 실행 오류: {tool_err}"
                if isinstance(result, dict):
                    payload = result
                else:
                    payload = {"result": str(result)[:20000]}
                resp_parts.append(types.Part.from_function_response(name=name, response=payload))
                yield f"[STAGE:{name}]"  # 진행 마커 — 표시단(정제)에서 제거됨
            contents.append(types.Content(role="tool", parts=resp_parts))
        yield "\n⚠️ 도구 반복 상한(8회)에 도달해 여기까지의 내용으로 마무리합니다."

    def think(self, prompt: str, history: list = None, model_name: str = None, images: list = None, system_instruction: str = None,
              skip_fallback: bool = False, provider: str = None, skip_tools: bool = False,
              tools: list = None, session_id: str = None, token: str = None, aegis_url: str = None,
              override_config: dict = None, is_master: bool = False,
              interactive: bool = True, skip_grounding: bool = False,
              stream_tokens: bool = False,
              coding_intent: bool = None) -> Generator[str, None, None]:
        """
        Autonomous thinking loop with Tool Use (Function Calling).
        Supports multi-modal input if images are provided.
        Accepts system_instruction for persona injection.
        Tries provided model, and auto-falls back to configured secondary on failure.
        """
        # [🧠 정직 게이트] 도구(recall_memory/count_history)가 호출 시점에 쓸 요청 컨텍스트 저장
        self._request_ctx.token = token
        self._request_ctx.aegis_url = aegis_url
        self._request_ctx.session_id = session_id

        _lap_t = [time.time()]

        def _lap(name):  # ⏱️ think() 전처리 프로파일 (TTFT 추적용)
            now = time.time()
            if now - _lap_t[0] > 0.15:  # 150ms 이상 걸린 단계만 로그
                logger.info(f"⏱️ [think-prep] {name}: {(now - _lap_t[0])*1000:.0f}ms")
            _lap_t[0] = now

        # [🎖️ Guardian Job Auto-Routing] 직업 설정이 "auto"인 경우 prompt 분석을 통해 적절한 직업 자동 매핑
        job_setting = self.dynamic_config.get("job")
        _lap("pre-job")
        from guardian_jobs import _all_jobs as _available_jobs
        if job_setting == "auto" and _available_jobs(aegis_url):
            classified_job = self._classify_job_by_prompt(prompt)
            logger.info(f"[🧠 Brain] Prompt classified auto-job: {classified_job} for prompt: {prompt[:35]}...")
            try:
                from guardian_jobs import (
                    apply_job_to_capabilities,
                    get_job as _get_auto_job,
                    resolve_effective_tools_skills as _resolve_auto_ets,
                )
                temp_caps = {
                    "job": "auto",
                    "max_iterations": self.dynamic_config.get("max_iterations", 10),
                }
                temp_caps = apply_job_to_capabilities(temp_caps, classified_job, aegis_url=aegis_url)

                # 🎮 [앱 세트] apply_job_to_capabilities 는 toggle/rule 만 병합한다.
                # 도구/스킬 화이트리스트는 여기서 직업 정의를 직접 읽어 장착 앱(equipped_apps)
                # 까지 도구/스킬로 펼쳐 반영한다(정적 경로 core.py:~583 과 동일 동작).
                _auto_job = _get_auto_job(classified_job, aegis_url=aegis_url) or {}
                _auto_tools, _auto_skills = _resolve_auto_ets(_auto_job)
                self.dynamic_config["allowed_tools"] = _auto_tools
                self.dynamic_config["tool_mode"] = _auto_job.get("tool_mode", "all")
                self.dynamic_config["equipped_skills"] = _auto_skills
                self.dynamic_config["obsidian_job_path"] = temp_caps.get("obsidian_job_path", "")
                self.dynamic_config["job_source"] = temp_caps.get("job_source", "cloud")
                if temp_caps.get("num_ctx"):
                    self.dynamic_config["num_ctx"] = temp_caps.get("num_ctx")
                if temp_caps.get("num_predict"):
                    self.dynamic_config["num_predict"] = temp_caps.get("num_predict")
                
                # job_rule이 주입되었다면 system_instruction에 병합하거나 덮어씁니다.
                job_rule = temp_caps.get("job_rule", "")
                if job_rule:
                    if system_instruction:
                        system_instruction = f"{system_instruction}\n\n[직업 행동 지침: {classified_job}]\n{job_rule}"
                    else:
                        system_instruction = f"[직업 행동 지침: {classified_job}]\n{job_rule}"
            except Exception as _job_err:
                logger.warning(f"[🧠 Brain] 자동 직업 라우팅 적용 실패: {_job_err}")

        _lap("job-routing")
        # [🛡️] Dynamic Configuration Refresh: 로컬 에이전트 모드인 경우 매 요청마다 Aegis에서 최신 설정을 가져옵니다.
        if os.environ.get("LOCAL_AGENT_MODE", "false").lower() == "true":
            try:
                new_config = self.get_dynamic_config(token=token, override_config=override_config, aegis_url=aegis_url)
                if new_config:
                    self.dynamic_config = new_config
                    logger.info("[🧠 Brain] Dynamic config refreshed via delegated token.")
            except Exception as e:
                logger.warning(f"[🧠 Brain] Failed to refresh dynamic config: {e}")

        _lap("dyn-config")
        # 🎖️ [Ollama 정수 설정 강제 반영] override_config(persona_settings)가 넘긴
        # num_ctx/num_predict는 LOCAL_AGENT_MODE 여부와 무관하게 항상 dynamic_config에
        # 반영한다. (직통채팅은 LOCAL_AGENT_MODE가 꺼져 있어도 직업의 응답 토큰 상한이
        # _think_ollama까지 도달해야 하므로 — 안 그러면 num_predict가 ∞로 무시됨.)
        if override_config:
            _merge = {}
            for _k in ("num_ctx", "num_predict", "job_source",
                       "obsidian_job_path", "obsidian_context"):
                if override_config.get(_k) is not None:
                    _merge[_k] = override_config[_k]
            if _merge:
                self.dynamic_config = {**(self.dynamic_config or {}), **_merge}

        # 🧭 [Imperial Obsidian Context Profile]
        # CLOUD는 기존 Aegis 직업, LOCAL(vault)은 Vault 직업·스킬·지식을 사용한다.
        try:
            from imperial_obsidian_context import build_context as _build_obsidian_context
            _obs_ctx = _build_obsidian_context(self.dynamic_config or {}, prompt)
            if _obs_ctx.enabled:
                system_instruction = (system_instruction or "") + _obs_ctx.system_context
                if _obs_ctx.mode == "LOCAL":
                    self.dynamic_config["job_rule"] = _obs_ctx.job_rule
                    self.dynamic_config["equipped_skills"] = _obs_ctx.equipped_skills
                    self.dynamic_config["allowed_tools"] = _obs_ctx.allowed_tools
                    self.dynamic_config["tool_mode"] = "whitelist"
                logger.info(
                    "🧭 [Obsidian Context] mode=%s policy=%s sources=%s",
                    _obs_ctx.mode, _obs_ctx.policy_path, len(_obs_ctx.sources),
                )
            elif _obs_ctx.warnings:
                logger.warning("[Obsidian Context] %s", " | ".join(_obs_ctx.warnings))
        except Exception as _obs_ctx_err:
            logger.warning("[Obsidian Context] 안전 폴백(SYSTEM): %s", _obs_ctx_err)

        if provider is None:
            provider = self.primary_provider
            # [🔱 Auto-Provider Routing] 모델명에 특정 키워드가 있으면 프로바이더 자동 전환
            m_low = (model_name or "").lower()
            if "minimaxai/" in m_low:
                provider = "nvidia"
                logger.info(f"[🧠 Brain] Auto-routed to NVIDIA based on model: {model_name}")
            elif any(k in m_low for k in ["nemotron", "nvidia", "qwen", "meta/", "mistral/", "anthropic/"]):
                provider = "openrouter"
                logger.info(f"[🧠 Brain] Auto-routed to OpenRouter based on model: {model_name}")
        provider = str(provider or "").lower()

        # [🔱 Model-Namespace Override] 모델 접두사가 공급자를 명확히 지시하는데 provider가 어긋나면
        # 강제 교정한다. (헤임달이 NVIDIA NIM을 골랐어도 provider를 nvidia로 안 보내고 google로 보내는
        # 불일치 → minimaxai/* 모델이 Gemini 엔드포인트로 가 404 나던 문제 차단.)
        # minimax/glm 네임스페이스는 해당 공급자 전용이라 안전하게 강제 가능.
        _m_ns = (model_name or "").lower()
        _forced = None
        if "minimaxai/" in _m_ns:
            _forced = "nvidia"
        elif "zai-org/" in _m_ns or _m_ns.startswith("glm"):
            _forced = "glm"
        if _forced and provider != _forced:
            logger.info(f"[🧠 Brain] 모델 네임스페이스로 provider 교정: '{provider}' → '{_forced}' (model={model_name})")
            provider = _forced

        _lap("obsidian-ctx")
        # 🛠️🌐 [Imperial Universal Gate] 코딩/검색 지시 주입을 모든 프로바이더에 적용.
        # (이 분기 아래의 OpenAI 호환 경로들은 Gemini 경로의 주입 로직을 거치지 않으므로
        #  여기서 먼저 주입한다 — "대통령 누구지"에 모델 기억으로 답하던 문제의 근본 차단)
        # 🙋 [Ask 오염 방지] 과거 응답에 저장된 imperial-ask 블록이 히스토리/프롬프트로 재주입되면
        # 모델이 ask_user 도구 호출 대신 블록 텍스트를 흉내 낸다(가짜 ask_id → UI 미표시, 가짜 대기).
        # 모델이 도구 호출 대신 인자 JSON({"question":...,"options":...})을 본문에 쓴 실패 사례도
        # 히스토리에 남으면 같은 흉내를 반복하므로 함께 제거한다.
        _ask_block_re = re.compile(
            r"```imperial-ask[\s\S]*?```"
            r"|\{[^{}]*\"ask_id\"[^{}]*\}"
            r"|\{[^{}]*\"question\"[^{}]*\"options\"[^{}]*\}"
        )
        _ask_taint = ('imperial-ask', '"ask_id"', '"question"')
        if prompt and any(t in prompt for t in _ask_taint):
            prompt = _ask_block_re.sub("[선택지 질문 — 완료됨]", prompt)
        if history and isinstance(history, list):
            for _hm in history:
                _hc = _hm.get("content") if isinstance(_hm, dict) else None
                if _hc and any(t in str(_hc) for t in _ask_taint):
                    _hm["content"] = _ask_block_re.sub("[선택지 질문 — 완료됨]", str(_hc))

        # [🔱 Performance Bypass] If use_history is disabled, bypass history context completely
        _use_hist_val = (self.dynamic_config or {}).get("use_history")
        if _use_hist_val is False:
            logger.info("[🧠 Brain] use_history is False — bypassing all conversation history context.")
            history = []

        # inject_guidelines가 꺼져 있으면 동적 시스템 지침 주입을 생략한다.
        # to maximize Ollama Prompt Cache hit rate and avoid GPU/CPU overhead.
        # Fallback to checking memory_recall if inject_guidelines is not explicitly set.
        _inj_val = (self.dynamic_config or {}).get("inject_guidelines")
        if _inj_val is not None:
            is_no_job = bool(_inj_val) is False
        else:
            is_no_job = (self.dynamic_config or {}).get("memory_recall", True) is False

        # 🧠 [정직 게이트 — 기억 저장 환각 차단] 모델에게는 저장 도구가 없다(저장은 백그라운드 자동).
        # "등록/보관/저장했다" 단정 발화를 상시 금지하고, 저장소 장애 시 장애 사실을 고지한다.
        if not is_no_job and '기억 정직 수칙' not in (system_instruction or ''):
            memory_honesty = (
                "\n\n## 🧠 기억 정직 수칙\n"
                "너에게는 기억을 직접 저장하는 도구가 없다. 대화 기억 저장은 백그라운드에서 "
                "자동으로 시도되며 실패할 수도 있다. 따라서 절대 '저장했다/등록했다/보관했다/"
                "영구히 기록했다'고 단정하지 마라. 사용자가 기억해 달라고 하면 "
                "'기억해 두도록 하겠습니다' 수준으로만 답하고, 실제 기억 여부는 "
                "recall_memory 검색 결과로만 판단하라."
            )
            health_notice = sovereign_memory.memory_health_notice()
            if health_notice:
                memory_honesty += (
                    f"\n⚠️ {health_notice} "
                    "이 상태에서는 기억 약속 자체를 하지 말고, 기억 기능 장애를 솔직하게 알려라."
                )
            system_instruction = (system_instruction or "") + memory_honesty

        # 🗺️ [정직 게이트 — 접근 범위 고지] 파일 검색 범위가 화이트리스트로 제한됨을 모델에 알려
        # "범위 밖이라 확인 불가"와 "존재하지 않음"을 구분해 답하게 한다. (잘못된 폴더 기준 추측 차단)
        if not is_no_job and '접근 범위 수칙' not in (system_instruction or ''):
            scope_note = (
                "\n\n## 🗺️ 접근 범위 수칙\n"
                "너의 파일 검색 도구는 허용된 폴더(화이트리스트) 안만 볼 수 있다."
            )
            allowed_dirs = self._get_mcp_allowed_dirs()
            if allowed_dirs:
                scope_note += f" 현재 허용 범위: {', '.join(allowed_dirs)}."
            scope_note += (
                "\n규칙: 시스템 구성/파일 위치 질문에는 파일 검색 전에 반드시 search_knowledge로 "
                "지식 베이스를 먼저 조회하라. 허용 범위 안에서 못 찾았다면 '존재하지 않는다'가 아니라 "
                "'내 접근 범위 밖일 수 있어 확인 불가'라고 답하라. 범위 안에서 보이는 파일이 "
                "구버전/백업일 가능성도 항상 고려하고, 부분 검색 결과로 전체 시스템 구조를 단정하지 마라. "
                "새로운 사실을 확인하면 save_knowledge로 즉시 기록하라 "
                "(이 PC에서만 참인 사실은 scope='environment', 어디서나 참인 사실은 scope='global')."
                "\n신뢰 등급 수칙: 답변 시 근거 등급을 구분하라 — "
                "✅방금 check_port/find_process 등으로 이 환경에서 직접 관측한 사실은 단정 가능, "
                "📚지식 베이스 기록은 '기록상 ~이다(확인일)'로 전달, 🤔추론은 반드시 추정임을 밝혀라. "
                "환경 의존적 사실(포트, 실행 상태, 경로)은 가능하면 관측 도구로 재검증한 뒤 답하라."
            )
            system_instruction = (system_instruction or "") + scope_note

        # 🙋 [선택지 질문 수칙] 대화형 경로에서는 유저에게 고르게 할 질문이 생기면
        # 텍스트 목록 나열 대신 ask_user 도구를 능동적으로 호출하게 한다.
        # (비대화형 릴레이(interactive=False)는 선택지 UI가 뜰 수 없으므로 주입하지 않음)
        if not is_no_job and interactive and '선택지 질문 수칙' not in (system_instruction or ''):
            ask_note = (
                "\n\n## 🙋 선택지 질문 수칙\n"
                "답변 중 유저에게 둘 이상의 선택지 가운데 하나를 고르게 하려는 경우"
                "(A/B/C, ①②③ 번호 목록, '어느 쪽으로 할까요?' 등), 선택지를 텍스트로 "
                "나열하고 턴을 끝내지 마라. 반드시 ask_user 도구를 호출해 버튼 UI로 질문하라. "
                "유저가 ask_user를 쓰라고 명시하지 않아도, 선택이 필요한 상황이면 스스로 "
                "ask_user를 사용하는 것이 기본 동작이다. "
                "예외: 선택지가 7개 이상이거나, 자유 서술형 답변이 필요하거나, 단순 수사적 "
                "질문일 때는 텍스트로 물어라. 도구 호출 직전/직후에 같은 선택지를 텍스트로 "
                "중복 나열하지 마라. "
                "경고: ask_user는 반드시 함수 호출(tool call)로만 실행해야 한다. "
                "본문에 {\"question\": ..., \"options\": ...} 형태의 JSON이나 imperial-ask "
                "블록을 직접 출력하는 것은 도구 호출이 아니며, 아무 UI도 뜨지 않고 유저의 "
                "답도 영원히 오지 않는다. JSON을 쓰고 있다면 즉시 멈추고 도구를 호출하라."
            )
            system_instruction = (system_instruction or "") + ask_note

        # 🗺️ [Place Card] 장소/맛집 추천 시 주소 포함 수칙 동적 주입
        _place_keywords = ('맛집', '식당', '카페', '음식점', '추천', '여행', '관광', '호텔', '숙소',
                           'restaurant', 'cafe', 'travel', 'hotel', 'recommend', '명소', '볼거리',
                           '놀거리', '가볼만한', '어디', '어딜', '투어', '코스', '스팟')
        if not is_no_job and any(kw in prompt for kw in _place_keywords) and '장소 주소 수칙' not in (system_instruction or ''):
            place_note = (
                "\n\n## 🗺️ 장소 주소 수칙 (필수 — 위반 시 지도 카드 미표시)\n"
                "장소, 맛집, 카페, 명소, 호텔, 관광지를 추천하거나 언급할 때는 "
                "반드시 **정확한 한국 도로명 주소**를 장소명 바로 뒤 괄호 안에 명시하라.\n"
                "필수 형식 예시:\n"
                "- 에버랜드 (경기도 용인시 처인구 포곡읍 에버랜드로 199)\n"
                "- 한국민속촌 (경기도 용인시 기흥구 민속촌로 90)\n"
                "- 화담숲 (경기도 광주시 도척면 도척윗로 278)\n"
                "주소 형식: '시/도 + 시/군/구 + 읍/면/동 + 로/길 + 번지' 순서로 작성.\n"
                "주소가 없으면 UI 지도·사진·후기 카드가 절대 표시되지 않는다.\n"
                "주소를 URL로 변환하거나 링크로 감싸지 말고 순수 텍스트로만 작성하라.\n"
                "번지 숫자를 반드시 포함하라 (예: '199', '90', '278').\n"
                "추천 장소가 여러 개라면 각 장소마다 개별 주소를 명시하라."
            )
            system_instruction = (system_instruction or "") + place_note

        if not is_no_job and self._wants_obsidian(prompt) and 'Imperial Obsidian Directive' not in (system_instruction or ''):
            system_instruction = (system_instruction or "") + _OBSIDIAN_DIRECTIVE
            logger.info("🗂️ [IMPERIAL OBSIDIAN MODE] 옵시디언 의도 감지 — 노트 작성/검색 지시 주입.")
        elif (not is_no_job
              # 🛠️ [Coding Intent Gate] 호출부가 코딩 여부를 명시(coding_intent)하면 키워드
              # 추측(_wants_coding)을 쓰지 않는다 — cwd 없는 잡담("테스트야" 등)이 코딩
              # 모드로 빨려가 code_* 재촉 루프에 4~8초를 태우던 TTFT 낭비 차단(2026-07-10 실측).
              and (self._wants_coding(prompt) if coding_intent is None else coding_intent)
              and 'Imperial Coding Protocol' not in (system_instruction or '')):
            system_instruction = (system_instruction or "") + CODING_AGENT_GUIDE
            logger.info("🛠️ [IMPERIAL CODING MODE] 코딩 의도 감지 — 에이전틱 코딩 수칙 주입 (전 프로바이더).")
        elif not is_no_job and self._wants_media(prompt):
            # Deterministic media routing:
            # if the user explicitly asks for images/photos/GIF/video/3D, do not
            # ask them to choose a search engine. The media noun already decides it.
            system_instruction = (system_instruction or "") + (
                "\n\n## Imperial Media Direct Routing\n"
                "The user explicitly asked for media. Do NOT ask whether to use web search.\n"
                "Routing rules (apply in order):\n"
                "1) If the user NAMED a specific source, go there directly (do NOT call ask_user):\n"
                "   - 유튜브/youtube -> call search_youtube(query, limit).\n"
                "   - Pexels -> call search_media(query, media_type='video', provider='pexels').\n"
                "   - Pixabay -> call search_media(query, media_type='video', provider='pixabay').\n"
                "2) If the request is a VIDEO/동영상/영상 search but NO source was named, you MUST FIRST call "
                "ask_user(question='어디서 영상을 찾을까요?', options=['Pexels','Pixabay','유튜브']) and let the "
                "routing handle the choice. Do NOT pick a source yourself in this case.\n"
                "3) Otherwise choose directly from the request (no ask_user):\n"
                "   - 3D, 3D model, GLB, asset -> call search_3d_model(query, limit).\n"
                "   - image, photo, picture, wallpaper -> call search_media(query, media_type='photo').\n"
                "   - GIF, gif, 움짤, 짤 -> call search_media(query, media_type='gif').\n"
                "If the user gives a count such as 2개/2장/two, pass it as limit when possible. "
                "After the tool result/card is shown, answer briefly that the requested media was found."
            )
            logger.info("[IMPERIAL MEDIA DIRECT] Media request detected -> direct media tool routing.")
        elif not is_no_job and self._wants_realtime_search(prompt):
            # [🔎 검색 엔진 선택 메뉴] 검색 요청이면 자동으로 엔진을 고르지 말고, ask_user로 먼저 물어본다.
            #   - 네이버 웹검색  → web_search (네이버 headless Playwright: 시사·인물·뉴스·날씨·시세 등)
            #   - 이지스 미디어검색 → search_media / search_3d_model (Pexels·Pixabay 이미지/GIF/동영상/3D)
            system_instruction = (system_instruction or "") + (
                "\n\n## 🔎 검색 엔진 선택 수칙 (반드시 준수)\n"
                "이 요청은 검색이다. 절대 바로 답하거나 임의 도구로 검색하지 말고, "
                "**가장 먼저 ask_user 도구를 호출**해 사용자에게 검색 방식을 물어라.\n"
                "ask_user 인자: question='어디서 검색할까요?', "
                "options=['유튜브', '네이버', '다음', '구글', '이지스']\n"
                "사용자 선택 후 라우팅:\n"
                "- '유튜브' 선택 → youtube_search(query) 호출.\n"
                "- '네이버' 선택 → web_search(query, engine='naver') 호출.\n"
                "- '다음' 선택 → web_search(query, engine='daum') 호출.\n"
                "- '구글' 선택 → web_search(query, engine='google') 호출.\n"
                "- '이지스' 선택 → 원문에서 종류 추론: 3D/GLB→search_3d_model, 이미지/사진→search_media(media_type='photo'), "
                "영상→search_media(media_type='video'), GIF/움짤→search_media(media_type='gif'). 여러 종류면 각각 호출.\n"
                "단, 원문에 '유튜브'/'youtube'가 명시돼 있으면 ask_user 없이 곧바로 youtube_search(query)를 호출하라.\n"
                "도구 결과(카드/URL)를 그대로 제시하고 사이트 추천으로 대체하지 마라."
            )
            logger.info("🔎 [IMPERIAL SEARCH MENU] 검색 요청 감지 — ask_user 엔진 선택 메뉴 수칙 주입.")

        _lap("inject-rules")
        if provider in ["github", "copilot"]:
            yield from self._think_copilot(prompt, model_name, system_instruction=system_instruction, skip_tools=skip_tools, tools=tools)
            return

        if provider == "atlas":
            yield from self._think_atlas(prompt, model_name, system_instruction=system_instruction, skip_tools=skip_tools, tools=tools)
            return

        if provider == "ollama":
            yield from self._think_ollama(
                prompt,
                model_name,
                history=history,
                system_instruction=system_instruction,
                images=images,
                token=token,
                aegis_url=aegis_url,
                skip_tools=skip_tools,
                tools=tools,
                skip_grounding=skip_grounding,
                stream_tokens=stream_tokens,
            )
            return

        if provider in ["openrouter", "openclaw", "relay", "nous", "hermes_proxy", "nvidia", "glm"]:
            yield from self._think_openrouter(prompt, model_name, system_instruction=system_instruction, skip_tools=skip_tools, tools=tools, session_id=session_id, provider=provider, interactive=interactive, images=images)
            return

        # [🔱 V7.0 Stateless Client Resolution]
        client = self._get_client(aegis_url=aegis_url, token=token)
        
        if not client:
            if skip_fallback:
                yield "❌ Critical Brain Error: Neural core offline (API Key 수혈 실패)."
                return
            yield from self._think_fallback(prompt, system_instruction=system_instruction, skip_tools=skip_tools, tools=tools)
            return

        try:
            actual_model_name = self._resolve_actual_model(provider, model_name)
            
            if any(x in actual_model_name.lower() for x in ["risk", "audit", "security", "auditor"]):
                logger.info(f"Imperial Security Protocol Engaged for: {actual_model_name}")
                prompt = (
                    "[MISSION: IMPERIAL SECURITY AUDIT / RISK ASSESSMENT]\n"
                    "당신은 제국의 보안 수호자로서 해당 요청의 잠재적 위험을 분석하고 보안 취약점을 식별해야 합니다.\n\n"
                ) + prompt
                
                if "auditor" in actual_model_name.lower():
                    actual_model_name = "gemini-1.5-pro"
            
            logger.info(f"Targeting Brain Model: {actual_model_name}")
            
            # 🌐 검색 발동 조건은 Aegis common_retrieval_policy만 사용한다.
            # 🛠️ [Imperial Coding Gate] 코딩 의도가 감지되면 검색 모드보다 우선한다.
            # ("버그 찾아줘" 같은 요청이 '찾아' 키워드로 검색 그라운딩에 빨려가던 문제 차단)
            # 🗂️ 옵시디언 의도는 코딩 모드보다 우선 (파일/터미널 도구 범람 차단)
            is_obsidian = self._wants_obsidian(prompt)
            is_coding = self._wants_coding(prompt) and not is_obsidian
            # 🖼️ 미디어 검색 의도는 검색 그라운딩보다 우선 — search_media 도구로 처리해야 함
            is_media = self._wants_media(prompt)
            is_pure_search = (
                not images
                and not is_coding
                and not is_media
                and self._should_use_google_grounding(prompt, aegis_url=aegis_url)
            )
            if is_obsidian and 'Imperial Obsidian Directive' not in (system_instruction or ''):
                system_instruction = (system_instruction or "") + _OBSIDIAN_DIRECTIVE
                logger.info("🗂️ [IMPERIAL OBSIDIAN MODE] 옵시디언 의도 감지 — 노트 작성/검색 지시 주입.")
            elif is_coding and 'Imperial Coding Protocol' not in (system_instruction or ''):
                system_instruction = (system_instruction or "") + CODING_AGENT_GUIDE
                logger.info("🛠️ [IMPERIAL CODING MODE] 코딩 의도 감지 — 에이전틱 코딩 수칙 주입.")

            if is_pure_search:
                logger.info("🌐 [IMPERIAL TURBO SEARCH] 검색 질문 감지 — 구글 검색 그라운딩 모드 진입.")
                READABILITY_GUIDE = (
                    "\n\n## 🔱 [제국 가독성 표준 / Imperial Readability Standard]\n"
                    "1. 핵심 주제나 단락을 나눌 때 관련성 높은 **이모지(Emoji)**를 제목 앞에 반드시 활용하여 시각적 직관성을 향상시키십시오.\n"
                    "2. 정보 요약 시, 단순 줄글보다는 굵은 글씨(**강조**)를 사용해 핵심 키워드를 한눈에 파악할 수 있도록 구조화하십시오.\n"
                    "3. 항목 구분 시 명확한 목록 기호(`*` 또는 `-`)를 활용하고, 단락 사이에 적절한 빈 줄(줄바꿈)을 넣어 가독성을 극대화하십시오.\n"
                    "4. 주요 경제 지표나 수치는 별도의 섹션으로 정리하고 보기 좋게 정돈하십시오.\n"
                )
                system_instruction = (system_instruction or "") + READABILITY_GUIDE
            else:
                logger.info("🔱 [IMPERIAL MCP MODE] 비검색 질문 감지 — MCP 커스텀 도구 세트 주입.")

            genai_tools = []
            
            # [🔱 Imperial Fix] 매 요청마다 최신 MCP 도구 목록을 다시 동기화하여 지연 로딩된 도구도 포함되게 합니다.
            if get_mcp_client:
                try:
                    mcp_client = get_mcp_client()
                    if mcp_client:
                        mcp_callables = mcp_client.get_gemini_callables()
                        if mcp_callables:
                            existing_names = {t.__name__ for t in self._tools if hasattr(t, '__name__')}
                            for t in mcp_callables:
                                if t.__name__ not in existing_names:
                                    self._tools.append(t)
                                    logger.info(f"[🔱 Imperial Core] 동적 MCP 도구 주입됨: {t.__name__}")
                except Exception as e:
                    logger.warning(f"[MCP] 동적 도구 주입 실패: {e}")

            actual_tools = tools or self._tools
            if not skip_tools and actual_tools:
                genai_tools = list(actual_tools)
            else:
                logger.info("Skipping AFC tools for this thinking session.")
                genai_tools = []

            # 🌐 Prepare Unified Tools (MCP + Custom 통합)
            # [🔱 Imperial Fix v2.7] Hugin(브라우저 조작) 도구는 기본적으로 도구 세트에서 제외한다.
            # 마왕님께서 명시적으로 브라우저 제어/스크린샷을 요청하실 때만(_wants_hugin) 포함한다.
            # 그동안 모델이 파일 수정/이미지 분석 요청에서도 습관적으로 hugin_screenshot을 호출해
            # 첨부 이미지를 무시하고 'Hugin 연결 오류' 환각을 일으키던 문제를 근본 차단하기 위함.
            _HUGIN_TOOL_NAMES = {"hugin_navigate", "hugin_screenshot", "hugin_click", "hugin_type"}
            # [🔱 Imperial Fix v2.7] 팀 위임/히드라 디스패치 도구도 기본 제외.
            # 파일 만들기 같은 일반 작업에도 모델이 습관적으로 delegate_team_task(팀 테스크)/
            # dispatch_to_hydra(히드라)를 호출해 실패하며 화면을 어지럽히던 문제를 막는다.
            # 마왕님이 '팀 테스크/위임/히드라'를 명시적으로 요청할 때만 노출한다.
            _DELEGATION_TOOL_NAMES = {"delegate_team_task", "dispatch_to_hydra"}
            # [🔱 Imperial Fix v2.7] OS 앱 실행/키입력 도구도 기본 제외.
            # 파일 만들기에 notepad/cmd를 실행하려다 실패하며 화면을 어지럽히던 문제 차단.
            _OS_CONTROL_TOOL_NAMES = {"sovereign_run_app", "sovereign_type_text", "sovereign_press_key"}
            include_hugin = self._wants_hugin(prompt)
            include_delegation = self._wants_delegation(prompt)
            include_os_control = self._wants_os_control(prompt)

            # 🎖️ [Guardian Job M1] 직업 기반 도구 화이트리스트.
            # 이 수호자의 capabilities.tool_mode == 'whitelist' 이면, allowed_tools에
            # 매칭되는 도구만 노출한다(직업 특화). 'all'/미설정이면 기존 게이트 동작 유지.
            _cfg = getattr(self, "dynamic_config", None) or {}
            _tool_mode = (_cfg.get("tool_mode") or "all").lower()
            _allowed_tools = _cfg.get("allowed_tools") or []
            _use_whitelist = (_tool_mode == "whitelist") and bool(_allowed_tools)
            # 화이트리스트 모드라도 항상 보장하는 베이스 공용 도구 (일을 아예 못 하게 되는 것 방지)
            _BASE_TOOLS = {"jarvis_memory", "recall_memory", "count_history", "search_knowledge", "save_knowledge", "check_port", "find_process", "ask_user"}

            # 🔱 [통일] MCP 큐레이션 플래그 (OpenAI 경로와 동일 규칙)
            _mcp_flags = {
                "hugin": include_hugin,
                "search": self._wants_web_search(prompt),
                "coding": self._wants_coding(prompt),
            }
            unified_function_declarations = []
            for t in self._tools:
                _tn = getattr(t, "__name__", "")
                # 🔱 [통일] MCP 도구: 네이티브 중복 제거 + 무거운 그룹 의도 게이트 (모든 경로 공통)
                if _tn.startswith("mcp_") and not self._mcp_name_kept(_tn, _mcp_flags):
                    continue
                if _use_whitelist:
                    # 직업 화이트리스트: 허용목록 매칭 or 베이스도구만 통과
                    if _tn not in _BASE_TOOLS and not self._tool_allowed(_tn, _allowed_tools):
                        continue
                else:
                    # 기존 안전 게이트 (Hugin / 위임 / OS제어는 명시 요청 시에만)
                    if not include_hugin and _tn in _HUGIN_TOOL_NAMES:
                        continue
                    if not include_delegation and _tn in _DELEGATION_TOOL_NAMES:
                        continue
                    if not include_os_control and _tn in _OS_CONTROL_TOOL_NAMES:
                        continue
                unified_function_declarations.append(t)

            if _use_whitelist:
                _exposed = [getattr(t, "__name__", "?") for t in unified_function_declarations]
                logger.info(f"🎖️ [Guardian Job] 화이트리스트 모드 — 노출 도구 {len(_exposed)}개: {_exposed}")
            else:
                if include_hugin:
                    logger.info("🔱 [Imperial] 명시적 브라우저 제어 요청 감지 — Hugin 도구를 도구 세트에 포함했습니다.")
                else:
                    logger.info("🔱 [Imperial] Hugin 도구는 기본 제외 상태 (명시적 브라우저 요청 없음).")
                if include_delegation:
                    logger.info("🔱 [Imperial] 명시적 팀 위임/히드라 요청 감지 — 위임 도구를 도구 세트에 포함했습니다.")
                else:
                    logger.info("🔱 [Imperial] 팀 위임/히드라 도구는 기본 제외 상태 (명시적 위임 요청 없음).")
                if include_os_control:
                    logger.info("🔱 [Imperial] 명시적 OS 앱 실행/키입력 요청 감지 — OS 제어 도구를 포함했습니다.")
                else:
                    logger.info("🔱 [Imperial] OS 제어(notepad/cmd 등) 도구는 기본 제외 상태 (명시적 실행 요청 없음).")

            # 🎖️ [Guardian Job] 직업 기본 룰 적용.
            # 수호자 개별 룰(relay_system_prompt)은 이미 system_instruction에 반영돼 있다.
            # 개별 룰이 (거의) 비어 있을 때만, 직업의 기본 룰(job_rule)을 시스템 지시에 덧붙인다.
            # → "직업 하나 고르면 룰까지 자동 전환" + "개별 룰 쓰면 그게 우선" 둘 다 만족.
            try:
                _job_rule = (_cfg.get("job_rule") or "").strip()
                if _job_rule:
                    _has_custom_rule = bool(system_instruction and len(system_instruction.strip()) > 80)
                    if not _has_custom_rule:
                        system_instruction = (system_instruction or "") + f"\n\n## 🎖️ 직업 지침\n{_job_rule}"
                        logger.info("🎖️ [Guardian Job] 직업 기본 룰을 시스템 지시에 적용했습니다.")
                    else:
                        logger.info("🎖️ [Guardian Job] 수호자 개별 룰이 있어 직업 기본 룰은 생략합니다.")
            except Exception as _jr_err:
                logger.warning(f"[Guardian Job] 직업 룰 적용 실패(무시): {_jr_err}")

            # 🎖️ [Guardian Job M2] 장착 스킬(전문 지식)을 시스템 지시에 주입.
            # capabilities.equipped_skills 에 적힌 스킬 문서를 읽어 system_instruction 뒤에 덧붙인다.
            try:
                _equipped = _cfg.get("equipped_skills") or []
                if _equipped:
                    _skill_text = self._load_equipped_skills(_equipped)
                    if _skill_text:
                        system_instruction = (system_instruction or "") + _skill_text
                        yield f"[STAGE:📚 스킬 장착: {' · '.join(str(x) for x in _equipped[:4])}]\n"
            except Exception as _skill_err:
                logger.warning(f"[Guardian Job M2] 스킬 주입 실패(무시): {_skill_err}")

            # 🌐 [Hybrid Route A] 검색 질문: 구글 그라운딩 또는 Brave Search 주입
            # [🔱 Imperial Fix] Multimodal(Images)과 Grounding 도구는 동시 사용 시 충돌 가능성이 크므로 이미지가 있을 땐 비활성화
            if is_pure_search:
                logger.info("🌐 [IMPERIAL BYPASS] Google Search Grounding 원시 REST API 직접 호출 진입...")
                try:
                    # 1. API key 획득
                    api_key = getattr(self, "api_key", None)
                    if not api_key:
                        api_key = os.environ.get('GEMINI_API_KEY', '') or os.environ.get('GOOGLE_API_KEY', '')
                    
                    if not api_key and getattr(self, "_clients", None):
                        for k in self._clients.keys():
                            if k:
                                api_key = k
                                break
                    
                    if not api_key:
                        raise ValueError("GEMINI_API_KEY가 없습니다.")
                    
                    # 2. REST API 호출
                    url = f"https://generativelanguage.googleapis.com/v1beta/models/{actual_model_name}:generateContent?key={api_key}"
                    payload = {
                        "contents": [{"parts": [{"text": prompt}]}],
                        "tools": [{"google_search": {}}],
                        "generationConfig": {
                            "temperature": 0.0
                        }
                    }
                    if system_instruction:
                        payload["systemInstruction"] = {
                            "parts": [{"text": system_instruction}]
                        }
                    
                    logger.info(f"📡 REST API POST request to models/{actual_model_name}...")
                    import requests as _req
                    resp = _req.post(url, json=payload, timeout=30)
                    resp.raise_for_status()
                    
                    res_data = resp.json()
                    candidate = res_data.get('candidates', [{}])[0]
                    content = candidate.get('content', {})
                    parts = content.get('parts', [])
                    text = parts[0].get('text', '') if parts else ''
                    
                    grounding_sources = []
                    grounding_metadata = candidate.get('groundingMetadata', {})
                    attributions = grounding_metadata.get('groundingAttributions', [])
                    for attr in attributions:
                        web = attr.get('web', {})
                        if web:
                            grounding_sources.append(f"- [{web.get('title')}]({web.get('uri')})")
                    
                    if text:
                        # [🔱 Imperial Post-Process] 마왕님 극존칭 및 정체성 인격 합성
                        prefix = "마왕님, " if is_master else ""
                        text = f"{prefix}실시간 구글 검색 결과를 바탕으로 보고드립니다.\n\n{text}"
                        
                        # 청크 단위 yield로 스트리밍 효과 부여
                        chunk_size = 100
                        for i in range(0, len(text), chunk_size):
                            yield text[i:i+chunk_size]
                        
                        if grounding_sources:
                            sources_header = "\n\n---\n**🔱 [제국 정찰 정보원 / Imperial Recon Intel]**\n"
                            sources_text = sources_header + "\n".join(grounding_sources)
                            yield sources_text
                    else:
                        yield "⚠️ [브레인 무응답] 검색 결과를 파싱하지 못했습니다."
                    return
                except Exception as rest_err:
                    logger.error(f"🌐 [IMPERIAL BYPASS] REST API Grounding failed: {rest_err}. Falling back to standard flow...")
                    try:
                        search_tool = types.Tool(google_search=types.GoogleSearch())
                        genai_tools = [search_tool]
                        system_instruction = None
                    except Exception as e:
                        logger.error(f"Failed to load GoogleSearch tool: {e}")

            else:
                # 🔱 [Hybrid Route B] 비검색 질문: MCP + 커스텀 도구 세트 온전 주입
                # [🔱 Imperial Fix] SDK v1.x+ 에서는 함수 리스트를 직접 전달하는 것이 더 안정적임 (pydantic 검증 오류 방지)
                genai_tools = unified_function_declarations if unified_function_declarations else None

            # 🔱 [Imperial Tool Force Mode] AFC 기본 설정
            afc_config = types.AutomaticFunctionCallingConfig(disable=False)
            
            # 🔱 [Imperial Task Mode] 작업형 요청 감지 → 도구 강제 호출(mode=ANY)
            # gemini-flash-lite 경량 모델은 자발적으로 도구를 안 쓰는 경향이 있으므로
            # 파일/버그/UI/이미지 요청에서는 반드시 MCP 도구를 먼저 호출하도록 API 레벨에서 강제합니다.
            tool_config = None
            _diag_is_task = False
            if not is_pure_search and genai_tools:
                _diag_is_task = self._is_task_request(prompt, images)
                if _diag_is_task:
                    try:
                        tool_config = types.ToolConfig(
                            function_calling_config=types.FunctionCallingConfig(
                                mode="ANY"  # 모델이 반드시 도구 중 하나를 호출해야 함
                            )
                        )
                        logger.info("🔧 [Imperial Task Mode] 작업형 요청 감지 — 도구 강제 호출 모드(ANY) 활성화")
                    except Exception as tc_err:
                        logger.warning(f"[Task Mode] ToolConfig 생성 실패, 자동 모드 유지: {tc_err}")
                        tool_config = None

            # [🔱 Imperial Diag v2.7] 환각 진단: 요청별로 도구 세트/작업판정/강제모드를 기록.
            # mcp_tool_calls.log의 실제 CALL 기록과 대조하면, 모델이 '도구를 받고도 안 불렀는지'
            # vs '받지도 못했는지'를 객관적으로 가린다.
            try:
                import datetime as _dt
                _n_tools = len(genai_tools) if isinstance(genai_tools, list) else 0
                _diag = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scratch", "mcp_tool_calls.log")
                os.makedirs(os.path.dirname(_diag), exist_ok=True)
                with open(_diag, "a", encoding="utf-8") as _f:
                    _f.write(
                        f"[{_dt.datetime.now().isoformat()}] THINK model={actual_model_name} "
                        f"prompt='{(prompt or '')[:50]}' images={bool(images)} "
                        f"is_task={_diag_is_task} force_ANY={tool_config is not None} tools_in_set={_n_tools}\n"
                    )
            except Exception:
                pass

            # [🔱 Imperial V12] Unified Context Construction (History + Current Prompt)
            contents = []
            if history and isinstance(history, list):
                logger.info(f"🔱 [Neural Core] Injecting provided history context: {len(history)} entries.")
                for msg in history:
                    role = msg.get("role", "user")
                    content = msg.get("content", "")
                    if content:
                        contents.append(types.Content(role=role, parts=[types.Part(text=content)]))
            
            # Add current prompt
            current_parts = [types.Part(text=prompt)]
            if images:
                logger.info(f"🔱 [Neural Core] Multimodal Input Detected: Processing {len(images)} images...")
                for img in images:
                    try:
                        # [🔱] data:image/jpeg;base64, 접두어가 있는 경우 제거 (Heimdall은 이미 제거해서 보내지만 방어적 코드 추가)
                        img_data = img['data']
                        if isinstance(img_data, str) and ',' in img_data:
                            img_data = img_data.split(',')[1]

                        current_parts.append(types.Part.from_bytes(
                            data=base64.b64decode(img_data),
                            mime_type=img.get("mimeType", "image/jpeg")
                        ))
                    except Exception as img_err:
                        logger.error(f"Image Processing Error: {str(img_err)}")
                        yield f"⚠️ [Image Processing Error] {str(img_err)}"
                
            contents.append(types.Content(role="user", parts=current_parts))

            try:
                ImperialQuotaManager.validate_quota(actual_model_name)
            except QuotaExceededError as qe:
                yield f"\n{str(qe)}"
                return

            # [🔱 Imperial AFC & Grounding Fix v3.0]
            # 구글 검색 그라운딩(is_pure_search) 및 커스텀 도구(MCP) 사용 시 Non-Streaming 강제
            use_afc_non_streaming = (
                is_pure_search or
                (genai_tools and not isinstance(genai_tools[0], types.Tool))
            )

            full_text = ""
            grounding_sources = []
            self._afc_tool_results.clear()

            # 🌊 [Token Streaming + Manual FC] stream_tokens 옵트인 (태미니 CLI send-stream 등):
            # 커스텀 도구(callables)는 원래 AFC 비스트리밍으로 강제되어 응답이 한 덩어리로
            # 나왔다. 이 분기는 AFC를 끄고 '수동 함수호출 루프'를 돌며 텍스트 토큰을 즉시
            # 방출한다. 실패 시 기존(AFC 비스트리밍) 경로로 그대로 폴백 — 기존 호출자 무영향.
            if (stream_tokens and use_afc_non_streaming and not is_pure_search
                    and genai_tools and not isinstance(genai_tools[0], types.Tool)):
                try:
                    yield from self._gemini_manual_stream_loop(
                        client, actual_model_name, contents, genai_tools,
                        system_instruction, tool_config,
                    )
                    return
                except Exception as _ms_err:
                    logger.warning(f"🌊 [TokenStream] 수동 스트림 루프 실패 → AFC 비스트리밍 폴백: {_ms_err}")

            if use_afc_non_streaming:
                logger.info(f"🔱 [Neural Core][AFC-NonStream] Calling generate_content (non-streaming) with model={actual_model_name}")
                try:
                    _gen_config_args = dict(
                        system_instruction=system_instruction,
                        tools=genai_tools,
                        temperature=0.0,
                        automatic_function_calling=None if is_pure_search else afc_config,
                        safety_settings=[
                            types.SafetySetting(category="HARM_CATEGORY_HATE_SPEECH", threshold="BLOCK_NONE"),
                            types.SafetySetting(category="HARM_CATEGORY_HARASSMENT", threshold="BLOCK_NONE"),
                            types.SafetySetting(category="HARM_CATEGORY_SEXUALLY_EXPLICIT", threshold="BLOCK_NONE"),
                            types.SafetySetting(category="HARM_CATEGORY_DANGEROUS_CONTENT", threshold="BLOCK_NONE"),
                            types.SafetySetting(category="HARM_CATEGORY_CIVIC_INTEGRITY", threshold="BLOCK_NONE")
                        ]
                    )
                    # 🔱 [Task Mode] 도구 강제 호출 설정 주입
                    if tool_config is not None:
                        _gen_config_args["tool_config"] = tool_config
                    response = client.models.generate_content(
                        model=actual_model_name,
                        contents=contents,
                        config=types.GenerateContentConfig(**_gen_config_args)
                    )
                    self._emit_token_usage("google", actual_model_name, getattr(response, "usage_metadata", None))  # 🪙
                    try:
                        text = response.text or ""
                    except Exception:
                        text = ""
                    if not text and hasattr(response, 'candidates') and response.candidates:
                        cand = response.candidates[0]
                        if hasattr(cand, 'content') and hasattr(cand.content, 'parts'):
                            for part in cand.content.parts:
                                if hasattr(part, 'text') and part.text:
                                    text += part.text

                    # [🔱 Grounding Metadata Parsing for Non-Streaming]
                    if is_pure_search and response.candidates:
                        cand = response.candidates[0]
                        if hasattr(cand, 'grounding_metadata') and cand.grounding_metadata:
                            gm = cand.grounding_metadata
                            if hasattr(gm, 'grounding_chunks') and gm.grounding_chunks:
                                for chunk in gm.grounding_chunks:
                                    web = getattr(chunk, 'web', None)
                                    if web:
                                        title = getattr(web, 'title', 'Untitled')
                                        uri = getattr(web, 'uri', '#')
                                        grounding_sources.append(f"- [{title}]({uri})")

                    if text:
                        # [🔱 Imperial Post-Process] 구글 실시간 검색 완료 후, 마왕님 극존칭 및 그림자 두뇌 정체성 인격 합성
                        if is_pure_search:
                            if is_master:
                                if not text.strip().startswith("마왕님"):
                                    text = f"마왕님, 실시간 구글 검색 결과를 바탕으로 보고드립니다.\n\n{text}"
                            else:
                                if not text.strip().startswith("실시간"):
                                    text = f"실시간 구글 검색 결과를 바탕으로 보고드립니다.\n\n{text}"
                        
                        logger.info(f"🔱 [Neural Core][AFC-NonStream] Response text len={len(text)}")
                        full_text = text
                        # 청크처럼 나눠서 yield (자연스러운 스트리밍 효과)
                        chunk_size = 100
                        for i in range(0, len(text), chunk_size):
                            yield text[i:i+chunk_size]
                        
                        # [🔱 Grounding Sources Display]
                        if grounding_sources:
                            sources_header = "\n\n---\n**🔱 [제국 정찰 정보원 / Imperial Recon Intel]**\n"
                            sources_text = sources_header + "\n".join(grounding_sources)
                            yield sources_text

                        # [🔱 Image Attach Fix] 모델이 텍스트 요약을 만들면서 이미지 생성 결과(마크다운 ![](url))를
                        # 누락하는 경우가 잦다. 도구 결과에 이미지가 있고 본문에 아직 없으면 강제로 첨부하여
                        # 채팅에 이미지가 항상 표시되게 한다. (generate_image 등)
                        for _r in self._afc_tool_results:
                            if "![" in str(_r) and str(_r) not in full_text:
                                yield "\n\n" + _r
                                full_text += "\n\n" + _r
                    elif self._afc_tool_results:
                        logger.info(f"[AFC-NonStream] Tool results found: {len(self._afc_tool_results)}")
                        for result in self._afc_tool_results:
                            full_text += result + "\n"
                            yield result + "\n"
                    else:
                        logger.warning(f"[AFC-NonStream] Empty response from '{actual_model_name}'. Triggering fallback.")
                        if not skip_fallback:
                            yield from self._think_fallback(prompt, system_instruction=system_instruction, skip_tools=skip_tools, tools=None)
                        else:
                            yield "⚠️ [브레인 무응답] 모델이 응답을 생성하지 못했습니다."
                    self._afc_tool_results.clear()
                    return
                except Exception as afc_e:
                    logger.warning(f"[AFC-NonStream] Exception: {afc_e}. Falling back to streaming mode.")
                    # 실패 시 아래 스트리밍 모드로 계속 진행

            logger.info(f"🔱 [Neural Core] Calling generate_content_stream with model={actual_model_name}")
            _stream_config_args = dict(
                system_instruction=system_instruction,
                tools=genai_tools,
                temperature=0.0,
                automatic_function_calling=afc_config,
                safety_settings=[
                    types.SafetySetting(category="HARM_CATEGORY_HATE_SPEECH", threshold="BLOCK_NONE"),
                    types.SafetySetting(category="HARM_CATEGORY_HARASSMENT", threshold="BLOCK_NONE"),
                    types.SafetySetting(category="HARM_CATEGORY_SEXUALLY_EXPLICIT", threshold="BLOCK_NONE"),
                    types.SafetySetting(category="HARM_CATEGORY_DANGEROUS_CONTENT", threshold="BLOCK_NONE"),
                    types.SafetySetting(category="HARM_CATEGORY_CIVIC_INTEGRITY", threshold="BLOCK_NONE")
                ]
            )
            # 🔱 [Task Mode] 스트리밍 경로에도 도구 강제 호출 설정 주입
            if tool_config is not None:
                _stream_config_args["tool_config"] = tool_config
            response_stream = client.models.generate_content_stream(
                model=actual_model_name,
                contents=contents,
                config=types.GenerateContentConfig(**_stream_config_args)
            )
            logger.info("🔱 [Neural Core] response_stream obtained. Entering iteration loop...")
            _usage_acc = None  # 🪙 토큰 사용량(누적 청크)

            for chunk in response_stream:
                logger.info(f"🔱 [Neural Core] Chunk received (type={type(chunk)})")
                try:
                    _um = getattr(chunk, "usage_metadata", None)
                    if _um is not None:
                        _usage_acc = _um
                    # 🌐 Extract Grounding Metadata (Sources)
                    # Support both top-level and candidate-level metadata (SDK v1.x+)
                    gm = None
                    if hasattr(chunk, 'grounding_metadata') and chunk.grounding_metadata:
                        gm = chunk.grounding_metadata
                    elif hasattr(chunk, 'candidates') and chunk.candidates:
                        cand = chunk.candidates[0]
                        if hasattr(cand, 'grounding_metadata') and cand.grounding_metadata:
                            gm = cand.grounding_metadata
                    
                    if gm:
                        # [🔱 Imperial V11] Log raw grounding metadata for diagnostics
                        try:
                            diag_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "scratch", "grounding_diagnostics.log")
                            os.makedirs(os.path.dirname(diag_path), exist_ok=True)
                            with open(diag_path, "a", encoding="utf-8") as f:
                                # Safe string conversion for complex SDK objects
                                f.write(f"\n[{datetime.now().isoformat()}] metadata_len: {len(str(gm))}\n")
                        except: pass

                    # 🌐 Extract Text (V16 Deep Padding)
                    text = ""
                    try:
                        if hasattr(chunk, 'text') and chunk.text:
                            text = chunk.text
                    except: pass
                    
                    if not text and hasattr(chunk, 'candidates') and chunk.candidates:
                        cand = chunk.candidates[0]
                        if hasattr(cand, 'content') and hasattr(cand.content, 'parts'):
                            for part in cand.content.parts:
                                if hasattr(part, 'text'):
                                    text += part.text
                    
                    if text:
                        logger.info(f"🔱 [Neural Core] Text extracted (len={len(text)})")
                        full_text += text
                        yield text
                    else:
                        logger.warning("🔱 [Neural Core] Chunk received without printable text.")
                except Exception as chunk_e:
                    logger.warning(f"🔱 [Neural Core] Chunk processing bypass: {chunk_e}")
                    continue

            self._emit_token_usage("google", actual_model_name, _usage_acc)  # 🪙

            # 🌐 Append Citations at the end if found
            if grounding_sources:
                sources_header = "\n\n---\n**🔱 [제국 정찰 정보원 / Imperial Recon Intel]**\n"
                sources_text = sources_header + "\n".join(grounding_sources)
                # 🛠️ [🔱 Logic] Add instruction to check logs
                log_hint = "\n\n> [!TIP]\n> 상세 검색 근거와 AI가 참조한 원문 스니펫은 `shadow_brain_core/scratch/grounding_diagnostics.log`에서 확인하실 수 있습니다."
                sources_text += log_hint
                full_text += sources_text
                yield sources_text
            
            if not full_text:
                if self._afc_tool_results:
                    # Model executed tools but didn't generate summary text (common with lite models)
                    logger.info(f"[AFC Fallback] Returning {len(self._afc_tool_results)} tool result(s) as response.")
                    for result in self._afc_tool_results:
                        yield result + "\n"
                else:
                    # Empty response with no tool results — lite model failure, trigger fallback
                    logger.warning(f"Empty response from '{actual_model_name}' with no tool results. Triggering fallback.")
                    if not skip_fallback:
                        yield from self._think_fallback(prompt, system_instruction=system_instruction, skip_tools=skip_tools, tools=None)
                    else:
                        yield "⚠️ [브레인 무응답] 모델이 응답을 생성하지 못했습니다. 다른 모델을 선택하거나 다시 시도해주세요."
                self._afc_tool_results.clear()

        except Exception as e:
            err_msg = str(e)
            import traceback
            logger.error(f"Brain Error: {err_msg}\n{traceback.format_exc()}")
            
            # [🔱 Imperial Fix] 만약 도구/검색(Grounding) 문제로 에러가 났다면, 도구 없이 재시도 (Fallback 방지)
            logger.warning(f"Brain Error encountered: {err_msg}. Investigating for Tool Failure...")
            
            # 2026 SDK 버전에서 'google_search'나 'grounding' 관련 필드 처리 중 에러가 자주 발생하므로 범위를 넓힙니다.
            # 특히 'finish_reason'이나 'candidate' 관련 인덱스 에러는 대부분 도구 응답 파싱 실패입니다.
            is_search_error = any(kw in err_msg.lower() for kw in [
                "tool", "search", "grounding", "function", "grounding_metadata", 
                "candidate", "safety_ratings", "citation", "finish_reason", "index out of range"
            ])
            
            if not skip_tools and genai_tools and is_search_error:
                if self.brave_api_key:
                    logger.warning("🔱 [RECOVERY] Google Search Grounding failed. Bypassing to Brave Search API...")
                    try:
                        brave_results = self._search_brave(prompt)
                        if brave_results:
                            enriched_prompt = f"다음 실시간 Brave 검색 결과를 바탕으로 질문에 대해 명확히 답변하고, 답변 끝에 반드시 해당 검색 출처 URL 링크도 보기 좋게 포함하십시오.\n\n{brave_results}\n\n질문: {prompt}"
                            yield from self.think(enriched_prompt, model_name=model_name, images=images, 
                                                 system_instruction=system_instruction, skip_fallback=skip_fallback, 
                                                 provider=provider, skip_tools=True, session_id=session_id)
                            return
                    except Exception as brave_err:
                        logger.error(f"Brave Fallback failed: {brave_err}")
                
                logger.warning(f"🔱 [RECOVERY] Google Search/Tool failure detected ({err_msg}). Retrying WITHOUT tools to save Mission...")
                yield from self.think(prompt, model_name=model_name, images=images, 
                                     system_instruction=system_instruction, skip_fallback=skip_fallback, 
                                     provider=provider, skip_tools=True, session_id=session_id)
                return

            if skip_fallback:
                yield f"\n❌ Critical Brain Malfunction (Primary/Secondary failed): {err_msg}"
            else:
                logger.warning(f"Imperial Brain Malfunction: {err_msg}. Fail-Fast triggered.")
                yield f"\n❌ [Google API 에러] 모델 호출에 실패했습니다: {err_msg}"

    def _search_brave(self, query: str) -> str:
        """
        Brave Search API를 통해 실시간 웹 정보를 조회하고 마크다운 형태로 반환합니다.
        """
        if not self.brave_api_key:
            return ""
            
        logger.info(f"📡 Querying Brave Search API for: {query}")
        headers = {
            "Accept": "application/json",
            "Accept-Encoding": "gzip",
            "X-Subscription-Token": self.brave_api_key,
            "Cache-Control": "no-cache"
        }
        params = {
            "q": query,
            "count": 5
        }
        url = "https://api.search.brave.com/res/v1/web/search"
        try:
            response = requests.get(url, headers=headers, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            
            results = data.get("web", {}).get("results", [])
            if not results:
                return ""
                
            md_list = []
            md_list.append("\n=== [REAL-TIME BRAVE WEB SEARCH RESULTS] ===")
            for idx, item in enumerate(results, 1):
                title = item.get("title", "")
                url_val = item.get("url", "")
                snippet = item.get("description", "")
                md_list.append(f"[{idx}] {title}\nURL: {url_val}\nSnippet: {snippet}\n")
            md_list.append("============================================\n")
            return "\n".join(md_list)
        except Exception as e:
            logger.error(f"Brave Search API request failed: {e}")
            return ""

    def direct_search(self, query: str, model_name: str = None) -> Dict[str, Any]:
        """
        [🔱 Imperial High-Performance Search]
        AI 검색 도우미와 100% 동일한 로직을 쉐도우 브레인 엔진 내부에서 수행합니다.
        스트리밍이 아닌 한 번의 호출로 결과와 출처를 즉시 반환합니다.
        """
        if model_name is None:
            # [🔱 Imperial Fix] Use configured primary model or safe fallback
            model_name = getattr(self, "primary_model", "gemini-3.1-flash-lite")
            
        target_model = self._resolve_actual_model("google", model_name)
        
        try:
            # [🔱 Imperial Fix] google-genai SDK를 사용하여 다이렉트 호출 수행
            # generate_content_stream이 아닌 generate_content 사용
            search_tool = types.Tool(google_search=types.GoogleSearch())
            
            client = self._get_client() # Default client
            response = client.models.generate_content(
                model=target_model,
                contents=query,
                config=types.GenerateContentConfig(
                    tools=[search_tool],
                    temperature=0.0,
                    system_instruction="당신은 제국의 지능 정착병입니다. 한국어로 답변하십시오. 검색 결과를 바탕으로 팩트 위주로 보고하십시오."
                )
            )
            
            text = response.text or ""
            sources = []
            
            # Extract grounding metadata
            if response.candidates and response.candidates[0].grounding_metadata:
                gm = response.candidates[0].grounding_metadata
                if hasattr(gm, 'grounding_chunks') and gm.grounding_chunks:
                    for chunk in gm.grounding_chunks:
                        web = getattr(chunk, 'web', None)
                        if web:
                            sources.append({
                                "title": getattr(web, "title", "Untitled"),
                                "uri": getattr(web, "uri", "#")
                            })
                            
            return {
                "status": "SUCCESS",
                "text": text,
                "sources": sources,
                "model": target_model
            }
            
        except Exception as e:
            logger.error(f"[Turbo Search Error] {e}")
            if self.brave_api_key:
                logger.info("🔱 [RECOVERY] Google Search failed. Attempting fallback to Brave Search API...")
                try:
                    brave_results = self._search_brave(query)
                    if brave_results:
                        client = self._get_client()
                        response = client.models.generate_content(
                            model=target_model,
                            contents=f"다음 질문에 대해 제공된 Brave 검색 결과를 바탕으로 답변하십시오.\n질문: {query}\n\n{brave_results}",
                            config=types.GenerateContentConfig(
                                temperature=0.0,
                                system_instruction="당신은 제국의 지능 정착병입니다. 한국어로 답변하십시오. 제공된 검색 결과 자료를 완벽히 활용해 답변 끝에 출처 URL 링크도 포함해 마크다운 형식으로 보고하십시오."
                            )
                        )
                        # Parse out sources from brave_results
                        sources = []
                        import re
                        matches = re.findall(r"\[(\d+)\] (.*?)\nURL: (.*?)\n", brave_results)
                        for m in matches:
                            sources.append({
                                "title": m[1],
                                "uri": m[2]
                            })
                        return {
                            "status": "SUCCESS",
                            "text": response.text or "",
                            "sources": sources,
                            "model": target_model + " (Brave Fallback)"
                        }
                except Exception as ex:
                    logger.error(f"[Brave Fallback Error] {ex}")
            return {
                "status": "ERROR",
                "message": str(e)
            }