shadow_brain_core/brain/tool_runtime.py

"""🛡️ 제국 도구 런타임 (Shadow Brain Tool Runtime) — Phase 1: 디스패처 단일화.

3개 OpenAI식 대화 경로(`_think_copilot` / `_think_atlas` / `_think_openrouter`)가
공유하는 단일 도구 실행 런타임. 인자 검증·자동 보정·ask 가드/서킷브레이커·
텍스트→카드 수확을 한 곳에 모아, 복붙 디스패처의 드리프트 버그를 제거한다.

설계 원칙:
- 순환 import 방지를 위해 `core`를 import 하지 않는다. 필요한 헬퍼(_is_debug,
  _summarize)는 생성 시 주입받고, 도구 핸들러는 brain 인스턴스 메서드로 위임한다.
- ask_user처럼 UI 블록을 흘려야 하는 도구는 제너레이터로 처리하고(`yield from`),
  비-yield 도구는 `dispatch()` 한 곳에서 순수 함수로 처리한다.
"""
import json
import logging
import re

logger = logging.getLogger("JarvisBrain")

# 원형 숫자(①~⑳) — 모델이 옵션 번호로 자주 쓴다 (텍스트→카드 파서용)
_CIRCLED_NUM = "①②③④⑤⑥⑦⑧⑨⑩⑪⑫⑬⑭⑮⑯⑰⑱⑲⑳"


def _fn(name, desc, props=None, required=None):
    """OpenAI 호환 function 스키마 1건 생성."""
    fn = {"name": name, "description": desc,
          "parameters": {"type": "object", "properties": props or {}}}
    if required:
        fn["parameters"]["required"] = required
    return {"type": "function", "function": fn}


