shadow_brain_core/brain/tool_runtime.py

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

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

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

from imperial_i18n import t  # 🌐 사용자 노출 문자열 로컬라이즈 (X-Client-Locale)

logger = logging.getLogger("JarvisBrain")

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

# 🛰️ [TOOL_USE 마커 상세] 도구 호출 마커에 동봉할 핵심 인자 1개 — 우선순위 키.
# CLI/헤임달이 "code_edit_file|lib/foo.dart"처럼 무엇을 건드리는지 실시간 표시한다.
_TOOL_DETAIL_KEYS = ("path", "file_path", "command", "pattern", "query",
                     "url", "name", "guardian", "title")


def _tool_marker_detail(args) -> str:
    """도구 인자에서 표시용 핵심 값 추출 (48자 컷, 마커 파손 문자 제거)."""
    if not isinstance(args, dict):
        return ""
    for k in _TOOL_DETAIL_KEYS:
        v = args.get(k)
        if v:
            s = " ".join(str(v).split())  # 개행/연속 공백 정리
            if len(s) > 48:
                s = s[:47] + "…"
            return s.replace("]", ")").replace("[", "(").replace("|", "/")
    return ""


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)
#   새 도구는 이 리스트에만 추가하면 스키마/핸들러가 동시에 등록된다 → 드리프트 불가.
# ──────────────────────────────────────────────────────────────
def _delegate_handler(b, a, ctx):
    """🤝 위임 핸들러 — 약한 모델이 guardian 인자를 비워 보내면(실관측)
    유저 발화에서 수호자명을 찾아 자동 보정한다 (search_media 보정과 동일 패턴)."""
    g = (a.get("guardian") or "").strip()
    if not g:
        up = (getattr(ctx, "user_prompt", "") or "").lower()
        candidates = list(getattr(b, "_DELEGATE_ALIASES", {}).items())
        candidates += [(k, k) for k in getattr(b, "_DELEGATE_ROUTES", {})]
        for alias, canon in candidates:
            if alias in up:
                g = canon
                break
    return b.delegate_to_guardian(g, a.get("task", ""), a.get("context", ""))


def _shorts_api(path, payload=None):
    """쇼츠 스튜디오 로컬 API 자기호출 (blueprints/shorts_routes.py)."""
    import json as _json
    import os as _os
    import urllib.request as _rq
    key_path = _os.path.expanduser(_os.path.join("~", ".taemini", "shadow_local_token"))
    key = open(key_path).read().strip() if _os.path.exists(key_path) else ""
    req = _rq.Request(
        "http://127.0.0.1:18700" + path,
        data=_json.dumps(payload).encode() if payload is not None else None,
        headers={"Content-Type": "application/json", "X-Taemini-Key": key})
    return _json.load(_rq.urlopen(req, timeout=60))


def _create_shorts_handler(b, a, ctx):
    lines = a.get("lines") or []
    if len(lines) != 7:
        return ("실패: 내레이션 대사는 정확히 7줄이어야 합니다 (①훅 ②AI의 헛소리 ③전개 "
                "④학원/주제 소개 ⑤시스템 소개 ⑥성적/결과 ⑦킬러 멘트). 7줄을 채워 다시 호출하세요.")
    payload = {"lines": lines}
    if a.get("bgm_tags"):
        payload["bgmTags"] = a["bgm_tags"]
    voices = {}
    if a.get("narrator_voice"):
        voices["narrator"] = a["narrator_voice"]
    if a.get("ai_voice"):
        voices["ai"] = a["ai_voice"]
    if voices:
        payload["voices"] = voices
    try:
        r = _shorts_api("/api/shadow/shorts/create", payload)
    except Exception as e:
        return f"쇼츠 파이프라인 호출 실패: {e}"
    jid = r.get("job_id")
    if not jid:
        return f"쇼츠 생성 시작 실패: {str(r)[:200]}"
    return (f"쇼츠 제작이 시작되었습니다. job_id={jid}. 제작은 몇 분 걸립니다 — "
            f"shorts_video_status 도구로 job_id를 넣어 진행 상태를 확인하고, "
            f"status가 done이 되면 output 경로를 사용자에게 알려주세요.")


def _shorts_status_handler(b, a, ctx):
    jid = (a.get("job_id") or "").strip()
    if not jid:
        return "실패: job_id가 비었습니다. create_shorts_video가 돌려준 job_id를 넣으세요."
    try:
        r = _shorts_api(f"/api/shadow/shorts/status/{jid}")
    except Exception as e:
        return f"상태 조회 실패: {e}"
    status = r.get("status")
    if status == "done":
        return f"완성! 파일: {r.get('output')} (이 경로를 사용자에게 알려주세요)"
    if status == "error":
        return f"제작 실패: {r.get('error')}"
    return f"진행 중 — 단계: {r.get('step')} ({r.get('detail', '')}). 잠시 후 다시 확인하세요."


# ── 🩺 헤임달 개발자 직업 — 진단·에스컬레이션 도구 ──────────────────────────
# 자가 치유 루프: 진단(health/log) → 기지 레시피로 해결 시도 →
# 실패 시 진단 패키지 생성 + 스승(자비스) 채널로 에스컬레이션.

_LOG_WHITELIST = ["shadow_server.log", "daemon_stdout.log", "shadowbrain_boot.log"]


def _shadow_home():
    import os as _os
    return _os.path.expanduser(_os.path.join("~", "TaeminGames", "ShadowBrain"))


def _repo_root():
    import os as _os
    return _os.path.dirname(_os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))))


def _newest_log():
    import os as _os
    best, best_m = None, 0.0
    for name in _LOG_WHITELIST:
        p = _os.path.join(_shadow_home(), name)
        try:
            m = _os.path.getmtime(p)
        except OSError:
            continue
        if m > best_m:
            best, best_m = p, m
    return best


def _log_tail(path, n, needle=""):
    try:
        with open(path, "r", encoding="utf-8", errors="ignore") as f:
            lines = f.readlines()
    except OSError as e:
        return [f"(로그 읽기 실패: {e})"]
    if needle:
        low = needle.lower()
        lines = [l for l in lines if low in l.lower()]
    return [l.rstrip("\n") for l in lines[-n:]]


def _health_snapshot():
    import os as _os
    import socket as _sock
    ports = [("쉐도우 브레인(18700)", 18700), ("이지스 장고(18701)", 18701),
             ("클로드 채널(8790)", 8790), ("이리스(4000)", 4000), ("바하무트(3002)", 3002)]
    out = []
    for name, port in ports:
        try:
            with _sock.create_connection(("127.0.0.1", port), timeout=1):
                out.append(f"✅ {name} 응답")
        except OSError:
            out.append(f"❌ {name} 무응답")
    try:
        ver = open(_os.path.join(_repo_root(), "VERSION"), encoding="utf-8").read().strip()
    except OSError:
        ver = "?"
    log = _newest_log()
    err_n = 0
    if log:
        err_n = sum(1 for l in _log_tail(log, 200)
                    if any(k in l for k in ("Error", "ERROR", "Traceback", "Exception")))
    out.append(f"ℹ️ 제국 버전 {ver} · 최근 로그 200줄 중 오류 흔적 {err_n}건"
               f" ({_os.path.basename(log) if log else '로그 파일 없음'})")
    return out


def _health_check_handler(b, a, ctx):
    lines = _health_snapshot()
    return ("🩺 헤임달 시스템 건강 진단:\n" + "\n".join(lines) +
            "\n\n무응답 포트가 있으면 그 서비스가 꺼진 것이 1순위 용의자입니다. "
            "read_daemon_log 도구로 오류 로그를 확인하고, 직업 노트의 자가 치유 레시피에 "
            "같은 증상이 있으면 그 처방을 따르세요. 없으면 escalate_bug_report로 스승에게 보고하세요.")


def _daemon_log_handler(b, a, ctx):
    try:
        n = max(10, min(int(a.get("lines") or 80), 300))
    except (TypeError, ValueError):
        n = 80
    needle = (a.get("filter") or "").strip()
    log = _newest_log()
    if not log:
        return "로그 파일이 없습니다 (~/TaeminGames/ShadowBrain/*.log)."
    body = "\n".join(_log_tail(log, n, needle))
    import os as _os
    head = f"📜 {_os.path.basename(log)} 마지막 {n}줄" + (f" (필터: {needle})" if needle else "")
    return f"{head}:\n{body[-6000:]}"


def _escalate_handler(b, a, ctx):
    """진단 패키지 생성 → 클로드 채널 자동 연결 → 스승에게 발송, 답을 그대로 반환."""
    import datetime as _dt
    import os as _os
    import subprocess as _sp
    summary = (a.get("summary") or "").strip()
    if not summary:
        return "실패: summary(버그 한 줄 요약)가 비었습니다."
    symptoms = (a.get("symptoms") or "").strip()
    tried = (a.get("tried") or "").strip()
    ts = _dt.datetime.now().strftime("%Y%m%d_%H%M%S")
    rep_dir = _os.path.join(_shadow_home(), "bug_reports")
    _os.makedirs(rep_dir, exist_ok=True)
    try:
        commits = _sp.run(["git", "log", "--oneline", "-5"], cwd=_repo_root(),
                          capture_output=True, text=True, timeout=15).stdout.strip()
    except Exception:
        commits = "(git 조회 실패)"
    log = _newest_log()
    tail = "\n".join(_log_tail(log, 120)) if log else "(로그 없음)"
    report = (f"# 🐛 버그 진단 패키지 {ts}\n\n"
              f"## 요약\n{summary}\n\n"
              f"## 증상 (재현 방법 포함)\n{symptoms or '(미기재)'}\n\n"
              f"## 시도한 조치\n{tried or '(없음)'}\n\n"
              f"## 시스템 건강 스냅샷\n" + "\n".join(_health_snapshot()) + "\n\n"
              f"## 최근 커밋\n```\n{commits}\n```\n\n"
              f"## 데몬 로그 tail 120줄\n```\n{tail[-8000:]}\n```\n")
    rep_path = _os.path.join(rep_dir, f"bug_{ts}.md")
    with open(rep_path, "w", encoding="utf-8") as f:
        f.write(report)
    # 채널 자동 사전 연결 (죽어 있으면 START_CLAUDE_CHANNEL 자동 기동 — 방송 관제와 동일 배관)
    try:
        prep = _shorts_api("/api/shadow/broadcast/notify/prepare", {"target": "claude"})
    except Exception as e:
        return (f"진단 패키지는 저장했습니다: {rep_path}\n"
                f"하지만 스승 채널 연결 확인에 실패했습니다({e}). 마왕님께 패키지 경로를 보고하세요.")
    if prep.get("status") != "ready":
        return (f"진단 패키지는 저장했습니다: {rep_path}\n"
                f"스승(클로드) 채널을 열지 못했습니다(상태: {prep.get('status')}). "
                f"마왕님께 패키지 경로를 보고하고 채널 연결을 요청하세요.")
    msg = (f"[버그 관제] 헤임달 개발자(로컬 수호자)가 자력으로 해결하지 못한 버그를 보고합니다.\n"
           f"요약: {summary}\n진단 패키지: {rep_path}\n"
           f"heimdall-bug-doctor 스킬 절차대로: 패키지를 읽고 원인을 진단·해결한 뒤, "
           f"해결법을 소형 모델용 레시피로 압축해 헤임달 개발자 직업 노트의 "
           f"'자가 치유 레시피' 섹션에 장착해 주세요.")
    try:
        import json as _json
        import urllib.request as _rq
        key_path = _os.path.expanduser(_os.path.join("~", ".taemini", "shadow_local_token"))
        key = open(key_path).read().strip() if _os.path.exists(key_path) else ""
        req = _rq.Request("http://127.0.0.1:18700/api/shadow/claude-code/send",
                          data=_json.dumps({"message": msg}).encode(),
                          headers={"Content-Type": "application/json", "X-Taemini-Key": key})
        r = _json.load(_rq.urlopen(req, timeout=600))
    except Exception as e:
        return (f"진단 패키지 저장 + 채널 연결까지 됐지만 발송 중 오류: {e}\n"
                f"패키지: {rep_path} — 마왕님께 경로를 보고하세요.")
    reply = (r.get("content") or "").strip()
    if r.get("status") == "SUCCESS" and reply:
        return (f"📨 스승(자비스)에게 보고 완료. 진단 패키지: {rep_path}\n\n"
                f"스승의 답변:\n{reply[:3000]}")
    return (f"발송했지만 답변을 받지 못했습니다(상태: {r.get('status')}). "
            f"패키지: {rep_path} — 나중에 스승의 처방이 직업 노트에 장착될 수 있습니다.")


# ── 🎙️ 방송 PD 직업 — 대본 저장·관제 로그 분석 도구 ─────────────────────────

def _save_script_handler(b, a, ctx):
    name = (a.get("name") or "").strip()
    cast = a.get("cast") or []
    lines = a.get("lines") or []
    if not name:
        return "실패: name(대본 이름)이 비었습니다."
    if not cast:
        return "실패: cast가 비었습니다. [{guardian, voice}] 형식으로 등장 수호자를 넣으세요."
    if len(lines) < 2:
        return "실패: lines가 2줄 미만입니다. 오프닝+본편 최소 2줄을 쓰세요."
    first = lines[0] if isinstance(lines[0], dict) else {}
    if first.get("live") or first.get("talk"):
        return ("실패: 첫 줄이 라이브/대담입니다. 작법 수칙 — 오프닝 첫 줄은 반드시 각본"
                "(가릴 앞 대사가 없어 '생각 중'이 시청자에게 노출됨). 첫 줄을 각본 대사로 바꾸세요.")
    payload = {"name": name, "title": name, "cast": cast, "lines": lines,
               "memo": f"# {name}\n\n방송 PD 직업(로컬 수호자)이 작성한 대본. "
                       f"헤임달 라이브 방송 앱 → 불러오기 → '{name}' 으로 무대에 올립니다."}
    try:
        r = _shorts_api("/api/shadow/broadcast/script", payload)
    except Exception as e:
        return f"대본 저장 실패: {e}"
    if r.get("status") != "success":
        return f"대본 저장 실패: {str(r)[:200]}"
    return (f"📜 대본 저장 완료: {r.get('path')}\n"
            f"마왕님께 '헤임달 라이브 방송 앱에서 불러오기 → {name} 을 선택해 방송 시작을 "
            f"눌러 주세요'라고 안내하세요.")