# ──────────────────────────────────────────────────────────────
# 🧰 네이티브 도구 레지스트리 (단일 진실 출처: 스키마 ↔ 핸들러 1:1)
#   - schema  : 모델에 노출되는 OpenAI 함수 스키마
#   - handler : (brain, args, ctx) -> str  (None이면 특수 처리: ask_user)
#   새 도구는 이 리스트에만 추가하면 스키마/핸들러가 동시에 등록된다 → 드리프트 불가.
# ──────────────────────────────────────────────────────────────
NATIVE_TOOLS = [
    {
        "schema": _fn("recall_memory",
            "Searches long-term conversation memory (Aegis vector store). MUST be used before answering questions about past conversations or user facts. If no results, say you don't remember instead of guessing.",
            {"query": {"type": "string", "description": "What to search for in memory"},
             "limit": {"type": "integer", "description": "Max memories to return (default 5)"}},
            ["query"]),
        "handler": lambda b, a, ctx: b.recall_memory(a.get("query", ""), a.get("limit", 5)),
    },
    {
        # ⚠️ ask_user는 UI 카드 방출+대기가 필요해 dispatch()가 아닌 _handle_ask_user에서 처리 (handler=None)
        "schema": _fn("ask_user",
            "Pauses mid-task and asks the user to choose between options via UI buttons with a countdown timer. Use when a decision genuinely changes what you do next (A/B/C choices, destructive actions, ambiguous requirements). PROACTIVE RULE: whenever your reply would otherwise end by asking the user to pick from an enumerated list (A/B/C, ①②③, 'which one?'), call this tool INSTEAD of writing the options as plain text — even if the user did not ask you to use it. The user's choice is returned as the tool result; on timeout the question is auto-cancelled and you must proceed conservatively. CRITICAL: you MUST actually CALL this tool — NEVER write an 'imperial-ask' code block or ask_id JSON as plain text yourself; hand-written blocks render nothing and no answer will ever arrive. ALWAYS fill BOTH 'question' and 2-6 'options' — never call with empty arguments. If the user's request is vague (e.g. 'ask me something', 'just ask'), DO NOT return empty options or bounce the question back — INVENT sensible options yourself (e.g. for a general help request: ['잡담하기','작업 지시','정보 검색','코드/기술 지원','기타']). Do NOT use for questions you can answer yourself, free-form questions, or lists with more than 6 options.",
            {"question": {"type": "string", "description": "The question to ask (clear, ends with ?)"},
             "options": {"type": "array", "items": {"type": "string"}, "description": "2-6 choice labels (e.g., ['A안: 전체 재구축', 'B안: 부분 수정'])"},
             "timeout_seconds": {"type": "integer", "description": "Wait time in seconds, 30-600 (default 120)"}},
            ["question", "options"]),
        "handler": None,
    },
    {
        "schema": _fn("search_knowledge",
            "Searches the Imperial Knowledge Base (verified operational facts: config paths, ports, service locations, architecture). MUST be used before answering questions about system configuration or infrastructure. Prefer this over guessing from partial file scans.",
            {"query": {"type": "string", "description": "Keywords to search (e.g., 'OpenClaw 설정 경로')"},
             "limit": {"type": "integer", "description": "Max facts to return (default 5)"}},
            ["query"]),
        "handler": lambda b, a, ctx: b.search_knowledge(a.get("query", ""), a.get("limit", 5)),
    },
    {
        "schema": _fn("search_media",
            "Searches royalty-free, free-to-use media (photos, videos, GIFs) from Pexels and Pixabay and returns real result URLs (original, download, source page). MUST be used when the user asks to find/search images, photos, videos, GIFs, wallpapers, or stock media. Do NOT answer such requests by recommending websites (Pixabay/Unsplash/Canva) from memory — call this tool and return the actual URLs.",
            {"query": {"type": "string", "description": "Search keywords (e.g., '고양이', 'fire')"},
             "media_type": {"type": "string", "enum": ["photo", "video", "gif"], "description": "Media type (default photo)"},
             "provider": {"type": "string", "enum": ["pexels", "pixabay"], "description": "Source for photo/video (default pexels). Ignored for gif (always Pixabay)."},
             "limit": {"type": "integer", "description": "Max results 1-20 (default 8)"}},
            ["query"]),
        # 🔧 query가 비면 유저 발화로 자동 보정 (약한 모델 빈 인자 방어)
        "handler": None,
    },
    {
        # 🎥 yielding tool — handler=None, run_tool_calls가 imperial-youtube 카드를 스트림에 직접 방출
        "schema": _fn("search_youtube",
            "Searches REAL YouTube videos (YouTube Data API) and shows them in chat as playable cards "
            "with thumbnails, titles and channels. MUST be used whenever the user asks to find/search "
            "YouTube videos, or mentions '유튜브'/'youtube' in a search request (e.g. '유튜브에서 강아지 "
            "영상 찾아줘'). This is DIFFERENT from search_media (which returns stock Pexels/Pixabay clips) — "
            "if the user said YouTube, use THIS tool, not search_media. The cards render automatically; "
            "after calling, give only a short one-line confirmation and do NOT list the results yourself.",
            {"query": {"type": "string", "description": "Search keywords (e.g., '강아지', '용인 여행')"},
             "limit": {"type": "integer", "description": "Max videos 1-20 (default 6)"}},
            ["query"]),
        "handler": None,
    },
    {
        # 🧊 yielding tool — handler=None, run_tool_calls가 imperial-3d 카드를 스트림에 직접 방출
        "schema": _fn("search_3d_model",
            "Searches royalty-free 3D models (Pixabay 3D + Sketchfab) and shows them in chat as "
            "interactive cards with preview thumbnails and a live '3D viewer' button. MUST be used when "
            "the user asks to find/search 3D models, 3D assets, GLB or VRM models (e.g. '드래곤 3d 모델 "
            "찾아줘', 'vrm 모델 몇개 찾아줘'). Do NOT answer from memory or recommend websites. The cards "
            "AND the 3D viewer are rendered automatically by the client — after calling, give only a "
            "short one-line confirmation; do NOT list the results or URLs yourself.",
            {"query": {"type": "string", "description": "Search keywords (e.g., 'dragon', 'vrm', 'anime girl')"},
             "limit": {"type": "integer", "description": "Max models 1-20 (default 8)"},
             "source": {"type": "string", "enum": ["both", "pixabay", "sketchfab"],
                        "description": "Model source. 'sketchfab' is best for VRM/anime/character models; "
                                       "'pixabay' for simple objects; default 'both' mixes the two."}},
            ["query"]),
        "handler": None,
    },
    {
        "schema": _fn("web_search",
            "Searches the live web (Naver-first: 네이버→Daum→Google via headless browser) for real-time "
            "info such as current events, people/officials, news, weather, prices/FX. Returns grounded "
            "snippets to base your answer on. Use for time-sensitive/factual WEB questions. For images/"
            "photos/GIF/video use search_media; for 3D models use search_3d_model.",
            {"query": {"type": "string", "description": "Search keywords (e.g., '대한민국 대통령', '오늘 서울 날씨')"},
             "engine": {"type": "string", "enum": ["naver", "daum", "google", "all"], "description": "Search engine to use (default naver)"}},
            ["query"]),
        "handler": lambda b, a, ctx: b.web_search(
            (a.get("query") or "").strip() or ctx.user_prompt,
            a.get("engine", "naver"),
        ),
    },
    {
        "schema": _fn("save_knowledge",
            "Saves a newly verified operational fact to the shared Imperial Knowledge Base (all guardians can see it). Call whenever you discover or verify a system fact. Same topic updates the existing entry.",
            {"topic": {"type": "string", "description": "Short subject (e.g., 'OpenClaw 실설정 경로')"},
             "fact": {"type": "string", "description": "The verified fact, present tense"},
             "source": {"type": "string", "description": "How it was verified"},
             "scope": {"type": "string", "enum": ["global", "environment"], "description": "'global' if true on any PC (product behavior), 'environment' if only true on this machine (local paths, running ports)"},
             "verify_hint": {"type": "string", "description": "How to re-verify in a new environment (e.g., 'check_port 18789', 'find_process openclaw')"}},
            ["topic", "fact"]),
        "handler": lambda b, a, ctx: b.save_knowledge(a.get("topic", ""), a.get("fact", ""), a.get("source", ""), a.get("scope", "global"), a.get("verify_hint", "")),
    },
    {
        "schema": _fn("check_port",
            "Read-only: checks if a local TCP port is listening RIGHT NOW and which process owns it. MUST be used to verify environment-scoped knowledge (ports/services) before asserting it as current fact.",
            {"port": {"type": "integer", "description": "TCP port to check (e.g., 18789)"}},
            ["port"]),
        "handler": lambda b, a, ctx: b.check_port(a.get("port", 0)),
    },
    {
        "schema": _fn("find_process",
            "Read-only: finds running processes matching a keyword and shows their command lines (reveals actual config paths in use). Use to verify which install/config a running service actually uses.",
            {"keyword": {"type": "string", "description": "Substring to match in process name/command line (e.g., 'openclaw')"},
             "limit": {"type": "integer", "description": "Max processes (default 5)"}},
            ["keyword"]),
        "handler": lambda b, a, ctx: b.find_process(a.get("keyword", ""), a.get("limit", 5)),
    },
    {
        "schema": _fn("review_knowledge_proposals",
            "[Curator] Lists pending knowledge proposals from other guardians. You are the Imperial Knowledge librarian — verify each with observation tools before approving."),
        "handler": lambda b, a, ctx: b.review_knowledge_proposals(),
    },
    {
        "schema": _fn("judge_knowledge_proposal",
            "[Curator] Approves/rejects a pending knowledge proposal. ONLY approve after verifying the fact yourself. Approval replaces existing approved fact on the same topic.",
            {"proposal_id": {"type": "string", "description": "Proposal id from review_knowledge_proposals"},
             "approve": {"type": "boolean", "description": "true=approve, false=reject"},
             "note": {"type": "string", "description": "Verification basis or rejection reason"}},
            ["proposal_id", "approve"]),
        "handler": lambda b, a, ctx: b.judge_knowledge_proposal(a.get("proposal_id", ""), a.get("approve", False), a.get("note", "")),
    },
    {
        "schema": _fn("count_history",
            "Counts real chat history records in the DB. MUST be used for 'how many times / since when' questions. Never guess numbers.",
            {"query": {"type": "string", "description": "Text to count occurrences of. Empty counts all messages."},
             "message_type": {"type": "integer", "description": "0=user only, 1=AI only, -1=all (default)"}}),
        "handler": lambda b, a, ctx: b.count_history(a.get("query", ""), a.get("message_type", -1)),
    },
    {
        "schema": _fn("hugin_navigate",
            "Navigates the Master's visible browser to a specific URL.",
            {"url": {"type": "string", "description": "The URL to navigate to (e.g., 'https://www.google.com')"}},
            ["url"]),
        "handler": lambda b, a, ctx: b.browser.hugin_navigate(a.get("url")),
    },
    {
        "schema": _fn("hugin_screenshot",
            "Takes a screenshot of the Master's current visible browser state."),
        "handler": lambda b, a, ctx: b.browser.hugin_screenshot(),
    },
    {
        "schema": _fn("hugin_click",
            "Clicks an element containing the specified text in the visible browser.",
            {"text": {"type": "string", "description": "The text of the element to click."}},
            ["text"]),
        "handler": lambda b, a, ctx: b.browser.hugin_click(a.get("text")),
    },
    {
        "schema": _fn("hugin_type",
            "Types text into the currently active or focused element in the visible browser.",
            {"text": {"type": "string", "description": "The text to type."}},
            ["text"]),
        "handler": lambda b, a, ctx: b.browser.hugin_type(a.get("text")),
    },
    {
        "schema": _fn("sovereign_type_text",
            "Types text directly into the Master's active window (OS level).",
            {"text": {"type": "string", "description": "The text to type (e.g., chat message)."}},
            ["text"]),
        "handler": lambda b, a, ctx: b.sovereign_type_text(a.get("text")),
    },
    {
        "schema": _fn("sovereign_press_key",
            "Presses a keyboard key directly in the Master's OS. Use 'enter' to submit a chat message after typing.",
            {"keys": {"type": "string", "description": "The key to press (e.g., 'enter', 'esc')."}},
            ["keys"]),
        "handler": lambda b, a, ctx: b.sovereign_press_key(a.get("keys")),
    },
    {
        "schema": _fn("sovereign_run_app",
            "Launches an OS application directly (e.g. 'notepad', 'calc').",
            {"app_name": {"type": "string", "description": "The name or command to run."}},
            ["app_name"]),
        "handler": lambda b, a, ctx: b.sovereign_run_app(a.get("app_name")),
    },
    {
        "schema": _fn("delegate_team_task",
            "Records a formal task for a specialized guardian to pick up later.",
            {"title": {"type": "string", "description": "Brief task title"},
             "assignee_guardian": {"type": "string", "description": "Guardian name (e.g., 'Acubens', 'Hermes')"},
             "description": {"type": "string", "description": "Detailed requirements for the task"},
             "priority": {"type": "string", "enum": ["LOW", "MEDIUM", "HIGH", "CRITICAL"]}},
            ["title", "assignee_guardian"]),
        "handler": lambda b, a, ctx: b.delegate_team_task(a.get("title"), a.get("assignee_guardian"), a.get("description", ""), a.get("priority", "MEDIUM")),
    },
    {
        "schema": _fn("dispatch_to_hydra",
            "Immediately dispatches a goal to Hydra for parallel execution and synthesized reporting.",
            {"mission_goal": {"type": "string", "description": "The mission objective (e.g., 'Develop a python script and verify its output')"}},
            ["mission_goal"]),
        "handler": lambda b, a, ctx: b.dispatch_to_hydra(a.get("mission_goal")),
    },
]