def _list_scripts_handler(b, a, ctx):
    try:
        r = _shorts_api("/api/shadow/broadcast/scripts")
    except Exception as e:
        return f"대본 목록 조회 실패: {e}"
    names = [s.get("name") for s in (r.get("scripts") or [])]
    if not names:
        return "저장된 방송 대본이 없습니다."
    return "저장된 방송 대본 목록:\n- " + "\n- ".join(str(n) for n in names)


def _broadcast_log_handler(b, a, ctx):
    import glob as _glob
    import json as _json
    import os as _os
    work = _os.path.expanduser(_os.path.join("~", "TaeminGames", "ShadowBrain", "broadcast"))
    title = (a.get("title") or "").strip()
    if not title:
        try:
            topics = sorted([d for d in _os.listdir(work)
                             if _os.path.isdir(_os.path.join(work, d))])
        except OSError:
            topics = []
        return ("방송 주제 폴더 목록 (title로 지정):\n- " + "\n- ".join(topics)) if topics \
            else "방송 주제 폴더가 없습니다 (아직 녹화된 방송 없음)."
    logs = sorted(_glob.glob(_os.path.join(work, title, "log_*.json")))
    if not logs:
        return f"'{title}' 폴더에 관제 로그(log_*.json)가 없습니다."
    try:
        with open(logs[-1], encoding="utf-8") as f:
            log = _json.load(f)
    except Exception as e:
        return f"로그 읽기 실패: {e}"
    ag = log.get("aggregates") or {}
    gen_n = ag.get("gen_utterances", ag.get("live_cues"))
    out = [f"📊 '{title}' 최신 관제 로그 ({_os.path.basename(logs[-1])}):",
           f"- 큐 {ag.get('cues')}개 (라이브 {ag.get('live_cues')}"
           f" · 대담 발화 {ag.get('talk_turns', 0)})",
           f"- 침묵(생각 중) 노출 총 {ag.get('visible_thinking_total_ms')}ms"
           f" (최대 {ag.get('visible_thinking_max_ms')}ms) ← 0에 가까울수록 좋은 방송",
           f"- 브레인 평균 {ag.get('brain_ms_avg')}ms · TTS 평균 {ag.get('tts_ms_avg')}ms",
           f"- 사전 준비 적중 {ag.get('prefetch_hit')}/{gen_n}"]
    worst = None
    for c in (log.get("cues") or []):
        v = c.get("visible_thinking_ms") or 0
        if worst is None or v > (worst.get("visible_thinking_ms") or 0):
            worst = c
    if worst and (worst.get("visible_thinking_ms") or 0) > 500:
        out.append(f"- 최악 큐: #{worst.get('index')} ({worst.get('type')}) — "
                   f"생각 중 {worst.get('visible_thinking_ms')}ms 노출. "
                   f"첫 큐가 live면 오프닝 각본 한 줄을 앞에 넣는 것이 처방입니다.")
    return "\n".join(out)


NATIVE_TOOLS = [
    {
        # 🎬 쇼츠 스튜디오 — TTS(수퍼토닉)+BGM(ACE-Step)+연출(Remotion) 로컬 파이프라인
        "schema": _fn("create_shorts_video",
            "Creates a complete 30s vertical Shorts video (1080x1920, Korean TTS narration + AI-composed "
            "BGM + animated scenes) from a 7-line script, fully on this PC. MUST be used when the user asks "
            "to make a shorts/ad/promo video (쇼츠/광고 영상 만들어줘). The template is a comedy academy-ad: "
            "line1=hook(narrator), line2=the AI's silly wrong answer(AI voice), line3=escalation, "
            "line4=academy/topic intro(star-wars crawl scene), line5=system intro, line6=results, "
            "line7=killer closing line. Write ALL 7 lines yourself in Korean (short, punchy, funny), then "
            "call this tool. It returns a job_id — then poll with shorts_video_status until done.",
            {"lines": {"type": "array", "items": {"type": "string"},
                       "description": "Exactly 7 Korean narration lines in template order"},
             "bgm_tags": {"type": "string",
                          "description": "Optional English music style tags (default: comedy sitcom orchestral)"},
             "narrator_voice": {"type": "string", "description": "Supertonic voice M1-M5/F1-F5 (default M1)"},
             "ai_voice": {"type": "string", "description": "Supertonic voice for line2 (default F2)"}},
            ["lines"]),
        "handler": _create_shorts_handler,
    },
    {
        "schema": _fn("shorts_video_status",
            "Checks progress of a Shorts video job started by create_shorts_video. Steps: tts → bgm → "
            "render → done. When done it returns the output mp4 path — report that path to the user.",
            {"job_id": {"type": "string", "description": "job_id returned by create_shorts_video"}},
            ["job_id"]),
        "handler": _shorts_status_handler,
    },
    {
        # 🩺 헤임달 개발자 — 시스템 건강 진단
        "schema": _fn("heimdall_health_check",
            "Checks Heimdall/Imperial system health on this PC: core service ports (Shadow Brain 18700, "
            "Aegis 18701, Claude channel 8790, Iris 4000, Bahamut 3002), Imperial VERSION, and error "
            "traces in the newest daemon log. Use FIRST when the user reports a Heimdall bug/error "
            "(버그/오류/에러/안 돼요) to identify which service is the suspect.",
            {}, []),
        "handler": _health_check_handler,
    },
    {
        "schema": _fn("read_daemon_log",
            "Reads the tail of the newest Shadow Brain daemon log (fixed whitelist under "
            "~/TaeminGames/ShadowBrain). Use after heimdall_health_check to inspect actual error "
            "messages (Traceback, ERROR lines) before deciding a fix or escalating.",
            {"lines": {"type": "integer", "description": "How many tail lines (10-300, default 80)"},
             "filter": {"type": "string", "description": "Optional substring filter, e.g. 'Error' or a route name"}},
            []),
        "handler": _daemon_log_handler,
    },
    {
        "schema": _fn("escalate_bug_report",
            "Escalates a Heimdall bug you could NOT solve yourself to the master teacher (Jarvis/Claude). "
            "Builds a diagnostic package (summary + symptoms + health snapshot + log tail + recent commits) "
            "under ~/TaeminGames/ShadowBrain/bug_reports/, auto-opens the teacher channel if closed, sends "
            "the report, and returns the teacher's answer. The teacher will also install a self-healing "
            "recipe into the Heimdall-developer job so you can fix this class of bug yourself next time. "
            "Only escalate AFTER trying heimdall_health_check + read_daemon_log + existing recipes.",
            {"summary": {"type": "string", "description": "One-line bug summary in Korean"},
             "symptoms": {"type": "string", "description": "Symptoms + how to reproduce (what the user saw)"},
             "tried": {"type": "string", "description": "What you already tried and the results"}},
            ["summary"]),
        "handler": _escalate_handler,
    },
    {
        # 🎙️ 방송 PD — 대본을 Vault(Broadcasts/)에 저장, 헤임달 라이브 방송 앱이 불러온다
        "schema": _fn("save_broadcast_script",
            "Saves a Heimdall live-broadcast script to the Vault so the user can load it in the "
            "라이브 방송 app and press start. Use when the user asks you to write/plan a broadcast "
            "(방송 대본 써줘, 방송 기획해줘). Writing rules: short spoken Korean lines (~15 chars per "
            "breath), the FIRST line must be scripted (never live), live lines carry a specific "
            "question in 'text'. Voices: Supertonic M1-M5/F1-F5.",
            {"name": {"type": "string", "description": "Script name (Korean OK, becomes Broadcasts/<name>.md)"},
             "cast": {"type": "array", "items": {"type": "object"},
                      "description": "[{guardian:'shadow_brain', voice:'F3'}] — real guardian ids only"},
             "lines": {"type": "array", "items": {"type": "object"},
                       "description": "[{speaker:'shadow_brain', text:'대사', emotion:'happy|excited|love|sad|angry|thinking|fear|neutral', live:false, exit:false}]"}},
            ["name", "cast", "lines"]),
        "handler": _save_script_handler,
    },
    {
        "schema": _fn("list_broadcast_scripts",
            "Lists saved Heimdall broadcast scripts (Vault Broadcasts/). Use before writing a new "
            "script to avoid overwriting, or when the user asks what broadcasts exist.",
            {}, []),
        "handler": _list_scripts_handler,
    },
    {
        "schema": _fn("read_broadcast_log",
            "Reads the newest broadcast telemetry log for a topic folder and summarizes quality "
            "(visible-thinking exposure, brain/TTS latency, prefetch hits) with an improvement tip. "
            "Call with no title to list topic folders. Use after a broadcast to report quality "
            "(방송 어땠어? 로그 분석해줘).",
            {"title": {"type": "string", "description": "Broadcast topic folder name (empty = list topics)"}},
            []),
        "handler": _broadcast_log_handler,
    },
    {
        # 🎨 [Character Forge] 오리지널 캐릭터 실생성 — 루나 검증 파이프라인 원자 도구
        #   (character_artist 직업의 실행 수단. 픽셀 작업은 전부 character_forge.py 안에 가둠)
        "schema": _fn("character_generate",
            "Actually generates an original Imperial character image on this Mac (ComfyUI pipeline: "
            "verified prompt formula + green-screen generation -> lucida background matting -> despill "
            "-> transparent PNG with numeric quality gates). MUST be used when the user asks to create/"
            "generate/draw a character or avatar (캐릭터/아바타 만들어줘, 생성해줘, 그려줘) — do not just "
            "explain the recipe, call this tool. mode='explore' (default, Z-Image Turbo, ~1 min) for "
            "drafts and seed hunting; mode='final' (Krea2 Turbo + retroanime LoRA, several minutes, "
            "needs 36GB free RAM) only for the single confirmed final image — reuse the chosen seed. "
            "It starts/stops ComfyUI itself and refuses when memory is low (memory safety first). "
            "Returns the saved transparent PNG path + gate numbers: report them honestly and show the "
            "image to 마왕님 — aesthetic judgement belongs to 마왕님, not you.",
            {"description": {"type": "string", "description":
                "Character appearance in English: hair, eyes, outfit, accessories, expression "
                "(e.g. 'long silver-white hair, lavender eyes, white and gold dress'). "
                "Do NOT include background/pose words — the verified formula handles those."},
             "mode": {"type": "string", "enum": ["explore", "final"], "description":
                "explore = quick draft (default) | final = top-quality confirmed single image"},
             "seed": {"type": "integer", "description":
                "Optional. Reuse a draft's seed to keep the same character for mode='final'"}},
            ["description"]),
        # 🎨 yielding tool — handler=None, run_tool_calls가 imperial-media 카드(생성 이미지)를
        #    스트림에 직접 방출해 채팅창에서 바로 보인다 (_handle_character_generate)
        "handler": None,
    },
    {
        # 🎵 [Taemini Music] 음악 실생성 — composer(작곡가) 직업의 실행 수단.
        #   헤파이스토스 중계(WanGP)로 생성 → R2 정본화 → 공개 URL 반환.
        "schema": _fn("music_generate",
            "Actually generates a music track on the Imperial fleet GPU (Hephaestus/WanGP) and returns "
            "a public URL. MUST be used when the user asks to create/compose music, a song, BGM, "
            "ambience, or a sound effect (음악/노래/BGM 만들어줘, 작곡해줘) — do not just describe a "
            "genre, call this tool. Write 'style' in English (genre, instruments, mood, tempo). "
            "Engines: default 'ace_step_v1_5_xl_turbo_lm_4b' (full songs, supports lyrics, up to 240s); "
            "'stable_audio3_small' (ambience/texture, up to 60s, faster); 'stable_audio3_small_sfx' "
            "(short sound effects, up to 20s). For a vocal song pass 'lyrics' ([Verse]/[Chorus] tags) "
            "and instrumental=false. Generation takes 1-4 minutes — report the returned URL honestly; "
            "musical taste judgement belongs to 마왕님, not you.",
            {"style": {"type": "string", "description":
                "Music style prompt in English: genre, instruments, mood, tempo "
                "(e.g. 'epic orchestral, majestic brass, heroic, slow build')"},
             "duration_seconds": {"type": "integer", "description":
                "Track length in seconds (default 60; clamped to the engine's max)"},
             "model_type": {"type": "string", "description":
                "Fleet engine: ace_step_v1_5_xl_turbo_lm_4b (default) | ace_step_v1_5_turbo_lm_1_7b | "
                "stable_audio3_small | stable_audio3_small_sfx. With engine=mlx: the MLX model id the tool listed"},
             "lyrics": {"type": "string", "description":
                "Optional lyrics with [Verse]/[Chorus] tags (ACE-Step engines only)"},
             "instrumental": {"type": "boolean", "description":
                "true = no vocals (default). Set false only when lyrics are given"},
             "engine": {"type": "string", "enum": ["auto", "remote", "mlx"], "description":
                "auto (default) = fleet GPU, then this PC's MLX-Serve music models. "
                "mlx = this PC's MLX-Serve directly — MLX Core 26.10+ loads models on demand, so an installed "
                "model is usable right away (e.g. ACE-Step 8-bit: a 20s loop in a few seconds; the server answers "
                "503 with needed/free GB if memory is short). Pass model_type = the MLX model id "
                "(e.g. ddalcu/ace-step-1.5-xl-turbo-8bit). For video BGM make short 20-30s loops, not the full length"}},
            ["style"]),
        "handler": lambda b, a, ctx: b.music_generate(
            style=a.get("style") or (ctx.user_prompt if ctx else ""),
            lyrics=a.get("lyrics", ""),
            instrumental=a.get("instrumental", True),
            duration_seconds=a.get("duration_seconds", 60),
            model_type=a.get("model_type", ""),
            engine=a.get("engine", "auto")),
    },
    {
        # 🤝 [Imperial Orchestration] 동료 수호자 위임 — 기획(claude)→코딩(codex)→테스트(minimax)→검증(claude)
        "schema": _fn("delegate_to_guardian",
            "Delegates a task to a fellow local guardian AI and returns their result. Guardians: "
            "'codex' (coding, implementation), 'minimax' (coding, testing), 'claude' (planning, review, verification), "
            "'hermes' (general agent), 'lmstudio' (local LLM). Use when the user asks to have another guardian do "
            "something (e.g. '코덱스에게 시켜줘'), or to orchestrate multi-step work: plan with claude → implement "
            "with codex → test with minimax → verify with claude. Pass a self-contained task description; the "
            "guardian cannot see this conversation, so include all needed context in 'task'/'context'. "
            "Each delegation may take minutes — report the returned result honestly.",
            {"guardian": {"type": "string", "description": "codex | minimax | hermes | claude | lmstudio"},
             "task": {"type": "string", "description": "Self-contained task instruction for the guardian (Korean OK)"},
             "context": {"type": "string", "description": "Optional extra context (prior results, file paths, constraints)"}},
            ["guardian", "task"]),
        "handler": _delegate_handler,
    },
    {
        "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("get_weather",
            "Fetches REAL current weather and forecast (Open-Meteo, deterministic, no key) for a city. "
            "MUST be used for ANY weather question (오늘 날씨, 내일 비 와?, 기온, 예보). "
            "NEVER answer weather from memory or web search — every number in your answer must come "
            "from this tool's result. If the tool fails, say the lookup failed; do not invent numbers.",
            {"location": {"type": "string", "description": "City name in Korean or English (default 서울)"},
             "days": {"type": "integer", "description": "Forecast days 1-7 (default 3)"}},
            []),
        # 빈 인자 허용 — 약한 모델이 location을 빼먹어도 서울 기본값으로 동작
        "handler": lambda b, a, ctx: b.get_weather(
            str(a.get("location") or "서울"), int(a.get("days") or 3)),
    },
    {
        "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")),
    },
]