# 🗂️ 옵시디언 지식 레이어 및 Graphify — 로컬 또는 클라우드 볼트 활성 시 노출.
#   core를 import하지 않는 규칙은 유지: obsidian_router는 독립 모듈이라 순환 없음.
try:
    import obsidian_router as _ov
    if _ov.is_any_enabled():
        NATIVE_TOOLS += [
            {
                "schema": _fn("obsidian_search",
                    "Searches the Obsidian vault (hand-curated/distilled knowledge in markdown notes) — "
                    "a separate layer from conversation memory. Use to find organized notes, design docs and "
                    "reference material the user has written. Returns note paths + snippets. Works in both Local/Cloud. "
                    "By default searches the active vault (local when running locally). Pass source='cloud' to search the "
                    "R2 cloud vault via PGVector even while running locally (트리거: '클라우드 옵시디언/R2에서 찾아줘').",
                    {"query": {"type": "string", "description": "Keywords to search in note titles/bodies"},
                     "limit": {"type": "integer", "description": "Max notes (default 8)"},
                     "source": {"type": "string", "enum": ["local", "cloud"], "description": "Which vault to search. Omit for active vault; 'cloud' = R2/PGVector."}},
                    ["query"]),
                "handler": lambda b, a, ctx: b.obsidian_search(a.get("query", ""), a.get("limit", 8), a.get("source")),
            },
            {
                "schema": _fn("obsidian_read",
                    "Reads the full body of one Obsidian note by its vault-relative path (from obsidian_search). Works in both "
                    "Local/Cloud. Pass source='cloud' to read from the R2 cloud vault even while running locally.",
                    {"path": {"type": "string", "description": "Vault-relative path, e.g. 'projects/jarvis.md'"},
                     "source": {"type": "string", "enum": ["local", "cloud"], "description": "Which vault to read from. Omit for active vault; 'cloud' = R2."}},
                    ["path"]),
                "handler": lambda b, a, ctx: b.obsidian_read(a.get("path", ""), a.get("source")),
            },
            {
                "schema": _fn("obsidian_write",
                    "Creates/updates a markdown note in the Obsidian vault. MUST be called — and ACTUALLY "
                    "CALLED, not merely promised — whenever the user asks to create/write/save/organize/record a "
                    "note, wiki, document, page, or memo in Obsidian (트리거: '옵시디언에 ~ 만들어/작성/정리/저장해줘', "
                    "'위키 만들어줘', '노트로 정리해줘', '문서 작성해줘'). CRITICAL: NEVER reply '곧 만들겠습니다/생성하겠습니다' "
                    "without calling this tool in the SAME turn — a promise with no tool call creates nothing and is a "
                    "failure. YOU must generate the FULL markdown body yourself (use Obsidian syntax: # headings, "
                    "[[backlinks]], #tags, - lists) and pass it in 'content'. For a multi-page wiki, call this tool "
                    "repeatedly (one call per note) and link them with [[ ]]. If unsure whether a note exists, call "
                    "obsidian_search first, then write with mode='overwrite' or 'append' as appropriate. Works in both Local/Cloud.",
                    {"path": {"type": "string", "description": "Vault-relative path (e.g. 'Wiki/태민게임즈/index.md'); '.md' auto-added"},
                     "content": {"type": "string", "description": "FULL markdown body of the note (you write it, not a placeholder)"},
                     "mode": {"type": "string", "enum": ["create", "overwrite", "append"], "description": "create=fail if exists (default); overwrite=replace; append=add to end"}},
                    ["path", "content"]),
                "handler": lambda b, a, ctx: b.obsidian_write(a.get("path", ""), a.get("content", ""), a.get("mode", "create")),
            },
            {
                "schema": _fn("project_graph_query",
                    "Queries the Graphify knowledge graph (project map) of the code and documentation structure. "
                    "Use to check file import dependencies, class methods, and document references to understand "
                    "the project layout and avoid reading raw code files blindly.",
                    {"query": {"type": "string", "description": "Name of the file, module, or class to query (e.g., 'web_server.py', 'CodingToolkit')"},
                     "mode": {"type": "string", "enum": ["depends_on", "details", "all"], "description": "depends_on=import relations (default); details=class/file metadata; all=overall graph summary"}},
                    ["query"]),
                "handler": lambda b, a, ctx: b.project_graph_query(a.get("query", ""), a.get("mode", "depends_on")),
            },
        ]
        logger.info("[ToolRuntime] 옵시디언 및 지식 그래프 레이어 활성 — obsidian 및 project_graph_query 노출")
except Exception as _e:
    logger.warning(f"[ToolRuntime] 옵시디언/그래프 도구 등록 건너뜀: {_e}")

# 🧊 yielding 도구: dispatch가 아니라 run_tool_calls가 UI 블록을 스트림에 직접 방출 (handler=None 허용)
_YIELDING_TOOLS = {"ask_user", "search_media", "search_3d_model", "search_youtube"}

# 🧹 상황형 도구: 해당 의도일 때만 노출 (그 외엔 목록을 작게 유지 → 약한 모델 정확도↑)
_HUGIN_TOOLS = {"hugin_navigate", "hugin_screenshot", "hugin_click", "hugin_type"}
_OS_TOOLS = {"sovereign_type_text", "sovereign_press_key", "sovereign_run_app"}
_DELEGATION_TOOLS = {"dispatch_to_hydra", "delegate_team_task"}