# 🎬 모션그래픽 디자이너 도구 6종(상황형 — 모션 의도·직업 허용 때만 노출). 실행은 motion_studio.py.
try:
    from .tool_runtime_motion import MOTION_TOOLS, MOTION_TOOL_NAMES as _MOTION_TOOLS, wants_motion
except ImportError:  # 패키지 밖 단독 로드(verify_tool_runtime) — 같은 폴더 파일을 직접 읽는다
    import importlib.util as _ilu
    import os as _os_m
    _spec_m = _ilu.spec_from_file_location(
        "tool_runtime_motion_std", _os_m.path.join(_os_m.path.dirname(__file__), "tool_runtime_motion.py"))
    _mod_m = _ilu.module_from_spec(_spec_m)
    _spec_m.loader.exec_module(_mod_m)
    MOTION_TOOLS, _MOTION_TOOLS, wants_motion = _mod_m.MOTION_TOOLS, _mod_m.MOTION_TOOL_NAMES, _mod_m.wants_motion
NATIVE_TOOLS += MOTION_TOOLS

# 🗂️ 옵시디언 지식 레이어 및 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}")

# ──────────────────────────────────────────────────────────────
# 🧩 유저 도구 (user_tools/) — 배포판 커스텀 확장점 (2026-07-17)
#
# 유저가 자기 네이티브 도구를 '우리 소스 밖'(유저 영역)에 두고 쓰게 한다.
# 우리 트리 안에 코드를 넣으면 업데이트가 덮어쓰는 문제의 해법 —
# ComfyUI custom_nodes / 옵시디언 플러그인과 같은 패턴.
#
# 계약(파일당 도구 1개): 모듈 상수 TOOL = {name, description, params, required}
#   + 함수 run(args: dict) -> str.  '_' 로 시작하는 파일은 무시(비활성).
# 안전판: 이름 규칙·내장 도구와 충돌 거부·파일별 예외 격리(깨진 도구는 건너뜀,
#   데몬은 죽지 않음). 반영은 데몬 재시작 시.
# ──────────────────────────────────────────────────────────────

_USER_TOOL_NAME_RE = re.compile(r"^[a-z][a-z0-9_]{2,40}$")

# 🧾 유저 도구 계약 버전 — run(args)->str + TOOL dict 형태가 v1.
#   우리가 계약을 바꾸는 배포를 하면 이 숫자를 올리고, 구/신 버전 도구에
#   명확한 안내를 남긴다 ("소스가 바뀌면 유저가 자기 플러그인을 언제 고쳐야
#   하는지"를 기계적으로 알리는 장치).
USER_TOOLS_CONTRACT = 1
_USER_TOOLS_CONTRACT_MIN = 1  # 이보다 낮은 계약은 더 이상 지원하지 않음

# 🩺 마지막 스캔의 실패 기록 — 상태 API(/api/shadow/user-tools/status)가 노출해
#   "조용히 안 실리는" 유저 도구를 가시화한다. [{file, error}]
USER_TOOL_ERRORS: list = []

_USER_TOOLS_README = """# 🧩 Shadow Brain 유저 도구 (user_tools)

이 폴더의 `*.py` 파일은 쉐도우 브레인 부팅 시 자동으로 도구로 등록됩니다.
업데이트가 이 폴더를 건드리지 않으므로, 커스텀 도구는 여기에 두세요.

## 계약 (파일 하나 = 도구 하나)

```python
TOOL = {
    "name": "my_tool",              # 소문자/숫자/_ (3~41자), 내장 도구와 중복 불가
    "description": "이 도구가 무엇을 하고 언제 써야 하는지 — AI가 읽는 설명서",
    "params": {"q": {"type": "string", "description": "질의"}},
    "required": ["q"],
}

def run(args: dict) -> str:         # 실행부 — 결과를 문자열로 반환
    return f"결과: {args.get('q')}"
```

## 규칙
- `_` 로 시작하는 파일은 무시됩니다 (`_draft.py` 로 비활성화).
- 반영에는 쉐도우 브레인 재시작이 필요합니다.
- 깨진 도구는 경고 로그를 남기고 건너뜁니다 (데몬은 죽지 않음).
- 이 폴더의 코드는 본인 책임으로 실행됩니다.
- 직업(Jobs) 규칙에 "이 도구를 호출하라"를 적으면 로컬 AI가 사용법을 배웁니다.

## 계약 버전 (업데이트 호환)
- 현재 계약: **v1** (위 형태). TOOL 에 `"contract": 1` 을 명시할 수 있습니다(생략 시 1).
- 쉐도우 브레인 업데이트로 계약이 바뀌면, 구버전 도구는 로드 시 갱신 안내와 함께
  건너뛰어집니다 — 이 README 의 최신 계약에 맞춰 도구를 고쳐 주세요.
- 도구 상태 확인: `GET http://localhost:18700/api/shadow/user-tools/status`

## 직업(Jobs)과 연결
직업 노트 frontmatter 에 `user_tools: [내도구이름]` 을 선언하면:
- 직업 관리자에 🧩 플러그인 배지가 표시됩니다.
- 도구가 삭제/고장난 채 직업이 장착되면, AI 에게 "그 도구는 지금 없다"는 경고가
  자동 주입되어 헛호출 대신 사용자에게 설치 필요를 알립니다.

## 시작하기
옆의 `_example_disk_free.py` 가 완전한 예제입니다 — 열어 보고,
파일명의 `_` 를 지우면 실제 도구로 켜집니다.
"""

# 🎁 첫 실행 때 함께 생성되는 예제 플러그인 — '_' 접두사라 비활성(학습용).
#   유저가 접두사를 지우는 순간부터 실제 도구가 된다.
_USER_TOOLS_EXAMPLE = '''"""예제 유저 도구 — 이 컴퓨터의 디스크 남은 용량을 알려준다.

파일명 앞의 '_' 를 지우면(disk_free.py) 다음 재시작부터 실제 도구로 등록된다.
자기 도구를 만들 때 이 파일을 복사해 고치면 된다.
"""

TOOL = {
    "name": "disk_free",
    "description": (
        "Checks free disk space on this computer. Use when the user asks "
        "about remaining disk space or storage (디스크/저장공간 남은 용량)."
    ),
    "params": {},        # 인자가 필요하면: {"q": {"type": "string", "description": "..."}}
    "required": [],      # 필수 인자 이름 목록
    "contract": 1,       # 계약 버전 (README 참고, 생략 시 1)
}


def run(args: dict) -> str:
    import shutil
    total, used, free = shutil.disk_usage("/")
    gb = 1024 ** 3
    return (f"디스크: 전체 {total/gb:.0f}GB · 사용 {used/gb:.0f}GB · "
            f"남음 {free/gb:.0f}GB ({100*free/total:.0f}% 여유)")
'''


def _user_tools_dir():
    import os
    return (os.environ.get("SHADOW_USER_TOOLS_DIR")
            or os.path.expanduser("~/TaeminGames/ShadowBrain/user_tools"))


def load_user_tools(existing_names, directory=None):
    """user_tools/*.py 스캔 → 검증 통과한 도구 엔트리 목록 반환 (파일별 예외 격리)."""
    import os
    import importlib.util
    d = directory or _user_tools_dir()
    try:
        first_run = not os.path.isdir(d)
        os.makedirs(d, exist_ok=True)
        readme = os.path.join(d, "README.md")
        if not os.path.exists(readme):
            with open(readme, "w", encoding="utf-8") as f:
                f.write(_USER_TOOLS_README)
        # 🎁 첫 실행(폴더 신규 생성)에만 비활성 예제를 심는다 — 유저가 지우면 재생성 안 함
        if first_run:
            with open(os.path.join(d, "_example_disk_free.py"), "w", encoding="utf-8") as f:
                f.write(_USER_TOOLS_EXAMPLE)
    except Exception as e:  # noqa: BLE001
        logger.warning(f"[UserTools] 폴더 준비 실패({d}): {e}")
        return []
    entries = []
    taken = set(existing_names)
    USER_TOOL_ERRORS.clear()
    for fn in sorted(os.listdir(d)):
        if not fn.endswith(".py") or fn.startswith("_"):
            continue
        path = os.path.join(d, fn)
        try:
            spec = importlib.util.spec_from_file_location(f"shadow_user_tools.{fn[:-3]}", path)
            mod = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(mod)
            meta = getattr(mod, "TOOL", None)
            run = getattr(mod, "run", None)
            if not isinstance(meta, dict) or not callable(run):
                raise ValueError("TOOL dict 와 run(args) 함수가 필요합니다")
            name = str(meta.get("name") or "").strip()
            if not _USER_TOOL_NAME_RE.match(name):
                raise ValueError(f"이름 규칙 위반: '{name}' (소문자/숫자/_ 3~41자)")
            if name in taken:
                raise ValueError(f"이름 충돌: '{name}' 은 이미 존재하는 도구입니다")
            contract = int(meta.get("contract") or 1)
            if contract > USER_TOOLS_CONTRACT:
                raise ValueError(
                    f"계약 v{contract} 도구 — 이 쉐도우 브레인은 v{USER_TOOLS_CONTRACT}까지 지원. "
                    "쉐도우 브레인을 업데이트하세요")
            if contract < _USER_TOOLS_CONTRACT_MIN:
                raise ValueError(
                    f"계약 v{contract} 은 지원 종료(최소 v{_USER_TOOLS_CONTRACT_MIN}) — "
                    "user_tools/README.md 의 최신 계약으로 도구를 갱신하세요")
            desc = str(meta.get("description") or "").strip()
            if not desc:
                raise ValueError("description 이 비어 있습니다")
            entries.append({
                "schema": _fn(name, desc,
                              meta.get("params") or {},
                              list(meta.get("required") or [])),
                # 유저 도구는 brain 접근 없이 인자만 받는다 (계약 단순화 + 격리)
                "handler": (lambda m: lambda b, a, ctx: str(m.run(dict(a or {})))[:20000])(mod),
                "user_tool": True,
                "source": path,
            })
            taken.add(name)
            logger.info(f"[UserTools] 등록: {name} ← {fn}")
        except Exception as e:  # noqa: BLE001
            logger.warning(f"[UserTools] '{fn}' 로드 실패(건너뜀): {e}")
            USER_TOOL_ERRORS.append({"file": fn, "error": str(e)[:300]})
    return entries


try:
    USER_TOOLS = load_user_tools(
        {e["schema"]["function"]["name"] for e in NATIVE_TOOLS})
except Exception as _ut_e:  # noqa: BLE001
    logger.warning(f"[UserTools] 스캔 건너뜀: {_ut_e}")
    USER_TOOLS = []
NATIVE_TOOLS += USER_TOOLS

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