class RuntimeCtx:
    """한 턴(요청) 동안의 도구 루프 상태."""

    def __init__(self, user_prompt, interactive=True, debug=False, display_name="Brain"):
        self.user_prompt = (user_prompt or "")
        self.interactive = interactive
        self.debug = debug
        self.display_name = display_name
        # 루프 상태
        self.used_tools = False        # 🔱 Empty-Final Guard
        self.nudged_final = False
        self.ask_rescued = False       # 🛟 Fake-Ask Rescue (턴당 1회)
        self.blocked_ask = 0           # 🛡️ Ask Circuit-Breaker 누적
        self.force_no_tools = False    # 트립되면 도구 미부착 → 텍스트 마무리 강제
        self.ask_attempted = False     # 🍽 텍스트→카드 변환 게이트
        self.expect_obsidian_write = False  # 🗂️ 옵시디언 작성 의도 — 약속만 하고 안 쓰면 재촉
        self.obsidian_written = False   # 🗂️ 이번 턴에 obsidian_write 를 실제 호출했는지(중복 작성 방지)
        self.promise_nudges = 0        # 🗂️ 약속-미이행 재촉 횟수(무한루프 방지)
        self.suppress_stream = False   # 🔇 카드 변환 가능성 있는 텍스트를 라이브 스트리밍하지 않고 보류