# 🧹 상황형 도구: 해당 의도일 때만 노출 (그 외엔 목록을 작게 유지 → 약한 모델 정확도↑)
_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 user_tool_schemas(self):
        """🧩 user_tools/ 유저 도구 스키마 — 설치 자체가 사용 의도라 로컬 경로
        (Ollama 큐레이션)에 항상 곁들여 싣는다. 없으면 빈 리스트."""
        return [e["schema"] for e in NATIVE_TOOLS if e.get("user_tool")]

    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 ensure_declared_tools(self, tools, cfg):
        """🎖️ 직업이 `allowed_tools`(tool_mode=whitelist)로 선언한 도구를 **항상** 실어준다.

        ⚠️ [2026-07-31 실사고] Ollama 경로는 도구를 **의도로만** 붙인다(`if _wants_wx:` 등).
        그래서 기상 캐스터를 장착하고 「안녕」이라고 하면 날씨 의도가 없어 `get_weather` 가
        한 개도 안 실렸고(has_tools=False), 모델은 직업 규칙대로 "지금은 날씨를 조회할 수
        없습니다"라고 정직하게 답했다 — 게으름이 아니라 **도구를 뺏긴 것**이었다.

        직업이 명시한 도구는 그 수호자의 **선언된 능력**이다. 화이트리스트는 상한(ceiling)
        으로만 쓰여 왔지만(2026-07-20 추가) 하한(floor)이기도 해야 한다. 의도 추측이 선언을
        지우면 폴백 금지 헌장("직업 설정은 문자 그대로 집행")에 어긋난다.

        whitelist 모드일 때만 동작하고(무선언 직업은 기존 큐레이션 유지), 이미 실린 도구는
        중복 추가하지 않는다. 채워 넣으면 로그를 남겨 진단 가능하게 한다.
        """
        if str((cfg or {}).get("tool_mode") or "all").lower() != "whitelist":
            return tools
        declared = [str(t).strip() for t in ((cfg or {}).get("allowed_tools") or [])
                    if str(t).strip()]
        if not declared:
            return tools
        try:
            floor = self.curated_native_schemas({"only_tools": declared}) or []
        except Exception as err:  # noqa: BLE001
            logger.debug(f"[직업 도구 하한] 스키마 조회 실패(무시): {err}")
            return tools
        have = {s["function"]["name"] for s in (tools or [])}
        add = [s for s in floor if s["function"]["name"] not in have]
        if not add:
            return tools
        logger.info(
            "🎖️ [직업 도구 하한] 선언 도구를 의도와 무관하게 노출: %s",
            [s["function"]["name"] for s in add],
        )
        return (tools or []) + add

    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
            if nm in _MOTION_TOOLS and not flags.get("motion"):
                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_character_generate(self, args, ctx):
        """🎨 캐릭터 실생성 + 채팅 인라인 표시 — 생성 이미지를 imperial-media 카드로
        스트림에 직접 방출한다 (강아지 이미지 검색과 같은 렌더 경로). 이미지 바이트는
        옵시디언 첨부 API(/api/obsidian/attachment)가 서빙한다 (2026-07-17)."""
        desc = (str(args.get("description") or "").strip()) or ctx.user_prompt
        mode = str(args.get("mode") or "explore")
        seed = args.get("seed")
        text, res = self.brain.character_generate_meta(
            desc, mode, seed, user_prompt=ctx.user_prompt or "")
        vault_path = (res or {}).get("vault_path") or ""
        if res and res.get("ok") and vault_path:
            from urllib.parse import quote
            url = ("http://localhost:18700/api/obsidian/attachment?path="
                   + quote(vault_path))
            payload = {"query": desc[:60], "media_type": "photo", "items": [{
                "source": "character_forge",
                "type": "photo",
                "url": url,
                "preview": url,
                "download": url,
                "page": "",
                "title": f"생성 캐릭터 · seed {res.get('seed')} ({res.get('mode')})",
            }]}
            yield f"\n\n```imperial-media\n{json.dumps(payload, ensure_ascii=False)}\n```\n\n"
        return text

    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
            # [🔱 Playwright Safety] wait_for 호출 시 time/text/textGone 인자가 모두 없으면 time: 2000ms 자동 주입 (에러 방지)
            if "playwright" in func_name and "wait" in func_name:
                if not args.get("time") and not args.get("text") and not args.get("textGone"):
                    args["time"] = 2000
                    logger.info(f"[MCP Guard] '{func_name}' 인자 누락 자동 보정 → time: 2000ms 주입")

            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 t("⏱️ 선택 대기 시간({sec}초)이 초과되어 질문이 자동 취소되었습니다.", sec=_ask['timeout_seconds']) + "\n\n"
            return ("TIMEOUT: 유저가 제한 시간 내에 선택하지 않아 질문이 취소되었다. "
                    "보수적인 기본값으로 마무리하거나, 결정이 필수라면 답변 말미에 질문을 남기고 턴을 끝내라.")
        if _ans == "__CANCELLED__":
            yield t("🚫 질문이 취소되었습니다 (유저가 다른 작업으로 전환).") + "\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 "\n\n" + t("🎥 유튜브 검색으로 '{query}' 정찰을 시작합니다.", query=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 "\n\n" + t("🖼️ {provider} {mediaType} 검색으로 '{query}' 정찰을 시작합니다.", provider=provider, mediaType=media_type, query=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 "\n\n" + t("🔎 네이버 웹검색으로 '{query}' 정찰을 시작합니다.", query=query) + "\n\n"
            return self.brain.web_search(query, engine="naver")

        if (four_way_menu and selected_index == 1) or "다음" in ans:
            yield "\n\n" + t("🔎 다음 웹검색으로 '{query}' 정찰을 시작합니다.", query=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 "\n\n" + t("🔎 구글 웹검색으로 '{query}' 정찰을 시작합니다.", query=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 "\n\n" + t("🧊 이지스 3D 모델검색으로 '{query}' 정찰을 시작합니다.", query=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 "\n\n" + t("🖼️ 이지스 미디어검색으로 '{query}' 정찰을 시작합니다.", query=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
        # 🛡️ [필수 도구 추적] 어떤 도구가 실제 호출됐는지 이름을 남긴다
        #   (require_tool_call 재촉 판정용 — used_tools bool 만으론 부족).
        if not hasattr(ctx, "used_tool_names"):
            ctx.used_tool_names = set()
        for _tc in tool_calls:
            try:
                ctx.used_tool_names.add(str(_tc["function"]["name"]))
            except Exception:
                pass
        messages.append({"role": "assistant", "content": full_content, "tool_calls": tool_calls})

        _any_tool_error = False  # 📊 이번 배치에 도구 에러가 있었는지(스킬 오류율 귀속용)
        _images = []  # 🖼️ ToolResult.images — 이미지를 받는 두뇌(ctx.accept_images)에만 넘긴다
        # 🎖️ [실행 관문] 직업/임무 whitelist 는 노출(스키마 큐레이션)뿐 아니라 **실행
        # 시점**에도 집행한다. 노출만 막으면 가짜 텍스트 도구호출 구제·재촉 등 우회
        # 경로가 뚫린다 (2026-07-20 실사고: has_tools=False 인데 구제 파서가
        # character_generate 를 실행해 보고 임무가 생성으로 빗나감). 베이스 공용
        # 도구(_WHITELIST_BASE_TOOLS)는 통과 — 일을 아예 못 하게 되는 것 방지.
        _wl_cfg = getattr(self.brain, "dynamic_config", None) or {}
        _wl_on = str(_wl_cfg.get("tool_mode") or "all").lower() == "whitelist"
        _wl_allowed = {str(t) for t in (_wl_cfg.get("allowed_tools") or [])}
        _wl_base = getattr(self.brain, "_WHITELIST_BASE_TOOLS", None) or {
            "recall_memory", "count_history", "search_knowledge", "save_knowledge",
            "check_port", "find_process", "ask_user"}
        for tc in tool_calls:
            func_name = tc["function"]["name"]
            tc_id = tc["id"]
            if _wl_on and func_name not in _wl_base and func_name not in _wl_allowed:
                logger.info(f"[{ctx.display_name} Tool] 🚫 whitelist 실행 차단: {func_name} "
                            f"(허용: {sorted(_wl_allowed) or '없음'})")
                messages.append({
                    "role": "tool", "tool_call_id": tc_id, "name": func_name,
                    "content": ("거부: 이 턴에서 금지된 도구다(직업/임무 화이트리스트 밖 — "
                                "실행되지 않았다). 허용된 수단만으로 수행하거나, 불가하면 "
                                "'해당 작업은 이 편성으로 수행 불가'라고 정직하게 보고하라."),
                })
                _any_tool_error = True
                continue
            result = "Error: Unknown neural tool malfunction."
            args = {}
            try:
                try:
                    args = json.loads(tc["function"]["arguments"])
                except Exception:
                    args = {}
                # 🛰️ 도구 마커에 핵심 인자 요약 동봉 — CLI/헤임달 실시간 도구 피드용
                #    (형식: [TOOL_USE: name · detail] — 소비측 정규식은 [^\]]*로 관용적,
                #     헤임달 스트리밍 배지가 마커 내부를 그대로 보여주므로 가독 구분자 사용)
                detail = _tool_marker_detail(args)
                yield f"[TOOL_USE: {func_name} · {detail}]" if detail else f"[TOOL_USE: {func_name}]"
                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)
                elif func_name == "character_generate":
                    result = yield from self._handle_character_generate(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(AI-Health)에 축적한다.
            # 2026-07-17 재건: ①ctx.prompt(존재하지 않는 필드) AttributeError 로 기록이
            #   항상 조용히 생략되던 사망 상태 수리(→ ctx.user_prompt) ②vault 직업 턴
            #   한정 게이트 제거 — '기록에 없으면 정상'이 성립하려면 이 데몬의 모든
            #   실패를 잡아야 한다 ③한국어 실패 시그니처('실패:'·'거부:'·'오류') 포함.
            _head = str(result)[:80]
            _is_failure = (
                _head.lower().startswith("error")
                or _head.startswith(("❌", "오류", "거부:"))
                or "실패:" in _head or "실패했" in _head
            )
            if _is_failure:
                _any_tool_error = True
                try:
                    cfg = getattr(self.brain, "dynamic_config", None) or {}
                    from obsidian_health import record_tool_failure
                    record_tool_failure(
                        func_name, result, prompt=ctx.user_prompt,
                        guardian=ctx.display_name, job=cfg.get("job", ""),
                    )
                except Exception as health_err:
                    logger.warning("[Obsidian Health] 실패 경험 기록 실패: %s", health_err)

            messages.append({
                "role": "tool", "tool_call_id": tc_id, "name": func_name, "content": str(result),
            })
            if getattr(result, "images", None):
                _images += [(func_name, img) for img in result.images]

            # 🗂️ 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 메시지는 글자만 담을 수 있어 배치 뒤 사용자 메시지로 붙인다
        #    (tool_calls 직후엔 tool 메시지들이 끊김 없이 와야 하므로 반드시 루프 밖에서).
        if _images and getattr(ctx, "accept_images", False):
            parts = [{"type": "text", "text": "[도구 결과 이미지 — "
                      + ", ".join(dict.fromkeys(n for n, _ in _images)) + " 가 돌려준 화면이다. 직접 보고 판단하라.]"}]
            parts += [{"type": "image_url", "image_url": {"url": f"data:{img['mimeType']};base64,{img['data']}"}}
                      for _, img in _images[:8]]
            messages.append({"role": "user", "content": parts})

        # 📊 [스킬 오류율] 이번 턴에 주입된 스킬에 도구 성공/실패를 귀속(상관 신호).
        #   완벽한 인과는 아니지만 누적되면 '자주 깨지는 스킬'을 드러낸다.
        try:
            _skills = getattr(self.brain, "_last_equipped_skills", None)
            if _skills:
                self.brain.record_skill_outcome(_skills, ok=not _any_tool_error)
        except Exception as _sk_e:
            logger.debug("[Skill Outcome] 기록 생략: %s", _sk_e)

    # ──────────────────────────────────────────────
    # 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 t("⏱️ 선택 대기 시간({sec}초)이 초과되어 질문이 자동 취소되었습니다.", sec=_ask['timeout_seconds']) + "\n\n"
                    return "break"
                if _ans == "__CANCELLED__":
                    yield t("🚫 질문이 취소되었습니다 (유저가 다른 작업으로 전환).") + "\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_required_tool(self, full_content, messages, ctx, required):
        """🛡️ 직업 선언 필수 도구(require_tool_call) 미호출 재촉.

        직업 frontmatter 가 require_tool_call: [도구...] 를 선언하면, 그 직업 턴에서
        선언 도구를 하나도 실제 호출하지 않고 말로만 답을 끝내려 할 때 한 번 재촉한다.
        실사고(2026-07-18): 로드맵 서기가 히스토리의 옛 로드맵 답변을 베끼고 도구를
        건너뜀 — rule 문구 강화 2회로도 못 잡아 기계 가드로 격상.
        반환: "continue"(재촉 후 루프 계속) | "break"(해당 없음/소진).
        """
        req = [str(t).strip() for t in (required or []) if str(t).strip()]
        if not req:
            return "break"
        called = getattr(ctx, "used_tool_names", set()) or set()
        if any(t in called for t in req):
            return "break"
        if getattr(ctx, "required_tool_nudges", 0) >= 2:
            return "break"
        if not (full_content or "").strip():
            return "break"
        ctx.required_tool_nudges = getattr(ctx, "required_tool_nudges", 0) + 1
        names = ", ".join(req)
        messages.append({"role": "assistant", "content": full_content})
        messages.append({
            "role": "user",
            "content": (f"방금 답변은 무효다 — 필수 도구({names})를 호출하지 않았다. "
                        f"이전 대화 내용을 베끼지 말고, 지금 즉시 {names} 도구를 실제로 "
                        "호출한 뒤 그 결과만으로 다시 답하라."),
        })
        logger.info(
            f"🛡️ [Required-Tool Nudge] 필수 도구 미호출({names}) — 강제 재촉 "
            f"({ctx.required_tool_nudges}/2)"
        )
        return "continue"

    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"