class ToolRuntime:
    """3개 OpenAI식 대화 경로가 공유하는 도구 실행 런타임."""

    def __init__(self, brain, is_debug=None, summarize=None, get_mcp_client=None):
        self.brain = brain
        self._is_debug = is_debug or (lambda: False)
        self._summarize = summarize
        self._get_mcp_client = get_mcp_client
        # 레지스트리에서 name→handler 맵 구성 (handler=None인 ask_user 제외)
        self._handlers = {
            e["schema"]["function"]["name"]: e["handler"]
            for e in NATIVE_TOOLS if e["handler"] is not None
        }
        # 🔒 자가검증: 노출 스키마 ↔ 핸들러 정합성 (yielding 도구는 특수 처리로 허용)
        for e in NATIVE_TOOLS:
            nm = e["schema"]["function"]["name"]
            if e["handler"] is None and nm not in _YIELDING_TOOLS:
                logger.error(f"[ToolRuntime] '{nm}' 핸들러 누락 — dispatch 불가")

    def native_schemas(self):
        """모델에 노출할 네이티브 도구 OpenAI 스키마 리스트 (단일 진실 출처)."""
        return [e["schema"] for e in NATIVE_TOOLS]

    def _tool_default(self, tool_name, fb_limit, fb_max):
        """정책 tool_policy.tool_defaults.<tool>의 (limit, max). 실패 시 내장 폴백."""
        try:
            tp = getattr(self.brain, "_tool_policy", None)
            cfg = ((tp() if callable(tp) else {}).get("tool_defaults") or {}).get(tool_name) or {}
            lim = int(cfg.get("limit") or fb_limit)
            mx = int(cfg.get("max") or fb_max)
            return max(1, lim), max(1, mx)
        except Exception:  # noqa: BLE001
            return fb_limit, fb_max

    def curated_native_schemas(self, flags):
        """의도별 큐레이션. flags={'hugin','os','delegation','obsidian_only'}(bool).
        해당 의도가 아니면 상황형 도구(브라우저/OS제어/위임)를 숨겨 목록을 작게 유지한다.
        대화·검색·기억·지식 도구는 항상 노출.
        obsidian_only=True: Ollama 전용 경량 모드 — obsidian_write 1개만 반환."""
        # [🔱 Ollama 경량 모드] 정책이 정한 도구 묶음만 노출 (미디어/3D/유튜브 등).
        # gemma4 등 tools 지원 로컬 모델이 'vrm 모델 찾아줘'에 환각 답변하던 문제의 해법:
        # 도구를 아예 안 실어주던 기존 분기 대신 최소 도구를 실어준다.
        # only_tools: 이름 목록(정책 tool_policy.intents.<intent>.tools). media_only는 별칭.
        only = flags.get("only_tools")
        if flags.get("media_only") and not only:
            only = ["search_media", "search_3d_model", "search_youtube"]
        if only:
            keep = set(only)
            out = [e["schema"] for e in NATIVE_TOOLS
                   if e["schema"]["function"]["name"] in keep]
            if out:
                return out
            # 정책 오타 등으로 매칭 0이면 안전 폴백(도구 0개로 환각 유발 방지)
            keep = {"search_media", "search_3d_model", "search_youtube"}
            return [e["schema"] for e in NATIVE_TOOLS
                    if e["schema"]["function"]["name"] in keep]

        # [🔱 Ollama 경량 모드] 약한 로컬 모델 오버로드 방지
        if flags.get("obsidian_only"):
            for e in NATIVE_TOOLS:
                nm = e["schema"]["function"]["name"]
                if nm == "obsidian_write":
                    return [e["schema"]]
            return []

        out = []
        for e in NATIVE_TOOLS:
            nm = e["schema"]["function"]["name"]
            if nm in _HUGIN_TOOLS and not flags.get("hugin"):
                continue
            if nm in _OS_TOOLS and not flags.get("os"):
                continue
            if nm in _DELEGATION_TOOLS and not flags.get("delegation"):
                continue
            out.append(e["schema"])
        return out

    def _handle_search_media(self, args, ctx):
        query = (str(args.get("query") or "").strip()) or ctx.user_prompt
        media_type = str(args.get("media_type") or "photo").strip().lower()
        provider = str(args.get("provider") or "pexels").strip().lower()
        try:
            limit = max(1, min(20, int(args.get("limit", 8))))
        except (TypeError, ValueError):
            limit = 8
        if media_type not in ("photo", "video", "gif"):
            media_type = "photo"
        if provider not in ("pexels", "pixabay"):
            provider = "pexels"

        aegis_url, hits, err = self.brain.search_media_hits(query, media_type, provider, limit)
        if err:
            return err
        if not hits:
            return f"'{query}' {media_type} 검색 결과가 없다. 다른 검색어를 짧게 제안하라."

        items = []
        for h in hits:
            dl = h.get("download", "")
            if isinstance(dl, str) and dl.startswith("/") and aegis_url:
                dl = aegis_url + dl
            items.append({
                "source": h.get("source") or provider,
                "type": h.get("type") or media_type,
                "url": h.get("url") or "",
                # 🖼️ 미리보기 썸네일 — 비디오는 url이 mp4라 이미지로 못 그리므로
                # 별도 preview(썸네일 이미지)를 반드시 함께 전달한다.
                "preview": h.get("preview") or "",
                "download": dl,
                "page": h.get("page") or "",
                "title": h.get("title") or h.get("alt") or h.get("source") or "Media",
            })

        payload = {"query": query, "media_type": media_type, "items": items}
        yield f"\n\n```imperial-media\n{json.dumps(payload, ensure_ascii=False)}\n```\n\n"
        return f"미디어 {len(items)}개를 카드 그리드로 화면에 표시했다. 결과를 다시 나열하지 말고 한 줄로 마무리하라."

    def _handle_search_youtube(self, args, ctx):
        """🎥 실제 유튜브 검색 → imperial-youtube 카드 블록을 스트림에 직접 방출."""
        query = (str(args.get("query") or "").strip()) or ctx.user_prompt
        try:
            limit = max(1, min(20, int(args.get("limit", 6))))
        except (TypeError, ValueError):
            limit = 6
        result = self.brain.youtube_search(query, limit)
        # youtube_search 반환문에서 imperial-youtube 블록만 추출해 카드로 방출한다.
        m = re.search(r"```imperial-youtube[\s\S]*?```", result or "")
        if not m:
            # 키 미설정/결과없음/오류 → 모델에 그대로 전달(사용자에게 사유 안내)
            return result or "유튜브 검색 결과가 없습니다."
        yield f"\n\n{m.group(0)}\n\n"
        return f"유튜브 영상을 카드로 화면에 표시했다. 결과를 다시 나열하지 말고 한 줄로 마무리하라."

    # ──────────────────────────────────────────────
    # 비-yield 도구 디스패치 (ask_user 제외, 순수 함수)
    # ──────────────────────────────────────────────
    def dispatch(self, func_name, args, ctx):
        b = self.brain
        handler = self._handlers.get(func_name)
        if handler is not None:
            return handler(b, args, ctx)
        if b.coding.handles(func_name):
            return b.coding.execute(func_name, args)
        if func_name.startswith("mcp_"):
            return self._dispatch_mcp_tool(func_name, args, timeout=150, user_prompt=ctx.user_prompt)
        return f"Error: Tool '{func_name}' not found in Imperial Arsenal."

    # ──────────────────────────────────────────────
    # 🌐 MCP 도구 호출 가드 (필수 인자 사전 검증·자동 보정)
    # ──────────────────────────────────────────────
    def _dispatch_mcp_tool(self, func_name, args, timeout=150, user_prompt=None):
        """경량 모델이 `query` 등 필수 string 인자를 비운 채 MCP 도구를 호출하면
        그대로 보내 -32602로 떨어지고 같은 실수를 반복한다. 2단계로 방어한다.
          1) 누락 필수 string 1개 + 유저 발화가 있으면 그걸로 채워 재시도(검색 케이스 구제)
          2) 보정 불가 시 '무엇이 비었는지 + 예시'를 결과로 돌려 모델 자가 교정 유도
        """
        _mc = self._get_mcp_client() if self._get_mcp_client else None
        if not _mc:
            return "Error: MCP unavailable."
        if not isinstance(args, dict):
            args = {}
        try:
            body = func_name[4:]  # 'mcp_' 제거 → '<server>_<tool>'
            target = None
            for t in _mc.get_raw_tools():
                if f"{t.get('server')}_{t.get('name')}" == body:
                    target = t
                    break
            if target:
                schema = target.get("input_schema") or {}
                required = schema.get("required", []) or []
                props = schema.get("properties", {}) or {}
                missing = [f for f in required
                           if args.get(f) is None or (isinstance(args.get(f), str) and not args.get(f).strip())]
                if missing:
                    # 1) 자동 보정
                    if user_prompt and user_prompt.strip() and len(missing) == 1:
                        f0 = missing[0]
                        if (props.get(f0, {}) or {}).get("type", "string") == "string":
                            # [🔱 Playwright Safety] browser_navigate는 유저 발화로 보정하면 URL이 깨지므로 제외
                            if func_name == "mcp_playwright_browser_navigate" and f0 == "url":
                                return "ERROR: browser_navigate를 호출하려면 유효한 http/https URL 주소가 필요합니다. 검색 쿼리 문자열을 직접 넣지 마십시오."
                            args[f0] = user_prompt.strip()[:400]
                            logger.info(f"[MCP Guard] '{func_name}' 누락 인자 '{f0}'를 유저 발화로 자동 보정 → 재시도")
                            return _mc.call_tool_by_gemini_name(func_name, args, timeout=timeout)
                    # 2) 재시도 유도
                    example = {}
                    for field in required:
                        ptype = (props.get(field, {}) or {}).get("type", "string")
                        example[field] = (0 if ptype in ("number", "integer")
                                          else True if ptype == "boolean"
                                          else ["..."] if ptype == "array" else "...")
                    logger.warning(f"[MCP Guard] '{func_name}' 필수 인자 누락 {missing} — MCP 호출 차단")
                    return (f"ERROR: 도구 '{func_name}' 호출에 필수 인자 {missing}가 비어 있어 "
                            f"MCP 서버로 보내지 않았습니다. 올바른 JSON 인자를 채워 다시 호출하세요. "
                            f"예시: {json.dumps(example, ensure_ascii=False)}")
        except Exception as _ve:
            logger.warning(f"[MCP Guard] 인자 검증 경고({func_name}): {_ve}")
        return _mc.call_tool_by_gemini_name(func_name, args, timeout=timeout)

    # ──────────────────────────────────────────────
    # 🛟🍽 텍스트→ask 카드 복원 파서 (순수 함수)
    # ──────────────────────────────────────────────
    @staticmethod
    def _extract_fake_ask_payload(text):
        """모델이 ask_user 대신 본문에 쓴 {"question":...,"options":[...]} JSON을 찾아낸다.
        question+options(2개 이상)를 가진 첫 JSON을 반환, 없으면 None."""
        for m in re.finditer(r"\{[^{}]*\}", text or ""):
            try:
                obj = json.loads(m.group(0))
            except Exception:
                continue
            if (isinstance(obj, dict)
                    and str(obj.get("question", "")).strip()
                    and isinstance(obj.get("options"), list)
                    and len(obj["options"]) >= 2):
                return obj
        return None

    @staticmethod
    def _extract_ask_from_markdown(text):
        """모델이 옵션을 인자로 못 넣고 본문에 ①②③/번호/불릿 목록으로 쓴 경우,
        그 목록을 파싱해 {question, options}로 복원한다(2~6개). 아니면 None."""
        if not text:
            return None
        lines = text.splitlines()
        opt_re = re.compile(r'^(?:[' + _CIRCLED_NUM + r']|\(?\d{1,2}[.)])\s*(.+)$')
        bullet_re = re.compile(r'^[\-\*•]\s*')
        options = []
        for ln in lines:
            s = ln.strip()
            if not s:
                continue
            s = bullet_re.sub('', s)            # 선행 불릿 제거
            m = opt_re.match(s)                  # 그 뒤 ①.. 또는 1./1) 번호 필수
            if not m:
                continue
            label = re.sub(r'\*\*|__|`', '', m.group(1))  # 마크다운 강조 제거
            label = label.strip(' :：-—·').strip()
            if label:
                options.append(label[:40])
        if len(options) < 2:
            return None
        options = options[:6]
        question = ""
        for ln in lines:
            s = ln.strip()
            if not s.endswith('?'):
                continue
            if opt_re.match(bullet_re.sub('', s)):
                continue
            question = re.sub(r'^[^\w가-힣"\']+', '', s).strip()
            break
        if not question:
            question = "어떤 것으로 할까요?"
        return {"question": question[:200], "options": options}

    # ──────────────────────────────────────────────
    # ask_user (UI 카드 방출 + 대기) — 제너레이터, 결과 문자열 return
    # ──────────────────────────────────────────────
    def _handle_ask_user(self, args, ctx):
        ctx.ask_attempted = True
        if not ctx.interactive:
            # 🚷 비대화형(동기 릴레이) 경로 — 선택지 UI가 제때 뜰 수 없음
            return ("UNAVAILABLE: 이 채널은 비대화형(동기 릴레이) 경로라 ask_user를 "
                    "사용할 수 없다. 기다리지 말고, 답변 본문에 선택지를 텍스트로 제시한 "
                    "뒤 턴을 끝내라. 유저의 답은 다음 메시지로 온다.")

        # 🛡️ Ask Guard: 빈 인자({})면 질문/옵션이 빈 깨진 카드가 떠버린다 → 차단
        _q = str(args.get("question", "")).strip()
        _opts = [str(o).strip() for o in (args.get("options") or []) if str(o).strip()]
        if not _q or len(_opts) < 2:
            ctx.blocked_ask += 1
            logger.warning(f"[Ask Guard] ask_user 인자 불충분 "
                           f"(question={bool(_q)}, options={len(_opts)}) — 빈 카드 차단 ({ctx.blocked_ask}회)")
            if ctx.blocked_ask >= 2:
                # 🛡️ Circuit-Breaker: 두 번째 빈 호출 → 도구 비활성화로 무한 루프 차단
                ctx.force_no_tools = True
                # 🔇 다음(도구 없는) 응답은 카드로 변환될 수 있으므로 라이브 스트리밍 보류
                ctx.suppress_stream = True
                logger.warning("[Ask Guard] 빈 ask_user 2회 — 서킷 브레이커 작동, "
                               "도구 비활성화 후 텍스트 마무리 강제")
                return ("STOP: ask_user를 빈 인자로 두 번 호출했다. 이제 도구는 비활성화된다. "
                        "더 이상 도구를 부르지 말고, 유저에게 무엇을 도와줄지 본문 텍스트로 "
                        "직접 한국어로 물어보며 턴을 끝내라.")
            return ("ERROR: ask_user에는 비어있지 않은 'question'과 2~6개의 'options'가 "
                    "반드시 필요하다. 인자가 비어 빈 카드가 뜨므로 호출을 취소했다. 예시: "
                    '{"question": "어떤 방식으로 진행할까요?", "options": ["A안", "B안"]}. '
                    "정말 제시할 선택지가 없다면 도구 대신 본문 텍스트로 질문하고 턴을 끝내라.")

        # 정상: UI 카드 방출 → 응답/타임아웃 대기 (제너레이터 위임)
        return (yield from self._emit_ask_card(_q, _opts, args.get("timeout_seconds", 120), ctx))

    def _emit_ask_card(self, question, options, timeout_seconds, ctx=None):
        _ask = self.brain._begin_ask_user({
            "question": question, "options": options, "timeout_seconds": timeout_seconds,
        })
        yield f"\n\n```imperial-ask\n{json.dumps(_ask, ensure_ascii=False)}\n```\n\n"
        _ans = self.brain._wait_ask_user(_ask["ask_id"], _ask["timeout_seconds"])
        if _ans is None:
            yield f"⏱️ 선택 대기 시간({_ask['timeout_seconds']}초)이 초과되어 질문이 자동 취소되었습니다.\n\n"
            return ("TIMEOUT: 유저가 제한 시간 내에 선택하지 않아 질문이 취소되었다. "
                    "보수적인 기본값으로 마무리하거나, 결정이 필수라면 답변 말미에 질문을 남기고 턴을 끝내라.")
        if _ans == "__CANCELLED__":
            yield "🚫 질문이 취소되었습니다 (유저가 다른 작업으로 전환).\n\n"
            return ("CANCELLED: 유저가 질문을 취소하고 다른 작업으로 전환했다. "
                    "선택을 기다리지 말고 보수적으로 짧게 마무리하라.")
        routed = yield from self._route_search_menu_answer(question, options, _ans, ctx)
        if routed is not None:
            return routed
        return f"유저의 선택: {_ans}"

    def _route_search_menu_answer(self, question, options, answer, ctx=None):
        q = str(question or "")
        ans = str(answer or "")
        opt_list = [str(o or "") for o in (options or [])]
        joined = " ".join([q, ans] + opt_list)
        is_search_menu = (
            any(k in joined for k in ("검색", "search", "네이버", "다음", "구글", "웹검색", "이지스", "미디어", "유튜브", "youtube", "YouTube"))
            or (2 <= len(opt_list) <= 5 and ans in opt_list)
        )
        if not is_search_menu:
            return None

        user_prompt = getattr(ctx, "user_prompt", "") if ctx is not None else ""
        query = self._clean_search_query(user_prompt)
        selected_index = opt_list.index(ans) if ans in opt_list else -1
        # 🎥 유튜브 옵션이 끼면 옵션 개수가 5개가 되어 인덱스가 밀린다 →
        # 인덱스 기반 라우팅은 정확히 4지선다(유튜브 없음)일 때만 쓰고, 그 외엔 키워드로 판별.
        four_way_menu = (len(opt_list) == 4)

        # 🎥 유튜브: 선택했거나, 원문에 유튜브가 명시된 경우 곧바로 유튜브 검색
        if (
            "유튜브" in ans
            or "youtube" in ans.lower()
            or (selected_index < 0 and ("유튜브" in user_prompt or "youtube" in (user_prompt or "").lower()))
        ):
            yield f"\n\n🎥 유튜브 검색으로 '{query}' 정찰을 시작합니다.\n\n"
            # imperial-youtube 카드 블록을 채팅 스트림에 직접 방출(텍스트로 모델에 넘기면 카드가 안 뜬다)
            return (yield from self._handle_search_youtube({
                "query": query,
                "limit": self._extract_limit(user_prompt, 8),
            }, ctx))

        # 🖼️ Pexels / Pixabay: 미디어 소스 선택 메뉴(예: 동영상 출처 선택)
        if "pexels" in ans.lower() or "pixabay" in ans.lower():
            provider = "pixabay" if "pixabay" in ans.lower() else "pexels"
            media_type = self._infer_media_type(user_prompt, ans)
            yield f"\n\n🖼️ {provider} {media_type} 검색으로 '{query}' 정찰을 시작합니다.\n\n"
            return (yield from self._handle_search_media({
                "query": query,
                "media_type": media_type,
                "provider": provider,
                "limit": self._extract_limit(user_prompt, 8),
            }, ctx))

        if (four_way_menu and selected_index == 0) or "네이버" in ans:
            yield f"\n\n🔎 네이버 웹검색으로 '{query}' 정찰을 시작합니다.\n\n"
            return self.brain.web_search(query, engine="naver")

        if (four_way_menu and selected_index == 1) or "다음" in ans:
            yield f"\n\n🔎 다음 웹검색으로 '{query}' 정찰을 시작합니다.\n\n"
            return self.brain.web_search(query, engine="daum")

        if (four_way_menu and selected_index == 2) or "구글" in ans or any(k in ans for k in ("google", "Google", "GOOGLE")):
            yield f"\n\n🔎 구글 웹검색으로 '{query}' 정찰을 시작합니다.\n\n"
            return self.brain.web_search(query, engine="google")

        if (four_way_menu and selected_index == 3) or any(k in ans for k in ("이지스", "미디어", "이미지", "GIF", "gif", "동영상", "3D", "3d")):
            if self._looks_3d(user_prompt):
                yield f"\n\n🧊 이지스 3D 모델검색으로 '{query}' 정찰을 시작합니다.\n\n"
                return (yield from self._handle_search_3d({
                    "query": query,
                    "limit": self._extract_limit(user_prompt, 8),
                }, ctx))
            media_type = self._infer_media_type(user_prompt, ans)
            yield f"\n\n🖼️ 이지스 미디어검색으로 '{query}' 정찰을 시작합니다.\n\n"
            return (yield from self._handle_search_media({
                "query": query,
                "media_type": media_type,
                "provider": "pexels",
                "limit": self._extract_limit(user_prompt, 8),
            }, ctx))

        return None

    @staticmethod
    def _looks_3d(text):
        t = (text or "").lower()
        return any(k in t for k in ("3d", "glb", "모델", "에셋", "asset"))

    @staticmethod
    def _infer_media_type(prompt, answer=""):
        p = (prompt or "").lower()
        if any(k in p for k in ("gif", "움짤", "짤")):
            return "gif"
        if any(k in p for k in ("video", "동영상", "영상")):
            return "video"
        return "photo"

    @staticmethod
    def _extract_limit(text, default=8):
        m = re.search(r"(\d{1,2})\s*(?:개|장|건|매|results?|items?)", text or "", re.IGNORECASE)
        if not m:
            return default
        try:
            return max(1, min(20, int(m.group(1))))
        except Exception:
            return default

    @staticmethod
    def _clean_search_query(text):
        q = (text or "").strip()
        q = re.sub(r"\d{1,2}\s*(?:개|장|건|매)\s*", " ", q)
        q = re.sub(r"(찾아\s*주세요|찾아줘|검색\s*해주세요|검색해줘|구해줘|주세요|해줘)", " ", q)
        q = re.sub(r"(이미지|사진|그림|동영상|영상|gif|GIF|움짤|짤|3D\s*모델|3d\s*모델|모델|에셋)", " ", q)
        q = re.sub(r"\s+", " ", q).strip()
        return q or (text or "").strip() or "검색"

    # ──────────────────────────────────────────────
    # 🧊 search_3d_model (UI 카드 방출) — 제너레이터, 결과 문자열 return
    # ──────────────────────────────────────────────
    def _handle_search_3d(self, args, ctx):
        # 빈 query면 유저 발화로 자동 보정 (약한 모델 방어)
        query = (str(args.get("query") or "").strip()) or ctx.user_prompt
        d_limit, d_max = self._tool_default("search_3d_model", 8, 20)
        try:
            limit = int(args.get("limit") or 0)
        except (TypeError, ValueError):
            limit = 0
        if limit <= 0:
            # 모델이 limit을 빼먹으면 유저 발화의 개수("2개" 등)로 보정 (gemma4:12b 방어)
            m = re.search(r"(\d{1,2})\s*(?:개|장|건|매)", ctx.user_prompt or "")
            limit = int(m.group(1)) if m else d_limit
        limit = max(1, min(d_max, limit))
        source = str(args.get("source") or "both").strip().lower()
        if source not in ("pixabay", "sketchfab", "both"):
            source = "both"
        models = self.brain.search_3d_model(query, limit, source=source)
        if not models:
            return (f"'{query}' 3D 모델 검색 결과가 없다. 다른 검색어를 짧게 제안하라. "
                    "(URL이나 사이트를 지어내지 마라.)")
        # 🧊 imperial-3d 블록을 스트림에 직접 방출 → 헤임달이 프리뷰+뷰어 카드로 렌더
        payload = {"query": query, "models": models}
        yield f"\n\n```imperial-3d\n{json.dumps(payload, ensure_ascii=False)}\n```\n\n"
        return (f"3D 모델 {len(models)}개를 프리뷰+3D 뷰어 카드로 화면에 표시했다. "
                "결과를 다시 나열하지 말고, 한 줄로 짧게만 마무리하라.")

    # ──────────────────────────────────────────────
    # 도구 호출 일괄 실행 (제너레이터: UI 블록/디버그 액션을 yield)
    # ──────────────────────────────────────────────
    def run_tool_calls(self, tool_calls, full_content, messages, ctx):
        ctx.used_tools = True
        messages.append({"role": "assistant", "content": full_content, "tool_calls": tool_calls})

        for tc in tool_calls:
            func_name = tc["function"]["name"]
            yield f"[TOOL_USE: {func_name}]"
            tc_id = tc["id"]
            result = "Error: Unknown neural tool malfunction."
            args = {}
            try:
                try:
                    args = json.loads(tc["function"]["arguments"])
                except Exception:
                    args = {}
                logger.info(f"[{ctx.display_name} Tool] {func_name}({args})")

                if func_name == "ask_user":
                    result = yield from self._handle_ask_user(args, ctx)
                elif func_name == "search_media":
                    result = yield from self._handle_search_media(args, ctx)
                elif func_name == "search_3d_model":
                    result = yield from self._handle_search_3d(args, ctx)
                elif func_name == "search_youtube":
                    result = yield from self._handle_search_youtube(args, ctx)
                else:
                    result = self.dispatch(func_name, args, ctx)
            except Exception as tool_err:
                result = f"Error: tool execution failed: {tool_err}"
                logger.error(f"Neural Tool Error ({func_name}): {tool_err}")

            # 🩺 로컬 직업의 도구 실패를 Vault 경험 Inbox에 축적한다.
            if "error" in str(result).lower():
                try:
                    cfg = getattr(self.brain, "dynamic_config", None) or {}
                    if str(cfg.get("job_source") or "").lower() == "vault":
                        from obsidian_health import record_tool_failure
                        record_tool_failure(
                            func_name, result, prompt=ctx.prompt,
                            guardian=ctx.display_name, job=cfg.get("job", ""),
                        )
                except Exception as health_err:
                    logger.debug("[Obsidian Health] 실패 경험 기록 생략: %s", health_err)

            messages.append({
                "role": "tool", "tool_call_id": tc_id, "name": func_name, "content": str(result),
            })

            # 🗂️ obsidian_write 실제 호출 추적 — 재촉(nudge) 시 중복 작성 방지용
            if func_name == "obsidian_write" and "Error" not in str(result):
                ctx.obsidian_written = True

            # 🐞 도구 액션 로그 (Debug ON이면 채팅에도, OFF면 서버 로그만)
            if self._summarize:
                line = self._summarize(ctx.display_name, func_name, result)
            else:
                line = f"[{ctx.display_name} Action: {func_name}] {str(result)[:200]}"
            if self._is_debug():
                yield f"\n\n⚡ {line}\n\n"
            else:
                logger.info(f"⚡ {line}")

            # 🕸️ hugin 정찰 결과는 web_recon 메모리에 저장
            if func_name in ("hugin_navigate", "hugin_type") and "Error" not in str(result):
                try:
                    self.brain._store_web_recon(func_name, args, result)
                except Exception as mem_err:
                    logger.warning(f"Failed to store web recon memory: {mem_err}")

    # ──────────────────────────────────────────────
    # tool_calls가 없을 때(모델이 텍스트만 출력) 구제 처리
    # 반환: "continue"(루프 계속) | "break"(턴 종료)
    # ──────────────────────────────────────────────
    def handle_no_tool_calls(self, full_content, messages, ctx):
        # 🔱 Empty-Final Guard: 도구만 쓰고 최종 답변이 비면 한 번 재촉
        if ctx.used_tools and not full_content.strip() and not ctx.nudged_final:
            ctx.nudged_final = True
            messages.append({
                "role": "user",
                "content": "방금 실행한 도구 결과를 바탕으로, 원래 질문에 대한 최종 답변을 "
                           "한국어로 간결하게 작성하라. 도구를 다시 호출하지 마라.",
            })
            return "continue"

        # 🛟 Fake-Ask / 🍽 Markdown-Ask Rescue
        if ctx.interactive and not ctx.ask_rescued:
            _fake = self._extract_fake_ask_payload(full_content)
            # ask를 시도한 턴에서만 본문 목록(①②③ 등)을 카드로 변환 (오작동 방지)
            if not _fake and ctx.ask_attempted:
                _fake = self._extract_ask_from_markdown(full_content)
                if _fake:
                    logger.info(f"[Markdown-Ask Rescue] 본문 목록 → ask_user 카드 변환 "
                                f"(옵션 {len(_fake.get('options', []))}개)")
            if _fake:
                ctx.ask_rescued = True
                # 🔇 카드로 변환 성공 → 보류했던 원본 텍스트는 버린다(2중 표시 방지)
                ctx.suppress_stream = False
                logger.info("[Fake-Ask Rescue] 본문을 실제 ask_user 플로우로 변환")
                _ask = self.brain._begin_ask_user({
                    "question": _fake.get("question", ""),
                    "options": _fake.get("options") or [],
                    "timeout_seconds": _fake.get("timeout_seconds", 120),
                })
                yield f"\n\n```imperial-ask\n{json.dumps(_ask, ensure_ascii=False)}\n```\n\n"
                _ans = self.brain._wait_ask_user(_ask["ask_id"], _ask["timeout_seconds"])
                if _ans is None:
                    yield f"⏱️ 선택 대기 시간({_ask['timeout_seconds']}초)이 초과되어 질문이 자동 취소되었습니다.\n\n"
                    return "break"
                if _ans == "__CANCELLED__":
                    yield "🚫 질문이 취소되었습니다 (유저가 다른 작업으로 전환).\n\n"
                    return "break"
                messages.append({"role": "assistant", "content": full_content})
                messages.append({
                    "role": "user",
                    "content": (f"(선택지 질문에 대한 유저의 선택: {_ans}) 이 선택에 맞춰 이어서 "
                                "답하라. 같은 질문을 반복하거나 선택지 JSON을 다시 출력하지 마라."),
                })
                return "continue"

        # 🔇 카드로 변환되지 않았는데 스트림을 보류했다면, 보류한 텍스트를 지금 한 번 출력(flush)
        if ctx.suppress_stream and full_content.strip():
            yield full_content
            ctx.suppress_stream = False
        return "break"

    # ──────────────────────────────────────────────
    # 🗂️ 옵시디언 약속-미이행 재촉: 모델이 "생성/작성하겠습니다"라고 말만 하고
    #    obsidian_write를 호출하지 않으면, 실제 호출하도록 한 번 더 재촉한다.
    # 반환: "continue"(재촉 후 루프 계속) | "break"(재촉 불가/소진 → 종료)
    # ──────────────────────────────────────────────
    _PROMISE_PHRASES = ("하겠습니다", "할게요", "할까요", "만들겠", "생성하겠", "작성하겠",
                        "추가하겠", "저장하겠", "정리하겠", "기록하겠", "생성할", "작성할", "만들어 드리겠",
                        # 🔱 '올려줘' 도메인 어휘 추가 — "옵시디언에 ~ 올려드리겠습니다" 등
                        "올려드리", "올려 드리", "올리겠", "올려둘", "올려두겠", "업로드", "반영하겠")

    def nudge_obsidian_write(self, full_content, messages, ctx):
        if not getattr(ctx, "expect_obsidian_write", False):
            return "break"
        if ctx.promise_nudges >= 2:
            return "break"
        # 🗂️ 이미 이 턴에 obsidian_write 를 실제 호출했으면 재촉 불필요(중복 작성 방지).
        if getattr(ctx, "obsidian_written", False):
            return "break"
        text = (full_content or "")
        has_promise = any(ph in text for ph in self._PROMISE_PHRASES)
        # 🔱 [Fix] 약속 어구가 없어도, 이번 턴에 다른 도구(검색 등)를 쓰고 글은 안 쓴 경우엔
        #   재촉한다. "날씨 검색 → 보고만 하고 옵시디언에 안 올림" 패턴을 잡기 위함.
        #   (도구도 안 쓰고 약속도 없는 순수 대화는 재촉하지 않음 → 오작동 방지.)
        if not has_promise and not getattr(ctx, "used_tools", False):
            return "break"
        ctx.promise_nudges += 1
        messages.append({"role": "assistant", "content": full_content})
        messages.append({
            "role": "user",
            "content": ("말로만 하지 말고 지금 즉시 obsidian_write 도구를 실제로 호출하라. "
                        "여러 노트가 필요하면 obsidian_write를 노트 1개당 한 번씩 여러 번 호출하라. "
                        "각 content에는 네가 직접 작성한 전체 마크다운(# 제목, [[백링크]], #태그)을 넣어라. "
                        "도구를 호출하지 않으면 아무것도 만들어지지 않는다. 호출 전에는 '완료'라고 말하지 마라."),
        })
        logger.info(f"🗂️ [Obsidian Nudge] 약속-미이행 감지 — obsidian_write 강제 재촉 ({ctx.promise_nudges}/2)")
        return "continue"