🔱 Imperial Deployment: Shadow Brain Core ignition (part 2)

shadow_brain_core/runtime/grok_sessions.json

@@ -0,0 +1,12 @@
+{
+  "cli-grok-ce8cfb4e": {
+    "cwd": "/Users/sontaemin80/Git_Work/jarvis_taemin",
+    "session_id": "019f4512-86e8-7a93-b6a5-376f878d739c",
+    "updated_at": "2026-07-09T04:17:11Z"
+  },
+  "discord-grok-1518887916103794739": {
+    "cwd": "/Users/sontaemin80/Git_Work/jarvis_taemin",
+    "session_id": "019f4519-7436-7fe0-a6b1-69b640ef0f4c",
+    "updated_at": "2026-07-09T10:51:01Z"
+  }
+}

shadow_brain_core/scratch/mcp_tool_calls.log

@@ -28,3 +28,35 @@
 [2026-07-07T11:52:01.603817] THINK model=gemini-3.1-flash-lite prompt='[최근 로컬 대화 내역]
 [마왕님]: 안녕
 [로컬 섀도우]: ❌ [Google API 에러' images=False is_task=True force_ANY=True tools_in_set=33
+[2026-07-08T16:14:49.565712] THINK model=gemini-3.1-flash-lite prompt='안녕' images=False is_task=False force_ANY=False tools_in_set=29
+[2026-07-08T16:14:59.376200] THINK model=gemini-3.1-flash-lite prompt='너는 누구지?' images=False is_task=False force_ANY=False tools_in_set=1
+[2026-07-09T00:45:40.234996] THINK model=gemini-3.1-flash-lite prompt='한 문장으로 인사' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-09T00:45:40.865311] THINK model=gemini-3.1-flash-lite prompt='너는 대화에서 '오래 기억할 가치가 있는 사실'만 골라내는 추출기다.
+사용자의 발화에서, ' images=False is_task=True force_ANY=True tools_in_set=137
+[2026-07-09T00:52:13.113137] THINK model=gemini-3.1-flash-lite prompt='안녕' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-09T01:55:27.345795] THINK model=gemini-3.1-flash-lite prompt='하이' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-09T09:14:13.544000] THINK model=gemini-3.1-flash-lite prompt='comfy 로 이미지 생성할수있나요?' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-09T09:14:14.030154] THINK model=gemini-3.1-flash-lite prompt='너는 대화에서 '오래 기억할 가치가 있는 사실'만 골라내는 추출기다.
+사용자의 발화에서, ' images=False is_task=True force_ANY=True tools_in_set=137
+[2026-07-09T09:14:14.599607] CALL mcp_comfyui_comfy_status args=[]
+[2026-07-09T09:14:15.327155] CALL mcp_comfyui_comfy_status args=['request']
+[2026-07-09T09:14:17.435326] CALL mcp_comfyui_comfy_status args=['request']
+[2026-07-09T09:16:07.051282] THINK model=gemini-3.1-flash-lite prompt='그럼 수호자 매치라고알아?' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-09T09:16:07.363845] THINK model=gemini-3.1-flash-lite prompt='너는 대화에서 '오래 기억할 가치가 있는 사실'만 골라내는 추출기다.
+사용자의 발화에서, ' images=False is_task=True force_ANY=True tools_in_set=137
+[2026-07-09T09:38:06.410448] THINK model=gemini-3.1-flash-lite prompt='"달과 그림자 테마 배경 이미지 하나 생성해줘"' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-09T09:38:07.271073] THINK model=gemini-3.1-flash-lite prompt='너는 대화에서 '오래 기억할 가치가 있는 사실'만 골라내는 추출기다.
+사용자의 발화에서, ' images=False is_task=True force_ANY=True tools_in_set=137
+[2026-07-09T09:38:30.775664] THINK model=gemini-3.1-flash-lite prompt='comfy mcp 사용해서만들어줘' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-09T09:38:30.935382] THINK model=gemini-3.1-flash-lite prompt='너는 대화에서 '오래 기억할 가치가 있는 사실'만 골라내는 추출기다.
+사용자의 발화에서, ' images=False is_task=True force_ANY=True tools_in_set=137
+[2026-07-10T00:10:29.928912] THINK model=gemini-3.1-flash-lite prompt='안녕' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-10T00:10:43.066078] THINK model=gemini-3.1-flash-lite prompt='슬퍼해봐요' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-10T00:11:03.705949] THINK model=gemini-3.1-flash-lite prompt='화난 감정표현해주세요' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-10T00:11:03.749128] THINK model=gemini-3.1-flash-lite prompt='너는 대화에서 '오래 기억할 가치가 있는 사실'만 골라내는 추출기다.
+사용자의 발화에서, ' images=False is_task=True force_ANY=True tools_in_set=137
+[2026-07-10T01:44:57.201019] THINK model=gemini-3.1-flash-lite prompt='웃어봐요' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-10T01:45:51.565376] THINK model=gemini-3.1-flash-lite prompt='웃어봐요' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-10T01:46:18.216982] THINK model=gemini-3.1-flash-lite prompt='화내봐요' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-10T04:42:34.476147] THINK model=gemini-3.1-flash-lite prompt='안녕' images=False is_task=False force_ANY=False tools_in_set=137
+[2026-07-10T04:42:55.042710] THINK model=gemini-3.1-flash-lite prompt='화내봐요' images=False is_task=False force_ANY=False tools_in_set=137

shadow_brain_core/services/brain_service.py

@@ -20,7 +20,32 @@ def invalidate_perf_cache(persona_id):
         _PERF_CACHE.pop(str(persona_id), None)
 
 
+_PERF_FLAGS_CACHE = {}  # (user, persona_id) → (ts, flags) — 매 턴 Aegis 왕복(~0.8s) 차단
+_PERF_FLAGS_TTL = 60.0
+
+
 def get_job_perf_flags(persona_id, aegis_url=None, token=None):
+    """타깃 수호자의 직업 성능 플래그 (⏱️ 60초 TTL 캐시 — 실패 포함).
+    직업 변경은 최대 1분 뒤 반영. 실패를 캐시 안 하면 Aegis 401/다운 동안 매 턴
+    클라우드 왕복(~0.8s)이 TTFT에 얹힌다(2026-07-10 실측)."""
+    import time as _time
+    _uid = 'anonymous'
+    try:
+        from flask import has_request_context, request
+        if has_request_context():
+            _uid = request.headers.get('X-Imperial-User-Id', 'anonymous')
+    except Exception:
+        pass
+    _ck = (_uid, str(persona_id))
+    _c = _PERF_FLAGS_CACHE.get(_ck)
+    if _c and (_time.time() - _c[0]) < _PERF_FLAGS_TTL:
+        return dict(_c[1])
+    flags = _get_job_perf_flags_uncached(persona_id, aegis_url=aegis_url, token=token)
+    _PERF_FLAGS_CACHE[_ck] = (_time.time(), dict(flags))
+    return flags
+
+
+def _get_job_perf_flags_uncached(persona_id, aegis_url=None, token=None):
     """타깃 수호자의 직업 성능 플래그를 반환. 실패 시 전부 켜짐(기존 동작)."""
     default = {
         "memory_recall": True, "proactive_context": True, "stage_timing": True,
@@ -303,9 +328,28 @@ def raw_model_completion(prompt, model_name=None, provider=None, system_instruct
         logger.warning(f"[Brain Service] raw_model_completion 실패: {e}")
         return ""
 
+_PROACTIVE_CACHE = {}  # (context_id, persona) → (ts, block)
+_PROACTIVE_TTL = 120.0
+
+
 def proactive_context_block(context_id=None, sender_persona_id=None, aegis_url=None, token=None):
     """[🧠 기억 엔진 3] 그날 첫 대화면 이지스 선제 컨텍스트(날씨/생일)를 가져와
-    첫인사 주입 블록을 만든다. 첫 대화가 아니거나 실패하면 빈 문자열."""
+    첫인사 주입 블록을 만든다. 첫 대화가 아니거나 실패하면 빈 문자열.
+    ⏱️ 120초 TTL 캐시(실패 포함) — '그날 첫 대화' 판정은 하루 단위라 캐시 무해,
+    매 턴 클라우드 왕복(~0.8s) 차단(2026-07-10)."""
+    import time as _time
+    _ck = (str(context_id or ""), str(sender_persona_id or ""))
+    _c = _PROACTIVE_CACHE.get(_ck)
+    if _c and (_time.time() - _c[0]) < _PROACTIVE_TTL:
+        return _c[1]
+    _blk = _proactive_context_block_uncached(
+        context_id=context_id, sender_persona_id=sender_persona_id,
+        aegis_url=aegis_url, token=token)
+    _PROACTIVE_CACHE[_ck] = (_time.time(), _blk)
+    return _blk
+
+
+def _proactive_context_block_uncached(context_id=None, sender_persona_id=None, aegis_url=None, token=None):
     try:
         if os.environ.get("PROACTIVE_CONTEXT_DISABLED", "").lower() in ("1", "true"):
             return ""
@@ -373,7 +417,7 @@ def think(prompt, history=None, stream=True, model_name=None, system_instruction
           user_id=None, context_id='GLOBAL', sender_persona_id=None, target_persona_id=None,
           use_grounding=False, images=None, provider=None, tools=None, token=None, aegis_url=None,
           override_config=None, is_master=False, interactive=True, skip_grounding=False,
-          stream_tokens=False):
+          stream_tokens=False, coding_intent=None):
     """
     [🔱 Imperial Unified Intelligence Wrapper v19 Stateless]
     JarvisBrain.think를 통합 관리합니다. stream=True 시 제너레이터를 반환합니다.
@@ -383,6 +427,13 @@ def think(prompt, history=None, stream=True, model_name=None, system_instruction
         # 실시간 노출하기 위한 마커. stream 모드에서만 [STAGE:라벨|밀리초] 형식으로 흘린다.
         import time as _t
         _stage_t0 = _t.perf_counter()
+        _svc_lap = [_t.time()]
+
+        def _svclap(name):
+            now = _t.time()
+            if now - _svc_lap[0] > 0.15:
+                logger.info(f"⏱️ [svc-prep] {name}: {(now - _svc_lap[0])*1000:.0f}ms")
+            _svc_lap[0] = now
 
         def _stage(label):
             """직전 단계 종료 시점부터의 누적 경과(ms)를 마커로 반환. stream일 때만 사용."""
@@ -443,6 +494,7 @@ def think(prompt, history=None, stream=True, model_name=None, system_instruction
                 logger.error(f"[Brain Service] Aegis lookup failed: {aegis_err}")
 
         # 🚀 [직업 성능 토글] 타깃 수호자 직업 설정으로 준비 단계 ON/OFF 결정.
+        _svclap("identity-resolve")
         _perf = get_job_perf_flags(target_persona_id, aegis_url=aegis_url, token=token)
         _show_stage = stream and _perf.get("stage_timing", True)
         if _perf.get("job_source") == "vault" and history:
@@ -462,6 +514,7 @@ def think(prompt, history=None, stream=True, model_name=None, system_instruction
         if _show_stage:
             yield _stage("🪪 정체성 확인")
 
+        _svclap("perf-flags")
         target_model = final_model or getattr(brain, 'primary_model', 'gemini-3.1-flash-lite')
 
         # [🔱 Claude Code 채널 라우팅] claude_code 프로바이더 or claude_code/ 모델 → localhost:8790
@@ -502,6 +555,7 @@ def think(prompt, history=None, stream=True, model_name=None, system_instruction
             if _show_stage:
                 yield _stage("🧠 기억 검색")
         # [🧠 기억 엔진 3] 그날 첫 대화면 선제 컨텍스트(날씨/생일) 주입
+        _svclap("retrieval+recall")
         proactive_block = ""
         if _perf.get("proactive_context", True):
             proactive_block = proactive_context_block(
@@ -510,6 +564,7 @@ def think(prompt, history=None, stream=True, model_name=None, system_instruction
             )
             if _show_stage:
                 yield _stage("📜 컨텍스트 준비")
+        _svclap("proactive")
         extra = recall_block + proactive_block
         effective_instruction = (system_instruction or "") + extra if extra else system_instruction
 
@@ -531,6 +586,8 @@ def think(prompt, history=None, stream=True, model_name=None, system_instruction
                 skip_grounding=skip_grounding,
                 # 🌊 [Token Streaming] send-stream 옵트인 — Gemini 커스텀 도구 경로 토큰 스트림
                 stream_tokens=stream_tokens,
+                # 🛠️ [Coding Intent Gate] 호출부 명시가 있으면 키워드 추측을 덮는다
+                coding_intent=coding_intent,
                 # 🙋 [Ask User] 비스트리밍(동기 릴레이) 경로에서는 선택지 UI가 제때
                 # 뜰 수 없으므로 ask_user를 차단 (대기 없이 텍스트 질문으로 우회)
                 interactive=(stream and interactive),

shadow_brain_core/taemini_cli_core.py

@@ -0,0 +1,828 @@
+# -*- coding: utf-8 -*-
+"""🔱 태미니 CLI 코어 — 터미널/색/HTTP/설정/소환/수호자/신뢰/세션/전송
+
+taemini_cli.py 에서 분할(2026-07-10, god-file 방지). 공개 이름은 taemini_cli 가
+`from taemini_cli_core import *` 로 재수출하므로 기존 사용법/검증 스크립트는 그대로 동작한다.
+"""
+import argparse
+import hashlib
+import json
+import math
+import os
+import re
+import sys
+import threading
+import time
+import unicodedata
+import urllib.error
+import urllib.request
+
+VERSION = "1.5.0"
+DEFAULT_SERVER = os.environ.get("TAEMINI_SERVER") or os.environ.get("SHADOW_BRAIN_URL") or "http://127.0.0.1:18700"
+REQUEST_TIMEOUT = 900  # 코딩 턴은 수 분 걸릴 수 있음
+PROGRESS_POLL_SEC = 1.0
+
+
+# ─────────────────────────────────────────────────────────────────
+# 터미널 준비 (윈도우 UTF-8 + ANSI VT 활성화)
+# ─────────────────────────────────────────────────────────────────
+def _setup_terminal():
+    if os.name == "nt":
+        os.system("")  # ANSI VT 시퀀스 활성화 (Win10+)
+    for stream in (sys.stdout, sys.stderr):
+        try:
+            stream.reconfigure(encoding="utf-8", errors="replace")
+        except Exception:
+            pass
+
+
+_USE_COLOR = sys.stdout.isatty() and not os.environ.get("NO_COLOR")
+
+
+def _c(code, text):
+    if not _USE_COLOR:
+        return text
+    return f"\033[{code}m{text}\033[0m"
+
+
+def dim(t):     return _c("2", t)
+def bold(t):    return _c("1", t)
+def cyan(t):    return _c("36", t)
+def green(t):   return _c("32", t)
+def yellow(t):  return _c("33", t)
+def red(t):     return _c("31", t)
+def magenta(t): return _c("35", t)
+
+
+# ─────────────────────────────────────────────────────────────────
+# HTTP (표준 라이브러리)
+# ─────────────────────────────────────────────────────────────────
+def post_json(server, path, payload, timeout=REQUEST_TIMEOUT):
+    url = server.rstrip("/") + path
+    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+    headers = {"Content-Type": "application/json; charset=utf-8"}
+    headers.update(auth_headers(server))  # 원격 서버 = 클라우드 키 자동 첨부
+    req = urllib.request.Request(url, data=body, headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        return json.loads(resp.read().decode("utf-8", errors="replace"))
+
+
+def get_json(server, path, timeout=5):
+    url = server.rstrip("/") + path
+    req = urllib.request.Request(url, headers=auth_headers(server))
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        return json.loads(resp.read().decode("utf-8", errors="replace"))
+
+
+# ─────────────────────────────────────────────────────────────────
+# 로컬 설정 (온보딩 GUI 대체) — 런처와 동일한 local_settings.json / SQLite를 공유
+# ─────────────────────────────────────────────────────────────────
+def _settings_path():
+    # local_launcher.get_settings_path()와 동일 규칙 (tkinter import 회피를 위해 재구현)
+    base = os.environ.get("LOCALAPPDATA", os.path.expanduser("~"))
+    d = os.path.join(base, "TaeminGames", "ShadowBrain")
+    os.makedirs(d, exist_ok=True)
+    return os.path.join(d, "local_settings.json")
+
+
+def load_settings():
+    try:
+        with open(_settings_path(), "r", encoding="utf-8") as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+
+def save_settings(settings):
+    with open(_settings_path(), "w", encoding="utf-8") as f:
+        json.dump(settings, f, indent=4)
+
+
+def _store_get(key, default=""):
+    """local_taemini 공급자/모델 — 헤임달 '브레인 지능 설정'과 같은 SQLite SSOT."""
+    try:
+        from local_guardian_store import get_setting
+        return get_setting("anonymous", "local_taemini", key, default)
+    except Exception:
+        return default
+
+
+def _store_set(key, value):
+    from local_guardian_store import set_setting
+    set_setting("anonymous", "local_taemini", key, value)
+    _MODEL_LABEL_CACHE["t"] = 0.0  # /config set·/model 변경 즉시 상태바 반영
+
+
+_MODEL_LABEL_CACHE = {"t": 0.0, "v": ""}
+
+
+def _model_label():
+    """현재 선택된 AI 공급자/모델 라벨 (배너·상태바 표시용) — 헤임달 '브��인 지능
+    설정'과 같은 SQLite SSOT. 상태바가 초당 수회 갱신되므로 5초 캐시로 읽는다."""
+    now = time.time()
+    if now - _MODEL_LABEL_CACHE["t"] > 5.0:
+        prov = (_store_get("ai_provider", "") or "").strip()
+        model = (_store_get("ai_model", "") or "").strip()
+        if prov == "none":
+            prov = ""
+        if len(model) > 36:
+            model = model[:35] + "…"
+        _MODEL_LABEL_CACHE["v"] = f"{prov}/{model}" if prov and model else (model or prov)
+        _MODEL_LABEL_CACHE["t"] = now
+    return _MODEL_LABEL_CACHE["v"]
+
+
+# 설정 항목 레지스트리: 이름 → (저장 위치, 키, 비밀 여부, 설명)
+# 저장 위치 "json" = local_settings.json, "store" = local_data_*.sqlite3(local_taemini)
+CONFIG_KEYS = {
+    "gemini":      ("json", "GEMINI_API_KEY", True,  "Gemini API Key"),
+    "discord":     ("json", "DISCORD_BOT_TOKEN_SHADOW_LOCAL", True, "디스코드 봇 토큰 (로컬 Shadow)"),
+    "agent-token": ("json", "SHADOW_AGENT_TOKEN", True, "제국 Agent Token (비우면 부팅 시 자동 발급)"),
+    "nvidia":      ("json", "NVIDIA_API_KEY", True,  "NVIDIA NIM API Key"),
+    "openrouter":  ("json", "OPENROUTER_API_KEY", True, "OpenRouter API Key"),
+    "glm":         ("json", "GLM_API_KEY", True,  "Z.ai GLM API Key"),
+    "provider":    ("store", "ai_provider", False, "로컬 태미니 AI 공급자 (none·nvidia·openrouter·glm·ollama)"),
+    "model":       ("store", "ai_model", False, "로컬 태미니 모델 ID"),
+    "ultra":       ("json", "TAEMINI_ULTRA_MODE", False, "ULTRA MODE 연출 (on/off — 기본 on)"),
+    "cloud-url":   ("json", "TAEMINI_CLOUD_URL", False, "클라우드 쉐도우 브레인 URL (/server cloud 대상)"),
+    "cloud-key":   ("json", "TAEMINI_CLOUD_KEY", True,  "클라우드 접속 키 (X-Taemini-Key — HF 게이트)"),
+}
+
+DEFAULT_CLOUD_URL = "https://taemin1980-shadowbrain.hf.space"
+
+
+def cloud_url():
+    return (load_settings().get("TAEMINI_CLOUD_URL") or DEFAULT_CLOUD_URL).rstrip("/")
+
+
+def auth_headers(server):
+    """원격(비로컬) 서버 호출 시 클라우드 키 자동 첨부 — HF 공개 게이트 통과용."""
+    try:
+        if _is_local_server(server):
+            return {}
+    except Exception:
+        return {}
+    key = (load_settings().get("TAEMINI_CLOUD_KEY") or "").strip()
+    return {"X-Taemini-Key": key} if key else {}
+
+
+# ⚡ ULTRA MODE 토글 — 기본 켜짐. /config set ultra off 로 이전(무배지) 룩 복귀.
+# 나중에 '스킨' 시스템으로 확장 예정: 이 값이 'on/off'에서 스킨 id로 진화하고,
+# 유저 구매에 따라 팔레트/문구가 달라지는 구조의 자리를 여기 잡아둔다.
+_ULTRA_CACHE = {"v": None}
+
+
+def _ultra_enabled():
+    if _ULTRA_CACHE["v"] is None:
+        val = str(load_settings().get("TAEMINI_ULTRA_MODE", "on")).strip().lower()
+        _ULTRA_CACHE["v"] = val not in ("off", "0", "false", "no")
+    return _ULTRA_CACHE["v"]
+
+
+def _config_get(name):
+    where, key, _secret, _desc = CONFIG_KEYS[name]
+    return load_settings().get(key, "") if where == "json" else _store_get(key)
+
+
+def _config_set(name, value):
+    where, key, _secret, _desc = CONFIG_KEYS[name]
+    if where == "json":
+        s = load_settings()
+        if value:
+            s[key] = value
+        else:
+            s.pop(key, None)
+        save_settings(s)
+    else:
+        _store_set(key, value)
+    if name == "ultra":
+        _ULTRA_CACHE["v"] = None  # 다음 렌더부터 즉시 반영 (재시작 불필요)
+
+
+def _mask(value):
+    if not value:
+        return dim("(없음)")
+    return "•" * 8 + value[-4:] if len(value) > 4 else "•" * 8
+
+
+def cmd_config(arg):
+    parts = arg.split(None, 2)
+    if not parts:
+        print(bold("현재 설정") + dim(f"  ({_settings_path()})"))
+        for name, (_w, _k, secret, desc) in CONFIG_KEYS.items():
+            v = _config_get(name)
+            shown = _mask(v) if secret else (v or dim("(없음)"))
+            print(f"  {cyan(name.ljust(12))} {shown}  {dim(desc)}")
+        print(dim("변경: /config set <이름> <값> · 삭제: /config unset <이름> · 대화식: /setup"))
+        return
+    action = parts[0].lower()
+    name = parts[1].lower() if len(parts) > 1 else ""
+    if name not in CONFIG_KEYS:
+        print(red(f"알 수 없는 설정 이름: {name}") + dim(f"  ({'·'.join(CONFIG_KEYS)})"))
+        return
+    if action == "set" and len(parts) == 3:
+        _config_set(name, parts[2].strip())
+        print(green(f"→ {name} 저장 완료.") + dim(" (데몬 재시작 후 반영: 데몬 종료 뒤 taemini 재실행)"))
+    elif action == "unset":
+        _config_set(name, "")
+        print(green(f"→ {name} 삭제 완료."))
+    else:
+        print(red("사용법: /config set <이름> <값> | /config unset <이름>"))
+
+
+def run_setup_wizard():
+    """터미널 온보딩 — GUI 설정 창의 CLI 대체. Enter=기존 값 유지."""
+    import getpass
+    print(bold(cyan("\n🔱 쉐도우 브레인 터미널 설��")) + dim("  (Enter = 기존 값 유지, '-' 입력 = 값 삭제)"))
+    for name, (_w, _k, secret, desc) in CONFIG_KEYS.items():
+        current = _config_get(name)
+        shown = _mask(current) if secret else (current or dim("(없음)"))
+        prompt = f"  {desc} [{shown}]: "
+        try:
+            raw = getpass.getpass(prompt) if secret else input(prompt)
+        except (EOFError, KeyboardInterrupt):
+            print(dim("\n설정을 중단합니다 (지금까지 입력한 값은 저장됨)."))
+            return
+        raw = raw.strip()
+        if not raw:
+            continue
+        _config_set(name, "" if raw == "-" else raw)
+        print(green("    ✓ 저장"))
+    if not _config_get("agent-token"):
+        print(dim("  ℹ️ Agent Token이 없으면 데몬 첫 부팅 때 자동 발급되어 저장됩니다."))
+    print(green("✅ 설정 완료.") + dim(" 데몬이 이미 떠 있었다면 재시작해야 반영됩니다.\n"))
+
+
+# ─────────────────────────────────────────────────────────────────
+# 데몬 자동 소환 — 로컬 데몬이 꺼져 있으면 Shadow Brain을 백그라운드로 띄운다
+# ─────────────────────────────────────────────────────────────────
+_SPINNER = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏"
+
+SUMMON_WAIT_SEC = 120  # 데몬 점화(모델 로드 포함) 대기 상한
+
+# 🔮 소환 대기 배너 — Claude Code ULTRA 셀렉터풍 보라 그라데이션 물결.
+# 트루컬러 미지원 터미널은 xterm-256 보라 계열로 폴백.
+_WAVE_256 = [54, 55, 56, 57, 93, 99, 98, 92]  # 어두운→밝은 보라 왕복용
+
+
+def _summon_wave_line(text, phase):
+    """text를 셀 단위 보라 그라데이션 배경 위에 얹은 한 줄 ANSI 문자열.
+    phase가 증가하면 물결이 왼쪽→오른쪽으로 흐른다. 한글/이모지는 2셀 폭으로 계산."""
+    truecolor = "truecolor" in os.environ.get("COLORTERM", "") or \
+                "24bit" in os.environ.get("COLORTERM", "")
+    out = []
+    col = 0  # 화면 셀 위치 (전각 문자는 2칸)
+    for ch in text:
+        wide = unicodedata.east_asian_width(ch) in ("W", "F")
+        # 0..1 물결 밝기 — 위치·시간 위상으로 사인파
+        w = 0.5 + 0.5 * math.sin(col * 0.22 - phase * 0.35)
+        if truecolor:
+            r = int(52 + 88 * w)    # 52~140
+            g = int(16 + 44 * w)    # 16~60
+            b = int(110 + 110 * w)  # 110~220
+            bg = f"\033[48;2;{r};{g};{b}m"
+        else:
+            bg = f"\033[48;5;{_WAVE_256[int(w * (len(_WAVE_256) - 1) + 0.5)]}m"
+        out.append(f"{bg}\033[97;1m{ch}")
+        col += 2 if wide else 1
+    out.append("\033[0m")
+    return "".join(out)
+
+
+def daemon_alive(server, timeout=2):
+    try:
+        get_json(server, "/api/shadow/config", timeout=timeout)
+        return True
+    except Exception:
+        return False
+
+
+def _is_local_server(server):
+    from urllib.parse import urlparse
+    host = (urlparse(server).hostname or "").lower()
+    return host in ("127.0.0.1", "localhost", "::1")
+
+
+def _find_summon_command():
+    """플랫폼별 Shadow Brain 실행 명령 탐색. (문자열=셸 실행, 리스트=직접 실행)
+
+    우선순위: 환경변수 override → 리포 소스+venv(--autostart 헤드리스 확실) →
+    설치된 앱/exe(--autostart 인자 전달 — 구버전 빌드는 무시하고 설정 창을 띄우므로
+    그 경우 사용자가 '시작하기'를 눌러야 점화된다).
+    """
+    override = (os.environ.get("TAEMINI_SUMMON_CMD") or "").strip()
+    if override:
+        return override
+    # 1) 리포 소스 실행 — CLI가 리포 안에 살고 venv가 있으면 헤드리스 자동 시작 보장
+    root = os.path.dirname(os.path.abspath(__file__))
+    launcher = os.path.join(root, "local_launcher.py")
+    venv_py = os.path.join(root, "venv",
+                           "Scripts" if os.name == "nt" else "bin",
+                           "python.exe" if os.name == "nt" else "python3")
+    if os.path.isfile(launcher) and os.path.isfile(venv_py):
+        return [venv_py, launcher, "--autostart"]
+    # 2) 설치된 배포본
+    if sys.platform == "darwin":
+        for app in ("/Applications/ShadowBrain.app",
+                    os.path.expanduser("~/Applications/ShadowBrain.app")):
+            if os.path.isdir(app):
+                return ["open", "-a", app, "--args", "--autostart"]
+    elif os.name == "nt":
+        for base in (os.environ.get("ProgramFiles"), os.environ.get("ProgramFiles(x86)"),
+                     os.path.join(os.environ.get("LOCALAPPDATA", ""), "Programs")):
+            exe = os.path.join(base or "", "TaeminGames", "ShadowBrain", "ShadowBrain.exe")
+            if base and os.path.isfile(exe):
+                return [exe, "--autostart"]
+    # 3) venv 없는 소스 (표준 파이썬 — web_server 의존성 부족 가능성 있으나 최후 시도)
+    if os.path.isfile(launcher):
+        return [sys.executable, launcher, "--autostart"]
+    return None
+
+
+def summon_daemon(server):
+    """데몬이 꺼져 있으면 백그라운드로 소환하고 점화 완료까지 대기. 성공 여부 반환."""
+    import subprocess
+    cmd = _find_summon_command()
+    if not cmd:
+        print(red("❌ Shadow Brain 실행 파일을 찾지 못했습니다."))
+        print(dim("   TAEMINI_SUMMON_CMD 환경변수로 실행 명령을 지정하거나 직접 실행해 주십시오."))
+        return False
+    shown = cmd if isinstance(cmd, str) else " ".join(cmd)
+    print(dim(f"🔱 쉐도우 브레인이 꺼져 있어 소환합니다: {shown}"))
+    if not isinstance(cmd, str) and "--autostart" in cmd and (".app" in shown or ".exe" in shown):
+        # 구버전 배포본은 --autostart를 몰라 설정 창을 띄운다 — 사용자 클릭 안내
+        print(dim("   (설정 창이 뜨면 '시작하기'를 눌러 주십시오 — 최신 빌드는 창 없이 자동 점화됩니다)"))
+    try:
+        kwargs = dict(stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                      stdin=subprocess.DEVNULL)
+        if os.name == "nt":
+            kwargs["creationflags"] = 0x00000008 | 0x00000200  # DETACHED | NEW_PROCESS_GROUP
+        else:
+            kwargs["start_new_session"] = True
+        if isinstance(cmd, str):
+            subprocess.Popen(cmd, shell=True, **kwargs)
+        else:
+            subprocess.Popen(cmd, **kwargs)
+    except Exception as e:
+        print(red(f"❌ 소환 실패: {e}"))
+        return False
+
+    started = time.time()
+    i = 0
+    last_check = 0.0
+    tick = 0.08 if _USE_COLOR else 1.0  # 컬러면 물결 애니메이션(12.5fps), 아니면 1초 스피너
+    while time.time() - started < SUMMON_WAIT_SEC:
+        now = time.time()
+        if now - last_check >= 1.0:  # 생존 확인은 1초마다 (urlopen 부하 억제)
+            last_check = now
+            if daemon_alive(server):
+                sys.stdout.write("\r\033[K" if _USE_COLOR else "\r")
+                print(green(f"✅ 쉐도우 브레인 점화 완료 ({now - started:.0f}s)"))
+                return True
+        i += 1
+        elapsed = now - started
+        if _USE_COLOR:
+            msg = f"  🔮 ShadowBrain 로딩 중 — 잠시만 기다려주세요…  {elapsed:3.0f}s / {SUMMON_WAIT_SEC}s  "
+            sys.stdout.write("\r\033[K" + _summon_wave_line(msg, i))
+        else:
+            frame = _SPINNER[i % len(_SPINNER)]
+            sys.stdout.write(f"\r{frame} 쉐도우 브레인 점화 대기 중 · {elapsed:4.0f}s / {SUMMON_WAIT_SEC}s")
+        sys.stdout.flush()
+        time.sleep(tick)
+    sys.stdout.write("\r\033[K" if _USE_COLOR else "\r")
+    print(red(f"❌ {SUMMON_WAIT_SEC}초 안에 데몬이 응답하지 않았습니다. 직접 실행 상태를 확인해 주십시오."))
+    return False
+
+
+def ensure_daemon(server, auto_summon=True):
+    """데몬 생존 확인 → 로컬이면 자동 소환까지. True면 사용 가능."""
+    if daemon_alive(server):
+        return True
+    if not _is_local_server(server):
+        print(red(f"❌ 원격 쉐도우 브레인({server})에 연결할 수 없습니다."))
+        return False
+    if not auto_summon:
+        print(red(f"❌ 쉐도우 브레인 데몬({server})이 꺼져 있습니다. (--no-summon 지정됨)"))
+        return False
+    return summon_daemon(server)
+
+
+# ─────────────────────────────────────────────────────────────────
+# 수호자(중계) 정의 — shadow_brain_core/blueprints/relays/* 라우트와 1:1
+# ─────────────────────────────────────────────────────────────────
+GUARDIAN_LABELS = {
+    "taemini": "로컬 태미니 (Shadow Brain)",
+    "codex":   "로컬 코덱스 (Codex CLI)",
+    "minimax": "로컬 MiniMax Code (Mavis)",
+    "hermes":  "헤르메스 (Hermes ACP)",
+    "claude":  "자비스 (Claude Code 채널)",
+    "lmstudio": "LM Studio (로컬 LLM)",
+    "grok":    "로컬 그록 (Grok CLI)",
+}
+
+GUARDIAN_ALIASES = {
+    "태미니": "taemini", "shadow": "taemini", "타미니": "taemini",
+    "코덱스": "codex",
+    "미니맥스": "minimax", "mavis": "minimax",
+    "헤르메스": "hermes",
+    "자비스": "claude", "claude-code": "claude", "jarvis": "claude",
+    "lm": "lmstudio", "lms": "lmstudio", "엘엠스튜디오": "lmstudio",
+    "그록": "grok", "xai": "grok",
+}
+
+
+def resolve_guardian(name):
+    n = (name or "").strip().lower()
+    n = GUARDIAN_ALIASES.get(n, n)
+    return n if n in GUARDIAN_LABELS else None
+
+
+# 🎖️ [직업 강제 안내] 이 수호자가 직업(도구/스킬/앱)을 실제로 100% 따르는가.
+# 기준: "그 수��자의 디스패치 루프를 우리가 통제하는가"(규약: job-enforcement-axis).
+#  - taemini(=로컬 태미니, 우리 코딩 에이전트) → 직업 100% 강제
+#  - 그 외 릴레이(codex/minimax/hermes/claude/lmstudio) → 남의 에이전트, 직업 무시(자체 도구만)
+JOB_ENFORCED_GUARDIANS = {"taemini"}
+
+
+def job_enforce_tag(guardian, verbose=False):
+    """멘션 메뉴/목록에 붙일 짧은 직업 강제 표식. verbose면 설명형 문구."""
+    if guardian in JOB_ENFORCED_GUARDIANS:
+        return "직업 100% 사용" if verbose else "🎖️직업"
+    return "직업 무시 · 자체 도구만" if verbose else "직업✕"
+
+
+# ─────────────────────────────────────────────────────────────────
+# 📁 폴더 신뢰 (Claude Code 동일 철학) — 낯선 폴더의 프롬프트 인젝션 방어.
+# 신뢰한 폴더만 코딩 도구(파일 수정·명령 실행)의 작업 루트로 전달한다.
+# 저장소: SQLite(local_guardian_kv, SSOT) + local_settings.json 이중 기록.
+# 예전 JSON 단독 기록은 첫 조회 때 SQLite로 자동 이관. 하위 폴더 자동 신뢰.
+# ─────────────────────────────────────────────────────────────────
+def _load_trust_state():
+    """(신뢰 폴더 목록, 최근 사용 시각 맵) — SQLite 우선, 없으면 JSON에서 이관."""
+    dirs = used = None
+    try:
+        raw = _store_get("trusted_dirs", "")
+        if raw:
+            v = json.loads(raw)
+            if isinstance(v, list):
+                dirs = [str(x) for x in v]
+        raw = _store_get("trusted_dirs_used", "")
+        if raw:
+            v = json.loads(raw)
+            if isinstance(v, dict):
+                used = {}
+                for k, t in v.items():
+                    try:
+                        used[str(k)] = float(t)
+                    except (TypeError, ValueError):
+                        pass
+    except Exception:
+        dirs = used = None
+    if dirs is None or used is None:
+        s = load_settings()
+        if dirs is None:
+            v = s.get("TAEMINI_TRUSTED_DIRS")
+            dirs = [str(x) for x in v] if isinstance(v, list) else []
+        if used is None:
+            v = s.get("TAEMINI_TRUSTED_DIRS_USED")
+            used = {}
+            if isinstance(v, dict):
+                for k, t in v.items():
+                    try:
+                        used[str(k)] = float(t)
+                    except (TypeError, ValueError):
+                        pass
+        if dirs or used:
+            _save_trust_state(dirs, used)  # JSON → SQLite 1회 이관
+    return dirs, used
+
+
+def _save_trust_state(dirs, used):
+    # 신뢰 목록에서 빠진 최근 사용 기록은 정리
+    keep = set(os.path.abspath(t) for t in dirs)
+    used = {k: v for k, v in used.items() if os.path.abspath(k) in keep}
+    try:
+        _store_set("trusted_dirs", json.dumps(dirs, ensure_ascii=False))
+        _store_set("trusted_dirs_used", json.dumps(used))
+    except Exception:
+        pass  # SQLite 불가 환경 — JSON 폴백만으로 동작
+    s = load_settings()
+    s["TAEMINI_TRUSTED_DIRS"] = dirs
+    s["TAEMINI_TRUSTED_DIRS_USED"] = used
+    save_settings(s)
+
+
+def _trusted_dirs():
+    return _load_trust_state()[0]
+
+
+def is_trusted_dir(path):
+    p = os.path.abspath(path)
+    for t in _trusted_dirs():
+        t = os.path.abspath(t)
+        if p == t or (p + os.sep).startswith(t + os.sep):
+            return True
+    return False
+
+
+def trust_dir(path):
+    p = os.path.abspath(path)
+    dirs, used = _load_trust_state()
+    if p not in dirs:
+        dirs.append(p)
+    used[p] = time.time()
+    _save_trust_state(dirs, used)
+
+
+def touch_trusted_dir(path):
+    """신뢰 폴더 최근 사용 시각 기록 — 시작 선택창의 정렬/`(9h ago)` 표시용."""
+    p = os.path.abspath(path)
+    dirs, used = _load_trust_state()
+    used[p] = time.time()
+    _save_trust_state(dirs, used)
+
+
+def _fmt_ago(ts):
+    if not ts:
+        return ""
+    sec = max(0, time.time() - float(ts))
+    if sec < 3600:
+        return f"{int(sec // 60)}m ago"
+    if sec < 86400:
+        return f"{int(sec // 3600)}h ago"
+    return f"{int(sec // 86400)}d ago"
+
+
+def _tilde(path):
+    home = os.path.expanduser("~")
+    return path.replace(home, "~", 1) if path.startswith(home) else path
+
+
+def untrust_dir(path):
+    """신뢰 목록에서 제거 — 시작 선택창의 d 키 / `/untrust` 명령용."""
+    p = os.path.abspath(path)
+    dirs, used = _load_trust_state()
+    dirs = [d for d in dirs if os.path.abspath(d) != p]
+    used.pop(p, None)
+    _save_trust_state(dirs, used)
+
+
+def _start_dir_candidates(cur, dirs, used):
+    cands = [os.path.abspath(t) for t in dirs
+             if os.path.isdir(t) and os.path.abspath(t) != cur]
+    cands.sort(key=lambda p: used.get(p, 0), reverse=True)
+    return cands[:8]
+
+
+def _read_menu_key():
+    """1키 읽기 → 'up'/'down'/'enter'/'esc'/소문자 문자.
+    유닉스는 _pick_menu가 메뉴 동안 cbreak 모드를 유지한 상태에서 호출한다.
+    ⚠️ sys.stdin.read(1) 금지 — 버퍼드 IO가 뒤따르는 이스케이프 바이트를 내부
+    버퍼로 삼켜 fd 기준 select()가 항상 비어 보인다(화살표→Esc 오인 실사례)."""
+    if os.name == "nt":
+        import msvcrt
+        ch = msvcrt.getwch()
+        if ch in ("\x00", "\xe0"):
+            return {"H": "up", "P": "down"}.get(msvcrt.getwch(), "")
+        if ch in ("\r", "\n"):
+            return "enter"
+        if ch == "\x03":
+            raise KeyboardInterrupt
+        return "esc" if ch == "\x1b" else ch.lower()
+    import select as _sel
+    fd = sys.stdin.fileno()
+    ch = os.read(fd, 1).decode("utf-8", "ignore")
+    if ch == "\x1b":
+        if not _sel.select([fd], [], [], 0.05)[0]:
+            return "esc"
+        seq = os.read(fd, 2).decode("ascii", "ignore")
+        if seq.startswith("["):
+            c = seq[1:] or os.read(fd, 1).decode("ascii", "ignore")
+            return {"A": "up", "B": "down"}.get(c, "")
+        return "esc"
+    if ch in ("\r", "\n"):
+        return "enter"
+    if ch == "\x03":
+        raise KeyboardInterrupt
+    return ch.lower()
+
+
+def _menu_rows(cwd, cands, used):
+    rows = [("cur", cwd, f"{_tilde(cwd)}  (현재)")]
+    for p in cands:
+        ago = _fmt_ago(used.get(p))
+        name = os.path.basename(p) or p
+        rows.append(("dir", p, f"{name:<20s} {_tilde(p)}" + (f"  ({ago})" if ago else "")))
+    rows.append(("z", None, "직접 경로 입력"))
+    return rows
+
+
+def _pick_menu(cwd):
+    """화살표 선택 메뉴 — ↑↓ 이동, Enter 선택, 숫자 단축, d=항목 제거(신뢰 해제).
+    반환: ('cur'|'dir'|'z', 경로)."""
+    idx = 0
+    drawn = 0
+    old_attr = None
+    if os.name != "nt":
+        import termios
+        import tty
+        fd = sys.stdin.fileno()
+        old_attr = termios.tcgetattr(fd)  # 비TTY면 여기서 예외 → 숫자 폴백
+        # ⚠️ setraw(기본 TCSAFLUSH) 금지 — 렌더링 중 눌린 키가 통째로 버려져
+        # "첫 키만 먹히는" 실사례. cbreak를 메뉴 동안 1회만 유지(TCSANOW, 입력 보존).
+        tty.setcbreak(fd, termios.TCSANOW)
+    sys.stdout.write("\033[?25l")  # 커서 숨김
+    try:
+        while True:
+            dirs, used = _load_trust_state()
+            rows = _menu_rows(cwd, _start_dir_candidates(cwd, dirs, used), used)
+            idx = max(0, min(idx, len(rows) - 1))
+            if drawn:
+                sys.stdout.write(f"\033[{drawn}A")
+            for i, (_kind, _p, label) in enumerate(rows):
+                line = f"   {i + 1}) {label}"
+                if i == idx:
+                    line = f"\033[7m ❯ {i + 1}) {label}\033[0m"
+                sys.stdout.write("\033[K" + line + "\n")
+            sys.stdout.write("\033[J")  # 항목 삭제로 줄이 줄었을 때 잔여 줄 정리
+            drawn = len(rows)
+            sys.stdout.flush()
+            key = _read_menu_key()
+            if key == "up":
+                idx = (idx - 1) % len(rows)
+            elif key == "down":
+                idx = (idx + 1) % len(rows)
+            elif key in ("enter",):
+                return rows[idx][0], rows[idx][1]
+            elif key == "esc" or key == "q":
+                return "cur", cwd
+            elif key == "z":
+                return "z", None
+            elif key == "d" and rows[idx][0] == "dir":
+                untrust_dir(rows[idx][1])  # 다음 루프에서 목록 재구성
+            elif key and key.isdigit() and 1 <= int(key) <= len(rows):
+                return rows[int(key) - 1][0], rows[int(key) - 1][1]
+    finally:
+        sys.stdout.write("\033[?25h")
+        sys.stdout.flush()
+        if old_attr is not None:
+            import termios
+            termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, old_attr)
+
+
+
+def _pick_numeric(cwd, cands, used):
+    """숫자 입력 폴백 (원시 키 입력 불가 환경)."""
+    for i, (_kind, _p, label) in enumerate(_menu_rows(cwd, cands, used), start=1):
+        print(f"   {i}) {label}")
+    try:
+        ans = input("   선택 [1] ").strip().lower()
+    except (EOFError, KeyboardInterrupt):
+        ans = ""
+    if ans == "z" or (ans.isdigit() and int(ans) == len(cands) + 2):
+        return "z", None
+    if ans.isdigit() and 2 <= int(ans) <= len(cands) + 1:
+        return "dir", cands[int(ans) - 2]
+    return "cur", cwd
+
+
+def pick_start_dir(cwd):
+    """그록 빌드식 시작 폴더 선택창 — 대화식 시작 시 항상,
+    과거에 신뢰했던 폴더들을 최근 사용 순으로 보여주고 골라서 시작한다.
+    화살표 메뉴(↑↓·Enter·숫자 단축·d=목록 제거), 미지원 환경은 숫자 입력.
+    엔터 한 번 = 현재 폴더 그대로. 반환: (선택된 경로, 그 경로가 신뢰 폴더인지)."""
+    cur = os.path.abspath(cwd)
+    dirs, used = _load_trust_state()
+    cands = _start_dir_candidates(cur, dirs, used)
+    if not cands:
+        return cwd, is_trusted_dir(cwd)
+
+    print(bold("📁 프로젝트 폴더에서 시작할까요?"))
+    print(dim("   ↑↓ 이동 · Enter 선택 · 숫자 바로 선택 · d 목록에서 제거 · z 직접 경로"))
+    try:
+        kind, path = _pick_menu(cur)
+    except KeyboardInterrupt:
+        return cwd, is_trusted_dir(cwd)
+    except Exception:
+        kind, path = _pick_numeric(cur, cands, used)
+    if kind == "dir":
+        return path, True
+    if kind == "z":
+        try:
+            typed = input("   경로: ").strip()
+        except (EOFError, KeyboardInterrupt):
+            typed = ""
+        p = os.path.abspath(os.path.expanduser(typed)) if typed else cwd
+        if typed and os.path.isdir(p):
+            return p, is_trusted_dir(p)
+        if typed:
+            print(red(f"   폴더가 아닙니다: {typed} — 현재 폴더로 시작합니다."))
+    return cwd, is_trusted_dir(cwd)
+
+
+class Session:
+    """대화 상태 — 수호자별 세션 키(연속성) + 작업 루트."""
+
+    def __init__(self, server, cwd, guardian="taemini"):
+        self.server = server
+        self.cwd = os.path.abspath(cwd)
+        self.trusted = True  # main()의 신뢰 게이트가 결정 (False=안전 모드)
+        self.guardian = guardian
+        self.mode = "chat"  # chat=일반 · plan=플랜(계획만, 수정 금지) — Shift+Tab 전환
+        self.epoch = ""  # /new 시 바뀌는 세션 소금값
+        self.minimax_session = ""  # mavis가 돌려준 세션 ID 재사용(세션 폭증 방지)
+        # 🔊 Edge TTS (응답 읽어주기) — local_settings.json에 영속
+        _s = load_settings()
+        self.tts_enabled = bool(_s.get("CLI_TTS_ENABLED"))
+        self.tts_voice = _s.get("CLI_TTS_VOICE") or "ko-KR-SunHiNeural"
+
+    @property
+    def work_cwd(self):
+        """코딩 작업 루트 — 신뢰된 폴더만 서버에 전달. 비신뢰면 None(안전 모드)."""
+        return self.cwd if self.trusted else None
+
+    def key(self, prefix):
+        h = hashlib.sha1(self.cwd.encode("utf-8")).hexdigest()[:8]
+        return f"{prefix}{h}{self.epoch}"
+
+    def taemini_key(self):
+        # 'taemini-code-' 접두사 → 서버가 간결 수칙 + 코딩 에이전트 모드로 처리
+        return self.key("taemini-code-cli-")
+
+    def new_epoch(self):
+        self.epoch = "-" + hashlib.sha1(str(time.time()).encode()).hexdigest()[:6]
+        self.minimax_session = ""
+
+
+def send_to_guardian(sess, guardian, message, plain=False, session_override=None):
+    """수호자에게 메시지 전송 → (본문, todos, stats) 반환. 오류는 예외로.
+
+    plain=True: 코딩 콘솔 세션이 아닌 일반 대화 세션으로 보낸다 (/debate 등 —
+    코딩 수칙·코딩 모드가 주입되면 토론에 자꾸 code_* 도구 얘기를 하므로 분리).
+    session_override: 태미니 세션 키를 직접 지정 (토론은 회차마다 새 세션 —
+    이전 토론의 오염된 히스토리를 이어받는 사고 방지).
+    """
+    if guardian == "taemini":
+        if plain:
+            payload = {"message": message,
+                       "sessionKey": session_override or sess.key("cli-debate-"),
+                       "skipGrounding": True,
+                       # 시스템 레벨 토론 지시 — 코딩 페르소나의 도구 강박을 차단
+                       "extraSystem": ("지금은 '수호자 토론 모드'다. 코딩 작업이 아니다. "
+                                       "어떤 도구도 호출하지 말고, 주어진 토론 주제에 대한 "
+                                       "의견 텍스트만 간결히 출력하라. 경로/파일/도구 이야기를 꺼내지 마라.")}
+        else:
+            payload = {"message": message, "sessionKey": sess.taemini_key(), "cwd": sess.work_cwd}
+        data = post_json(sess.server, "/api/shadow/shadow/send", payload)
+    elif guardian == "codex":
+        data = post_json(sess.server, "/api/shadow/codex/send", {
+            "message": message,
+            "cwd": sess.work_cwd,
+            "context_id": sess.key("cli-codex-"),
+        })
+    elif guardian == "minimax":
+        data = post_json(sess.server, "/api/shadow/minimax-code/send", {
+            "message": message,
+            "cwd": sess.work_cwd,
+            "sessionKey": sess.minimax_session,
+        })
+        if isinstance(data, dict) and data.get("sessionKey"):
+            sess.minimax_session = str(data["sessionKey"])
+    elif guardian == "hermes":
+        data = post_json(sess.server, "/api/shadow/hermes/send", {
+            "message": message,
+            "sessionKey": sess.key("cli-hermes-"),
+            "workdir": sess.work_cwd,
+        })
+    elif guardian == "claude":
+        data = post_json(sess.server, "/api/shadow/claude-code/send", {
+            "message": message,
+            "context_id": sess.key("cli-claude-"),
+        })
+    elif guardian == "lmstudio":
+        data = post_json(sess.server, "/api/shadow/lmstudio/send", {
+            "message": message,
+            "sessionKey": sess.key("cli-lms-"),
+        })
+    elif guardian == "grok":
+        data = post_json(sess.server, "/api/shadow/grok/send", {
+            "message": message,
+            "cwd": sess.work_cwd,
+            "context_id": sess.key("cli-grok-"),
+        })
+    else:
+        raise ValueError(f"알 수 없는 수호자: {guardian}")
+
+    if not isinstance(data, dict):
+        raise RuntimeError(f"서버 응답 형식 오류: {data!r}")
+    if data.get("status") == "ERROR":
+        raise RuntimeError(data.get("message") or "알 수 없는 서버 오류")
+    content = (data.get("content") or "").strip()
+    return content, data.get("todos") or [], data.get("stats") or {}
+
+# 📝 마크다운 경량 정규식 (렌더러·TTS 정리 공용)
+_MD_BOLD = re.compile(r"\*\*(.+?)\*\*")
+_MD_INLINE_CODE = re.compile(r"`([^`\n]+)`")
+_MD_BULLET = re.compile(r"^(\s*)[-*] (?=\S)")
+
+__all__ = [_n for _n in list(globals()) if not _n.startswith("__")]

shadow_brain_core/taemini_cli_media.py

@@ -0,0 +1,529 @@
+# -*- coding: utf-8 -*-
+"""🔱 태미니 CLI 미디어 — Edge TTS 낭독 + 음성 입력(/voice)
+
+taemini_cli.py 에서 분할(2026-07-10, god-file 방지). 공개 이름은 taemini_cli 가
+`from taemini_cli_media import *` 로 재수출하므로 기존 사용법/검증 스크립트는 그대로 동작한다.
+"""
+import argparse
+import hashlib
+import json
+import math
+import os
+import re
+import sys
+import threading
+import time
+import unicodedata
+import urllib.error
+import urllib.request
+
+VERSION = "1.5.0"
+DEFAULT_SERVER = os.environ.get("TAEMINI_SERVER") or os.environ.get("SHADOW_BRAIN_URL") or "http://127.0.0.1:18700"
+REQUEST_TIMEOUT = 900  # 코딩 턴은 수 분 걸릴 수 있음
+PROGRESS_POLL_SEC = 1.0
+
+from taemini_cli_core import *  # noqa: F401,F403
+
+
+# ─────────────────────────────────────────────────────────────────
+# 🔊 Edge TTS — 응답을 데몬 /api/tts 로 합성해 백그라운드 재생
+# ─────────────────────────────────────────────────────────────────
+_TTS_PROC = None
+
+
+# 이모지·픽토그램 — Edge TTS가 "웃는 얼굴"처럼 읽어버리므로 낭독 전 제거
+_EMOJI_RE = re.compile(
+    "["
+    "\U0001F000-\U0001FAFF"   # 이모티콘·픽토그램·교통·보조 기호 전반
+    "☀-➿"           # 잡기호·딩벳 (☀☔✅✈ 등)
+    "⬀-⯿"           # 화살표·별 (⭐ 등)
+    "←-⇿"           # 화살표 (→ 등)
+    "⌀-⏿"           # 기술 기호 (⌛⏰ 등)
+    "■-◿"           # 도형 (▶■ 등)
+    "\U0001F1E6-\U0001F1FF"   # 국기 조합 문자
+    "︀-️"           # variation selector
+    "‍"                  # ZWJ (조합 이모지 연결자)
+    "⃣"                  # keycap 조합
+    "]+"
+)
+
+
+def _strip_for_speech(text):
+    t = re.sub(r"```[\s\S]*?```", " 코드 블록은 생략합니다. ", text or "")
+    t = re.sub(r"`([^`\n]+)`", r"\1", t)
+    t = _MD_BOLD.sub(r"\1", t)
+    t = re.sub(r"^#+\s*", "", t, flags=re.MULTILINE)
+    t = re.sub(r"[─│┌└╭╰╮╯]+", " ", t)
+    t = _EMOJI_RE.sub(" ", t)
+    return re.sub(r"\s+", " ", t).strip()
+
+
+_TTS_GEN = 0  # /tts stop 세대 카운터 — 합성 대기 중이던 낭독도 재생 전에 폐기
+
+
+def tts_stop():
+    """재생 중인 TTS 낭독 즉시 중단. 중단한 게 있으면 True."""
+    global _TTS_GEN, _TTS_PROC
+    _TTS_GEN += 1  # 합성(urlopen) 진행 중이던 work()도 재생 직전 검사에서 폐기됨
+    p = _TTS_PROC
+    if p and p.poll() is None:
+        try:
+            p.terminate()
+        except Exception:
+            pass
+        return True
+    return False
+
+
+def tts_speak(sess, text):
+    """응답 본문을 Edge TTS로 읽어준다 (비차단, 새 응답이 오면 이전 재생 중단)."""
+    if not getattr(sess, "tts_enabled", False):
+        return
+    body = _strip_for_speech(text)[:800]
+    if not body:
+        return
+    gen = _TTS_GEN
+
+    def work():
+        global _TTS_PROC
+        import subprocess
+        import tempfile
+        try:
+            req = urllib.request.Request(
+                sess.server.rstrip("/") + "/api/tts",
+                data=json.dumps({"text": body, "voice": sess.tts_voice},
+                                ensure_ascii=False).encode("utf-8"),
+                headers={"Content-Type": "application/json"}, method="POST")
+            with urllib.request.urlopen(req, timeout=90) as resp:
+                audio = resp.read()
+            if not audio or audio[:1] == b"{":  # 오류 JSON이면 무시
+                return
+            if gen != _TTS_GEN:  # 합성 중에 /tts stop — 재생하지 않고 폐기
+                return
+            f = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False)
+            f.write(audio)
+            f.close()
+            if _TTS_PROC and _TTS_PROC.poll() is None:
+                _TTS_PROC.terminate()
+            if sys.platform == "darwin":
+                player = ["afplay", f.name]
+            else:
+                import shutil
+                player = None
+                for cand in (["ffplay", "-nodisp", "-autoexit", "-loglevel", "quiet"],
+                             ["mpg123", "-q"]):
+                    if shutil.which(cand[0]):
+                        player = cand + [f.name]
+                        break
+                if player is None:
+                    if os.name == "nt":
+                        os.startfile(f.name)  # 기본 플레이어
+                    return
+            _TTS_PROC = subprocess.Popen(player, stdout=subprocess.DEVNULL,
+                                         stderr=subprocess.DEVNULL)
+            if _TTS_PROC:
+                _TTS_PROC.wait()  # 재생 종료까지 스레드 유지 (--once에서 join 가능)
+        except Exception:
+            pass  # TTS는 부가 기능 — 실패해도 대화를 막지 않는다
+
+    t = threading.Thread(target=work, daemon=True)
+    t.start()
+    sess._tts_thread = t  # --once 모드가 재생 완료를 기다릴 수 있게
+
+
+def cmd_tts(sess, arg):
+    parts = arg.split(None, 1)
+    action = parts[0].lower() if parts else ""
+    if action in ("", "on", "off"):
+        if action == "on":
+            sess.tts_enabled = True
+        elif action == "off":
+            sess.tts_enabled = False
+        else:
+            sess.tts_enabled = not sess.tts_enabled
+        s = load_settings()
+        s["CLI_TTS_ENABLED"] = sess.tts_enabled
+        save_settings(s)
+        state = green("🔊 켜짐") if sess.tts_enabled else dim("🔇 꺼짐")
+        print(f"→ 응답 음성 읽기(Edge TTS): {state}" + dim(f"  (음성: {sess.tts_voice})"))
+    elif action == "stop":
+        stopped = tts_stop()
+        print(dim("→ 낭독을 중단했습니다.") if stopped else dim("→ 재생 중인 낭독이 없습니다."))
+    elif action == "voice" and len(parts) > 1:
+        sess.tts_voice = parts[1].strip()
+        s = load_settings()
+        s["CLI_TTS_VOICE"] = sess.tts_voice
+        save_settings(s)
+        print(green(f"→ TTS 음성: {sess.tts_voice}"))
+    else:
+        print(red("사용법: /tts [on|off] · /tts stop · /tts voice <음성ID>"))
+        print(dim("  한국어 음성 예: ko-KR-SunHiNeural(여) · ko-KR-InJoonNeural(남) · ko-KR-HyunsuMultilingualNeural(남)"))
+
+
+# ─────────────────────────────────────────────────────────────────
+# 🎤 음성 입력 (/voice) — ffmpeg 마이크 녹음 → 데몬 faster-whisper 전사
+# ─────────────────────────────────────────────────────────────────
+VOICE_MAX_SEC = 60  # 녹음 상한 (안전장치)
+
+
+def warmup_stt(sess):
+    """small 전사 모델을 백그라운드로 미리 로드 — 첫 /voice의 20초 지연 제거."""
+    def work():
+        import struct
+        import tempfile
+        try:
+            # 0.2초 무음 wav (16kHz mono s16le) 생성 → quick 라우트 호출로 모델 로드 유도
+            sr, n = 16000, 3200
+            path = os.path.join(tempfile.gettempdir(), "taemini_stt_warmup.wav")
+            data = b"\x00\x00" * n
+            with open(path, "wb") as f:
+                f.write(b"RIFF" + struct.pack("<I", 36 + len(data)) + b"WAVEfmt " +
+                        struct.pack("<IHHIIHH", 16, 1, 1, sr, sr * 2, 2, 16) +
+                        b"data" + struct.pack("<I", len(data)) + data)
+            for size in ("base", "small"):  # base=실시간 자막, small=최종 확정
+                post_json(sess.server, "/api/transcribe/quick",
+                          {"path": path, "language": "ko", "model": size}, timeout=300)
+        except Exception:
+            pass  # 워밍업 실패는 무해 — 첫 /voice 때 로드될 뿐
+
+    threading.Thread(target=work, daemon=True).start()
+
+
+def _list_audio_devices():
+    """macOS avfoundation 오디오 입력 장치 목록 [(번호, 이름)]."""
+    import shutil
+    import subprocess
+    ffmpeg = shutil.which("ffmpeg")
+    if not ffmpeg or sys.platform != "darwin":
+        return []
+    out = subprocess.run([ffmpeg, "-f", "avfoundation", "-list_devices", "true", "-i", ""],
+                         capture_output=True, text=True).stderr
+    devices = []
+    in_audio = False
+    for line in out.splitlines():
+        if "audio devices" in line:
+            in_audio = True
+            continue
+        if in_audio:
+            m = re.search(r"\[(\d+)\]\s+(.+)$", line)
+            if m:
+                devices.append((m.group(1), m.group(2).strip()))
+            elif "]" not in line:
+                break
+    return devices
+
+
+def cmd_voice_mic(sess, arg):
+    """/voice mic [번호|default] — 녹음 마이크 확인/변경 (영구 저장).
+    인자 없이 부르면 목록을 보여주고 그 자리에서 번호를 골라 저장한다."""
+    current = load_settings().get("CLI_VOICE_MIC", "default")
+
+    def _save(value):
+        s = load_settings()
+        s["CLI_VOICE_MIC"] = value
+        save_settings(s)
+        print(green(f"→ 녹음 마이크: {value}"))
+
+    if arg:
+        _save(arg.strip())
+        return
+    devices = _list_audio_devices()
+    print(bold("🎙️ 오디오 입력 장치") + dim(f"  (현재: {current})"))
+    for idx, name in devices:
+        in_use = current != "default" and (idx == current or name == current or current in name)
+        mark = green(" ◀ 사용 중") if in_use else ""
+        print(f"  [{idx}] {name}{mark}")
+    print(f"  [d] 시스템 기본 입력 (macOS 사운드 설정을 따름)"
+          + (green(" ◀ 사용 중") if current == "default" else ""))
+    try:
+        choice = input(bold(cyan("  번호 선택 (Enter=유지): "))).strip().lower()
+    except (EOFError, KeyboardInterrupt):
+        print()
+        return
+    if not choice:
+        print(dim("  (유지)"))
+        return
+    if choice in ("d", "default"):
+        _save("default")
+        return
+    for idx, name in devices:
+        if choice == idx:
+            # 번호가 아니라 '이름'으로 저장 — 아이폰 연속성 마이크가 붙었다 떨어지면
+            # 장치 번호가 밀리는 실사례가 있어, 녹음 시점에 이름→현재 번호로 재해석한다.
+            _save(name)
+            return
+    print(red(f"  잘못된 선택: {choice}"))
+
+
+def _hear_binary():
+    """애플 SFSpeechRecognizer CLI(taemini-hear) 경로 — 없으면 swiftc로 1회 빌드."""
+    import shutil
+    import subprocess
+    tools = os.path.join(os.path.dirname(os.path.abspath(__file__)), "tools")
+    binary = os.path.join(tools, "taemini-hear")
+    if os.path.isfile(binary):
+        return binary
+    src = os.path.join(tools, "taemini_hear.swift")
+    if os.path.isfile(src) and shutil.which("swiftc"):
+        print(dim("🔧 애플 음성 인식기 도구를 처음 한 번 빌드합니다 (swiftc)..."))
+        r = subprocess.run(["swiftc", "-O", "-o", binary, src], capture_output=True)
+        if r.returncode == 0 and os.path.isfile(binary):
+            return binary
+        print(yellow("⚠️ 빌드 실패 — whisper 인식으로 진행합니다."))
+    return None
+
+
+def _record_apple(sess, binary):
+    """🍎 애플 네이티브 실시간 인식 — 말하는 즉시 글자가 자라고, 침묵(1.6초)이면 자동 확정.
+
+    반환: 인식 텍스트 / ""(무음) / None(엔진 오류 → 호출부가 whisper 폴백).
+    """
+    import select as _select
+    import subprocess
+    print(bold(cyan("🎤 듣고 있습니다...")) + dim("  (애플 실시간 인식 · 말이 끝나면 자동 전송, Enter=즉시)"))
+    try:
+        proc = subprocess.Popen([binary, "ko-KR", "1.6", str(VOICE_MAX_SEC)],
+                                stdout=subprocess.PIPE, stdin=subprocess.PIPE,
+                                text=True, bufsize=1)
+    except Exception as e:
+        print(yellow(f"⚠️ 애플 인식기 실행 실패({e}) — whisper로 폴백"))
+        return None
+    _term = sys.__stdout__ if (sys.__stdout__ and sys.__stdout__.isatty()) else sys.stdout
+
+    def _tw(t):
+        _term.write(t)
+        _term.flush()
+
+    state = {"final": None, "err": None}
+
+    def reader():
+        for line in proc.stdout:
+            line = line.rstrip("\n")
+            if line.startswith("P:"):
+                cap = "🎤 " + line[2:]
+                _tw(("\r\033[K" + green(cap)) if _USE_COLOR else ("\r" + cap))
+            elif line.startswith("F:"):
+                state["final"] = line[2:].strip()
+                return
+            elif line.startswith("E:"):
+                state["err"] = line[2:]
+                return
+
+    rt = threading.Thread(target=reader, daemon=True)
+    rt.start()
+    manual = False
+    while rt.is_alive():
+        try:
+            r, _, _ = _select.select([sys.stdin], [], [], 0.2)
+        except Exception:
+            break
+        if r:
+            sys.stdin.readline()
+            manual = True
+            try:
+                proc.stdin.close()  # Enter → hear가 지금까지 텍스트로 확정
+            except Exception:
+                pass
+    rt.join(timeout=5)
+    try:
+        proc.wait(timeout=5)
+    except Exception:
+        proc.kill()
+    if state["err"] is not None:
+        print(red(f"\n❌ 애플 인식기: {state['err']}"))
+        print(dim("   whisper 로컬 인식으로 폴백합니다."))
+        return None
+    text = " ".join((state["final"] or "").split())
+    _tw("\r\033[K" if _USE_COLOR else "\r")
+    if not text:
+        print(yellow("⚠️ 음성을 인식하지 못했습니다."))
+        return ""
+    tag = "" if manual else dim("  (침묵 감지 → 자동 전송)")
+    print(green("🎤 인식: ") + bold(text) + tag)
+    return text
+
+
+def record_and_transcribe(sess):
+    """마이크 녹음(Enter로 종료) → /api/transcribe 전사 → 인식 텍스트 반환(실패 시 None).
+
+    macOS에선 헤임달과 같은 애플 SFSpeechRecognizer(진짜 스트리밍)를 우선 사용하고,
+    빌드 불가/권한 거부 등일 때만 whisper(2초 조각) 방식으로 폴백한다.
+    """
+    import shutil
+    import signal
+    import subprocess
+    import tempfile
+    # 🍎 애플 네이티브 우선 (시뮬레이션 훅 TAEMINI_VOICE_INPUT은 whisper 경로로 검증)
+    if sys.platform == "darwin" and not os.environ.get("TAEMINI_VOICE_INPUT"):
+        binary = _hear_binary()
+        if binary:
+            result = _record_apple(sess, binary)
+            if result is not None:
+                return result or None  # ""(무음)이면 전송하지 않음
+    ffmpeg = shutil.which("ffmpeg")
+    if not ffmpeg:
+        print(red("❌ ffmpeg가 필요합니다 (brew install ffmpeg)."))
+        return None
+    sim_input = os.environ.get("TAEMINI_VOICE_INPUT")
+    if sim_input:
+        # 🧪 검증 훅: 오디오 파일을 실시간 속도(-re)로 마이크처럼 흘린다 (자동 테스트용)
+        in_args = ["-re", "-i", sim_input]
+    elif sys.platform == "darwin":
+        # ⚠️ ':0' 하드코딩 금지 — 아이폰 연속성 마이크가 0번을 차지하는 실사례.
+        # 설정값: 'default'(시스템 기본) / 숫자(레거시) / 장치 이름(권장 — 번호가 밀려도 안전).
+        mic = (load_settings().get("CLI_VOICE_MIC") or "default").strip()
+        if mic != "default" and not mic.isdigit():
+            resolved = None
+            for idx, name in _list_audio_devices():
+                if name == mic or mic in name:
+                    resolved = idx
+                    break
+            if resolved is None:
+                print(yellow(f"⚠️ 저장된 마이크('{mic}')를 찾지 못해 시스템 기본 입력을 사용합니다."))
+                mic = "default"
+            else:
+                mic = resolved
+        in_args = ["-f", "avfoundation", "-i", f":{mic}"]
+    elif sys.platform.startswith("linux"):
+        in_args = ["-f", "alsa", "-i", "default"]
+    else:
+        print(yellow("⚠️ 윈도우 마이크 녹음은 아직 미지원입니다 (macOS/Linux만)."))
+        return None
+    work_dir = tempfile.mkdtemp(prefix="taemini_voice_")
+    full_wav = os.path.join(work_dir, "full.wav")
+    seg_pattern = os.path.join(work_dir, "seg_%03d.wav")
+    # 출력 2개: ① 전체 녹음(최종 정확 전사용) ② 2초 조각(실시간 자막용)
+    cmd = [ffmpeg, "-y", *in_args,
+           "-ac", "1", "-ar", "16000", "-t", str(VOICE_MAX_SEC), full_wav,
+           "-f", "segment", "-segment_time", "1.2", "-reset_timestamps", "1",
+           "-ac", "1", "-ar", "16000", "-t", str(VOICE_MAX_SEC), seg_pattern]
+    mic_label = (load_settings().get("CLI_VOICE_MIC") or "default").strip()
+    if mic_label == "default":
+        mic_label = "시스템 기본 입력"
+    print(bold(cyan("🎤 듣고 있습니다...")) + dim(f"  말이 끝나면 자동 전송 (또는 Enter, 최대 {VOICE_MAX_SEC}초)"))
+    print(dim(f"   마이크: {mic_label}  ·  변경: /mic"))
+    try:
+        proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                                stdin=subprocess.DEVNULL)
+    except Exception as e:
+        print(red(f"❌ 녹음 시작 실패: {e}"))
+        return None
+
+    # 🗣️ 실시간 자막 워처 — 완성된 2초 조각을 즉시 small 모델로 받아 적어 표시
+    import glob as _glob
+    partials = []
+    stop_evt = threading.Event()
+    silence_evt = threading.Event()   # 말 끝(침묵) 감지 → 자동 전송
+    silent_chunks = [0]
+    first_slow_notice = [False]
+
+    # ⚠️ 흐름 모드의 patch_stdout은 '줄바꿈 없는 부분 출력'을 줄이 완성될 때까지
+    # 버퍼링한다 → \r 자막이 녹음 끝나고 한꺼번에 쏟아지던 실사례.
+    # 진짜 터미널(sys.__stdout__)에 직접 써서 말하는 즉시 화면에 나오게 한다.
+    _term = sys.__stdout__ if (sys.__stdout__ and sys.__stdout__.isatty()) else sys.stdout
+
+    def _tw(text):
+        _term.write(text)
+        _term.flush()
+
+    def _show_live():
+        line = "🎤 " + " ".join(partials)
+        _tw(("\r\033[K" + green(line)) if _USE_COLOR else ("\r" + line))
+
+    def watcher():
+        done = set()
+        while True:
+            segs = sorted(_glob.glob(os.path.join(work_dir, "seg_*.wav")))
+            # 마지막 파일은 아직 기록 중일 수 있음 — 녹음 종료 후엔 전부 처리
+            ready = segs if stop_evt.is_set() else segs[:-1]
+            for f in ready:
+                if f in done:
+                    continue
+                done.add(f)
+                t0 = time.time()
+                try:
+                    # 조각당 10초 상한 — 한 조각이 늦어도 자막 흐름을 막지 않는다 (건너뜀)
+                    r = post_json(sess.server, "/api/transcribe/quick",
+                                  {"path": f, "language": "ko", "model": "base"}, timeout=10)
+                    t = (r.get("text") or "").strip()
+                except Exception as _we:
+                    t = ""
+                    r = {"err": str(_we)}
+                dbg = os.environ.get("TAEMINI_DEBUG_LOG")
+                if dbg:
+                    with open(dbg, "a", encoding="utf-8") as _df:
+                        _df.write(f"[watcher] {os.path.basename(f)} {time.time()-t0:.2f}s -> {r}\n")
+                if time.time() - t0 > 8 and not first_slow_notice[0]:
+                    first_slow_notice[0] = True
+                    _tw("\n" + dim("   (최초 1회 small 모델 준비 중 — 다음부터는 즉시 표시됩니다)") + "\n")
+                if t:
+                    partials.append(t)
+                    silent_chunks[0] = 0
+                    _show_live()
+                elif partials and not stop_evt.is_set():
+                    # 말을 시작한 뒤 무음 조각 2개(~2.4초) 연속 → 말이 끝난 것으로 보고 자동 전송
+                    silent_chunks[0] += 1
+                    if silent_chunks[0] >= 2:
+                        silence_evt.set()
+            if stop_evt.is_set() and all(s in done for s in
+                                         _glob.glob(os.path.join(work_dir, "seg_*.wav"))):
+                return
+            time.sleep(0.3)
+
+    wt = threading.Thread(target=watcher, daemon=True)
+    wt.start()
+
+    # Enter(수동 종료) 또는 침묵 감지(자동 종료)를 함께 기다린다
+    import select as _select
+    auto_sent = False
+    try:
+        while True:
+            if silence_evt.is_set():
+                auto_sent = True
+                break
+            r, _, _ = _select.select([sys.stdin], [], [], 0.2)
+            if r:
+                sys.stdin.readline()
+                break
+    except (EOFError, KeyboardInterrupt):
+        pass
+    # SIGINT → ffmpeg가 파일을 정상 마무리(finalize)하고 종료
+    try:
+        proc.send_signal(signal.SIGINT)
+        proc.wait(timeout=5)
+    except Exception:
+        proc.terminate()
+    stop_evt.set()
+    wt.join(timeout=60)
+
+    try:
+        if not os.path.isfile(full_wav) or os.path.getsize(full_wav) < 2000:
+            print(red("❌ 녹음된 소리가 없습니다."))
+            print(dim("   ① 마이크 권한 확인: 시스템 설정 → 개인정보 보호 및 보안 → 마이크 → 터미널 허용"))
+            print(dim("   ② 다른 마이크가 잡혔을 수 있습니다 — 현재 장치 목록:"))
+            for idx, name in _list_audio_devices():
+                print(dim(f"      [{idx}] {name}"))
+            print(dim("      /mic 로 올바른 마이크를 선택해 주십시오."))
+            return None
+        # 최종 전사: 전체 파일을 한 번 더(조각 경계에서 끊긴 단어 보정) — small이라 1~3초
+        final_text = ""
+        try:
+            r = post_json(sess.server, "/api/transcribe/quick",
+                          {"path": full_wav, "language": "ko", "model": "small"}, timeout=180)
+            final_text = " ".join((r.get("text") or "").split())
+        except Exception:
+            pass
+        if not final_text:
+            final_text = " ".join(" ".join(partials).split())
+        _tw("\r\033[K" if _USE_COLOR else "\r")
+        if not final_text:
+            print(yellow("⚠️ 음성을 인식하지 못했습니다."))
+            return None
+        tag = dim("  (침묵 감지 → 자동 전송)") if auto_sent else ""
+        print(green("🎤 인식: ") + bold(final_text) + tag)
+        return final_text
+    finally:
+        import shutil as _sh
+        _sh.rmtree(work_dir, ignore_errors=True)
+
+__all__ = [_n for _n in list(globals()) if not _n.startswith("__")]

shadow_brain_core/taemini_cli_select.py

@@ -0,0 +1,192 @@
+"""🔱 태미니 CLI — 대화식 선택기·연결 모드 (taemini_cli_core 분할, 2026-07-10)
+
+화살표 메뉴(_arrow_choose)와 그 위에 얹힌 /model 2단 선택기, /server 연결 모드
+전환을 담는다. 신뢰 폴더 시작 선택창(_pick_menu)은 시작 게이트와 결합돼 있어
+core에 남겨뒀다. core의 색/키입력/설정 헬퍼를 그대로 쓴다.
+"""
+from taemini_cli_core import *  # noqa: F401,F403
+
+
+def _arrow_choose(options, title="", footer="↑↓ 이동 · Enter 선택 · 숫자 바로 선택 · Esc 취소"):
+    """범용 화살표 선택 메뉴 — options: [(값, 라벨)], 반환: 선택 값 또는 None(취소).
+    원시 키 입력 불가 환경(비TTY 등)은 숫자 입력 폴백."""
+    if not options:
+        return None
+    if title:
+        print(bold(title))
+    if footer:
+        print(dim(f"   {footer}"))
+    try:
+        old_attr = None
+        if os.name != "nt":
+            import termios
+            import tty
+            fd = sys.stdin.fileno()
+            old_attr = termios.tcgetattr(fd)  # 비TTY면 예외 → 숫자 폴백
+            tty.setcbreak(fd, termios.TCSANOW)  # setraw(TCSAFLUSH)는 대기 입력 폐기 — 금지
+        idx = 0
+        drawn = 0
+        sys.stdout.write("\033[?25l")
+        try:
+            while True:
+                if drawn:
+                    sys.stdout.write(f"\033[{drawn}A")
+                for i, (_v, label) in enumerate(options):
+                    line = f"   {i + 1}) {label}"
+                    if i == idx:
+                        line = f"\033[7m ❯ {i + 1}) {label}\033[0m"
+                    sys.stdout.write("\033[K" + line + "\n")
+                drawn = len(options)
+                sys.stdout.flush()
+                key = _read_menu_key()
+                if key == "up":
+                    idx = (idx - 1) % len(options)
+                elif key == "down":
+                    idx = (idx + 1) % len(options)
+                elif key == "enter":
+                    return options[idx][0]
+                elif key in ("esc", "q"):
+                    return None
+                elif key and key.isdigit() and 1 <= int(key) <= len(options):
+                    return options[int(key) - 1][0]
+        finally:
+            sys.stdout.write("\033[?25h")
+            sys.stdout.flush()
+            if old_attr is not None:
+                import termios
+                termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, old_attr)
+    except KeyboardInterrupt:
+        return None
+    except Exception:
+        pass  # 원시 키 불가 → 숫자 폴백
+    for i, (_v, label) in enumerate(options, start=1):
+        print(f"   {i}) {label}")
+    try:
+        ans = input("   선택 (엔터=취소) ").strip()
+    except (EOFError, KeyboardInterrupt):
+        ans = ""
+    if ans.isdigit() and 1 <= int(ans) <= len(options):
+        return options[int(ans) - 1][0]
+    return None
+
+
+def _fetch_providers(server):
+    """공급자/모델 목록 — 데몬 /api/shadow/config (헤임달 드롭다운과 동일 소스)
+    + 로컬 Ollama(11434)가 살아 있으면 실제 설치 모델로 upsert."""
+    provs = []
+    try:
+        with urllib.request.urlopen(f"{server}/api/shadow/config", timeout=6) as r:
+            data = json.loads(r.read().decode("utf-8"))
+        provs = [p for p in data.get("providers", [])
+                 if isinstance(p, dict) and p.get("id") and p.get("models")]
+    except Exception:
+        pass
+    try:
+        with urllib.request.urlopen("http://127.0.0.1:11434/api/tags", timeout=2) as r:
+            tags = [m.get("name") for m in json.loads(r.read().decode("utf-8")).get("models", [])]
+        tags = [t for t in tags if t]
+        if tags:
+            provs = [p for p in provs if p.get("id") != "ollama"]
+            provs.append({"id": "ollama", "name": "🦙 Ollama (로컬)", "models": tags})
+    except Exception:
+        pass
+    return provs
+
+
+def cmd_model_picker(server):
+    """/model (인자 없음) — 공급자 → 모델 2단 화살표 선택."""
+    cur_prov = (_config_get("provider") or "").strip()
+    cur_model = (_config_get("model") or "").strip()
+    provs = _fetch_providers(server)
+    if cur_prov and cur_prov not in [p["id"] for p in provs]:
+        provs.append({"id": cur_prov, "name": cur_prov,
+                      "models": [cur_model] if cur_model else []})
+    if not provs:
+        print(yellow("공급자 목록을 가져오지 못했습니다 — /model <모델ID> 로 직접 지정하세요."))
+        return
+    opts = [(p["id"], f'{p.get("name", p["id"])} ({p["id"]})'
+             + ("  ✓ 현재" if p["id"] == cur_prov else "")) for p in provs]
+    pid = _arrow_choose(opts, title="🧠 AI 공급자 선택")
+    if pid is None:
+        print(dim("변경 없음."))
+        return
+    models = next((p.get("models") or [] for p in provs if p["id"] == pid), [])
+    mopts = [(m, m + ("  ✓ 현재" if pid == cur_prov and m == cur_model else "")) for m in models]
+    mopts.append(("__manual__", "직접 입력…"))
+    model = _arrow_choose(mopts, title=f"📦 모델 선택 — {pid}")
+    if model is None:
+        print(dim("변경 없음."))
+        return
+    if model == "__manual__":
+        try:
+            model = input("   모델 ID: ").strip()
+        except (EOFError, KeyboardInterrupt):
+            model = ""
+        if not model:
+            print(dim("변경 없음."))
+            return
+    _config_set("provider", pid)
+    _config_set("model", model)
+    print(green(f"→ 로컬 태미니: {pid}/{model}") + dim(" (데몬 재시작 후 반영)"))
+
+
+def cmd_server(sess, arg):
+    """/server [local|cloud|URL] — 연결 모드 전환 (헤임달 로컬/클라우드와 동일 개념).
+    인자 없으면 화살표 메뉴. 전환은 저장되어 다음 시작에도 유지."""
+    arg = (arg or "").strip()
+    local_url = "http://127.0.0.1:18700"
+    sel = None
+    if not arg:
+        is_loc = _is_local_server(sess.server)
+        cur_cloud = (not is_loc) and sess.server.rstrip("/") == cloud_url()
+        sel = _arrow_choose([
+            ("local",  f"🏠 로컬 데몬   {local_url}" + ("  ✓ 현재" if is_loc else "")),
+            ("cloud",  f"☁️ 클라우드    {cloud_url()}" + ("  ✓ 현재" if cur_cloud else "")),
+            ("manual", "직접 URL 입력…"),
+        ], title="🌐 쉐도우 브레인 연결 모드")
+        if sel is None:
+            print(dim("변경 없음."))
+            return
+    else:
+        sel = arg.lower() if arg.lower() in ("local", "cloud") else "url"
+    if sel == "manual":
+        try:
+            arg = input("   서버 URL: ").strip()
+        except (EOFError, KeyboardInterrupt):
+            arg = ""
+        if not arg:
+            print(dim("변경 없음."))
+            return
+        sel = "url"
+    target = {"local": local_url, "cloud": cloud_url()}.get(sel, arg).rstrip("/")
+    if not target.startswith(("http://", "https://")):
+        target = "https://" + target
+    print(dim(f"   연결 확인 중… {target}"))
+    if not daemon_alive(target, timeout=6):
+        if _is_local_server(target):
+            if not ensure_daemon(target):
+                print(red("전환 취소 — 로컬 데몬을 켤 수 없습니다."))
+                return
+        else:
+            print(red(f"전환 취소 — 서버 응답 없음: {target}"))
+            return
+    sess.server = target
+    if _is_local_server(target):
+        sess.trusted = is_trusted_dir(sess.cwd)
+        print(green(f"🏠 로컬 데몬으로 전환: {target}"))
+    else:
+        # 원격 컨테이너에는 이 폴더가 없다 — 코딩 도구 OFF(대화 전용)
+        sess.trusted = False
+        print(green(f"☁️ 클라우드로 전환: {target}"))
+        print(dim("   대화 전용 모드 — 파일 수정·명령 실행(코딩 도구)은 로컬 모드에서만 동작합니다."))
+        if not (load_settings().get("TAEMINI_CLOUD_KEY") or "").strip():
+            print(yellow("   ⚠️ cloud-key 미설정 — 게이트가 켜진 서버면 401이 납니다: /config set cloud-key <키>"))
+    s = load_settings()
+    s["TAEMINI_SERVER_MODE"] = "local" if _is_local_server(target) else "cloud"
+    if not _is_local_server(target):
+        s["TAEMINI_CLOUD_URL"] = target
+    save_settings(s)
+    print(dim("   이 모드는 저장되어 다음 taemini 시작에도 유지됩니다."))
+
+
+__all__ = [_n for _n in list(globals()) if not _n.startswith("__")]

shadow_brain_core/utils/emotion.py

@@ -72,6 +72,9 @@ def invalidate_guidelines_cache() -> None:
     _emotion_logger.info("[Guidelines] 캐시 무효화 — 다음 조회에서 최신 지침을 다시 가져옵니다.")
 
 
+_FETCH_FAIL_CACHE: dict = {}  # aegis_url → 실패 시각 (네거티브 캐시)
+
+
 def _fetch_guidelines_from_aegis(aegis_url: str) -> dict | None:
     """Aegis REST API에서 공통 지침을 가져옵니다. 실패 시 None 반환.
 
@@ -79,6 +82,11 @@ def _fetch_guidelines_from_aegis(aegis_url: str) -> dict | None:
     401 → None → 하드코딩 폴백이 쓰여, '공통 지침 관리자' 편집이 브레인에 전혀 반영되지
     않는다. 반드시 서비스 토큰을 실어 보낸다(저장 경로 get_job_perf_flags와 동일 패턴)."""
     global _RUNTIME_TTL
+    # 🛡️ 실패 네거티브 캐시 — 401/네트워크 실패가 반복되는 동안 모든 소비자
+    # (retrieval_policy/emotion/readability)가 매 턴 클라우드 왕복을 반복하지 않게.
+    _fail_at = _FETCH_FAIL_CACHE.get(aegis_url)
+    if _fail_at and (time.time() - _fail_at) < _eff_ttl():
+        return None
     try:
         import os
         import requests
@@ -97,9 +105,12 @@ def _fetch_guidelines_from_aegis(aegis_url: str) -> dict | None:
                     _RUNTIME_TTL = float(data['cache_ttl_seconds'])
                 except (TypeError, ValueError):
                     pass
+            _FETCH_FAIL_CACHE.pop(aegis_url, None)
             return data  # {"readability": "...", "emotion": "...", "cache_ttl_seconds": N}
+        _emotion_logger.debug(f"[Guidelines] Aegis HTTP {resp.status_code} — 네거티브 캐시")
     except Exception as e:
         _emotion_logger.debug(f"[Guidelines] Aegis 조회 실패 (폴백 사용): {e}")
+    _FETCH_FAIL_CACHE[aegis_url] = time.time()
     return None
 
 
@@ -116,6 +127,10 @@ def get_dynamic_retrieval_policy(aegis_url: str | None = None) -> dict:
     if isinstance(value, dict):
         _GUIDELINES_CACHE[cache_key] = {'value': value, 'ts': time.time()}
         return value
+    # 🛡️ 네거티브 캐시 — Aegis 실패/누락도 TTL 동안 기억. 안 하면 의도 감지 헬퍼
+    # (_tool_policy/_intent_cfg 등)가 매 턴 클라우드 왕복(~840ms)을 반복해 TTFT를
+    # 수 초씩 태운다(2026-07-10 실측).
+    _GUIDELINES_CACHE[cache_key] = {'value': {}, 'ts': time.time()}
     return {}
 
 

shadow_brain_core/verify_grok_relay.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""로컬 Grok 릴레이 단위·연동 스모크 검증 (데몬 없이도 모듈 경로 검증 가능)."""
+import json
+import os
+import sys
+
+ROOT = os.path.dirname(os.path.abspath(__file__))
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)
+
+passed = 0
+failed = 0
+
+
+def ok(name, cond, detail=""):
+    global passed, failed
+    if cond:
+        passed += 1
+        print(f"[OK] {name}" + (f" — {detail}" if detail else ""))
+    else:
+        failed += 1
+        print(f"[FAIL] {name}" + (f" — {detail}" if detail else ""))
+
+
+def main():
+    # 1) 모듈 로드
+    from blueprints.relays import grok as grok_mod
+    ok("import grok relay module", True)
+
+    # 2) CLI 탐지
+    cmd = grok_mod._grok_command()
+    ok("grok CLI installed", cmd is not None, cmd or "not found")
+
+    # 3) JSON 파서
+    sample = json.dumps({
+        "text": "HELLO",
+        "sessionId": "019f4500-aaaa-bbbb-cccc-ddddeeeeffff",
+        "stopReason": "EndTurn",
+    })
+    text, sid = grok_mod._parse_grok_json(sample)
+    ok("parse json text", text == "HELLO", text)
+    ok("parse json sessionId", sid and sid.startswith("019f"), sid)
+
+    # 4) 라이브 헤드리스 (CLI 있을 때만)
+    if cmd:
+        content, session_id = grok_mod._run_grok_single(
+            command=cmd,
+            prompt="Reply with exactly: GROK_VERIFY_OK",
+            cwd="/tmp",
+            model="",
+            session_id=None,
+        )
+        ok("headless single-turn", "GROK_VERIFY_OK" in (content or ""), content[:80] if content else "")
+        ok("session id returned", bool(session_id), session_id or "")
+        if session_id:
+            content2, session_id2 = grok_mod._run_grok_single(
+                command=cmd,
+                prompt="Repeat the exact token from my previous message (GROK_VERIFY_OK).",
+                cwd="/tmp",
+                model="",
+                session_id=session_id,
+            )
+            ok(
+                "resume continuity",
+                "GROK_VERIFY_OK" in (content2 or "") or session_id2 == session_id,
+                (content2 or "")[:120],
+            )
+
+    # 5) VALID_RELAYS
+    from local_guardians import VALID_RELAYS
+    ok("VALID_RELAYS has grok", "grok" in VALID_RELAYS)
+
+    # 6) taemini CLI 맵
+    import taemini_cli as cli
+    ok("CLI GUARDIAN_LABELS grok", "grok" in cli.GUARDIAN_LABELS)
+    ok("CLI alias 그록", cli.resolve_guardian("그록") == "grok")
+    ok("CLI alias xai", cli.resolve_guardian("xai") == "grok")
+
+    print()
+    print(f"passed={passed} failed={failed}")
+    return 0 if failed == 0 else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

shadow_brain_core/verify_taemini_cli.py

@@ -122,6 +122,15 @@ def main():
         if not cond:
             failures.append(name)
 
+    # 0) 🚧 God-File 상한 래칫 (2026-07-10 분할: cli 2688 / core 789 / media 529)
+    #    상한을 넘기면 새 기능은 파일 추가/모듈 이동으로 배치할 것 (AGENTS.md §13-C 정신).
+    _CAPS = {"taemini_cli.py": 2800, "taemini_cli_core.py": 950,
+             "taemini_cli_media.py": 650, "taemini_cli_select.py": 400}
+    for _f, _cap in _CAPS.items():
+        _p = os.path.join(ROOT, _f)
+        _n = sum(1 for _ in open(_p, encoding="utf-8")) if os.path.exists(_p) else 0
+        check(f"파일 상한 {_f} ≤ {_cap}", 0 < _n <= _cap, f"{_n}줄")
+
     # 1) --once 태미니 경로: 본문 + 진행판 + 통계
     p = run_cli(["--once", "안녕 태미니", "--cwd", ROOT], server=server)
     check("태미니 응답 본문 렌더링", "STUB_TAEMINI_REPLY" in p.stdout, p.stdout[-400:])

shadow_brain_core/web_server_prelude.py

@@ -0,0 +1,535 @@
+# -*- coding: utf-8 -*-
+"""web_server 모듈 수준 헬퍼/로깅 (God-File 분해).
+
+create_web_server 및 라우트는 web_server.py 에 유지.
+"""
+
+import os
+import re
+import secrets
+import json
+import uuid
+import base64
+import random
+import logging
+import time
+import subprocess
+import sys
+from collections import deque
+from datetime import datetime, timezone, timedelta
+import threading
+import importlib.util
+import yaml # Added for frontmatter parsing
+try:
+    import pyautogui
+    HAS_GUI = True
+except Exception:
+    # 🔱 Headless Cloud Fallback
+    HAS_GUI = False
+    pyautogui = None
+import requests
+from flask import Flask, render_template, jsonify, send_from_directory, request, Response, stream_with_context
+from dotenv import load_dotenv
+from google import genai
+from google.genai import types
+
+logger = logging.getLogger(__name__)
+
+
+def _sanitize_a2ui_for_relay(text: str) -> str:
+    """디스코드/텔레그램 같은 비대화형 릴레이용으로 A2UI 카드 블록을 평문화한다.
+
+    헤임달 Flutter 전용 ```imperial-*``` 블록(미디어 그리드/유튜브/지도/선택카드)은
+    외부 채널에서 렌더링되지 않고 날것 JSON으로 노출되므로:
+      - imperial-media / imperial-youtube / imperial-3d / imperial-map(s):
+        블록 안의 URL만 추출해 평문 줄로 남긴다(디스코드가 자동 임베드).
+      - imperial-ask 및 그 외 imperial-* 블록: 그냥 제거.
+    """
+    if not text or '```imperial-' not in text:
+        return (text or '').strip()
+
+    url_kinds = ('media', 'youtube', '3d', 'map', 'maps')
+
+    def _extract_urls(block: str):
+        seen = []
+        # imperial-media는 JSON이라 item별 대표 URL(url→preview→page)만 골라낸다.
+        try:
+            inner = re.sub(r'^```imperial-[\w-]+\s*|\s*```$', '', block.strip())
+            data = json.loads(inner)
+            for it in (data.get('items') or []):
+                u = it.get('url') or it.get('preview') or it.get('page') or it.get('download')
+                if u and u not in seen:
+                    seen.append(u)
+        except Exception:
+            pass
+        if not seen:
+            for u in re.findall(r'https?://[^\s"\'\\)]+', block):
+                if u not in seen:
+                    seen.append(u)
+        return seen[:5]
+
+    def _repl(m):
+        kind = (m.group(1) or '').lower()
+        if kind in url_kinds:
+            urls = _extract_urls(m.group(0))
+            return ('\n' + '\n'.join(urls) + '\n') if urls else ''
+        return ''
+
+    text = re.sub(r'```imperial-([\w-]+)[\s\S]*?```', _repl, text).strip()
+    # [TOOL_USE: ...] 패턴 제거 — 릴레이 채널(디스코드/텔레그램)에서 날것 노출 방지
+    text = re.sub(r'\[TOOL_USE:[^\]]*\]\s*', '', text).strip()
+    return text
+
+
+from utils.emotion import extract_emotion as _extract_emotion_from_text, strip_emotion_tags as _strip_emotion_tags
+
+
+# [🔱 Imperial v2.5.2] Standardized Path Resolution (PyInstaller EXE 호환)
+# PyInstaller로 패키징된 EXE 실행 시 __file__은 임시 폴더를 가리킴.
+# sys.frozen이 True이면 실제 EXE 위치(sys.executable)를 기준으로 경로를 잡습니다.
+if getattr(sys, 'frozen', False):
+    # EXE 모드: dist/ShadowBrain/ 내부가 core_root
+    core_root = os.path.dirname(os.path.abspath(sys.executable))
+else:
+    # 일반 Python 실행 모드
+    core_root = os.path.dirname(os.path.abspath(__file__))
+project_root = os.path.dirname(core_root) # One level up to d:\Git_Work\jarvis_taemin
+
+# [🔱 Imperial Writable Data Root]
+# 배포본(.exe)은 Program Files 등 쓰기가 막힌 보호 폴더에 설치된다. 이 경우 project_root가
+# Program Files를 가리켜 docker_data/로그/설정 생성 시 PermissionError(WinError 5)로 부팅이
+# 멈춘다. frozen(.exe) 또는 LOCAL_AGENT_MODE에서는 project_root를 쓰기 가능한 사용자 폴더
+# (%LOCALAPPDATA%\TaeminGames\ShadowBrain)로 강제 리다이렉트한다. (읽기용 core_root는 유지)
+def _resolve_writable_data_root():
+    explicit = os.environ.get("IMPERIAL_DATA_ROOT")
+    if explicit:
+        base = explicit
+    else:
+        local_base = os.environ.get("LOCALAPPDATA", os.path.expanduser("~"))
+        base = os.path.join(local_base, "TaeminGames", "ShadowBrain")
+    try:
+        os.makedirs(base, exist_ok=True)
+    except Exception:
+        return None
+    return base
+
+if getattr(sys, 'frozen', False) or os.environ.get("LOCAL_AGENT_MODE", "false").lower() == "true":
+    _writable_root = _resolve_writable_data_root()
+    if _writable_root:
+        project_root = _writable_root
+        print(f"[SYSTEM] -> Writable data root: {project_root}", flush=True)
+
+def _read_shadow_version() -> str:
+    """VERSION 파일을 읽어 버전 문자열을 반환합니다."""
+    version_path = os.path.join(core_root, "VERSION")
+    if os.path.exists(version_path):
+        try:
+            with open(version_path, 'r', encoding='utf-8') as f:
+                return f.read().strip()
+        except: pass
+    return "2.6.72" # Fallback
+
+# Add core directory and tools to sys.path before any local imports
+sys.path.append(core_root) 
+sys.path.append(os.path.join(project_root, "scripts", "tools"))
+
+# [🔱 Imperial] Persistent Console Title
+def set_imperial_title(title_name):
+    if sys.platform == "win32":
+        import ctypes
+        try:
+            ctypes.windll.kernel32.SetConsoleTitleW(title_name)
+        except Exception:
+            pass
+    else:
+        sys.stdout.write(f"\033]2;{title_name}\007")
+        sys.stdout.flush()
+
+print("[SYSTEM] Booting Imperial Shadow Brain Core...", flush=True)
+set_imperial_title("SHADOW BRAIN (Core Intelligence) - Port 18700")
+
+
+# Load environment variables from .env if it exists
+load_dotenv()
+
+# [🔱 Imperial Local Agent Mode Auto-Detection]
+# 1. sys.frozen이 True이면 PyInstaller로 빌드된 .exe 배포본 상태입니다.
+# 2. 배포본은 기본적으로 '로컬 에이전트 모드(Stateless)'로 작동하도록 자동 설정합니다.
+if getattr(sys, 'frozen', False):
+    LOCAL_AGENT_MODE = True
+    os.environ["LOCAL_AGENT_MODE"] = "true" # [🔱] 하위 모듈(sovereign_memory 등) 인식을 위해 환경변수 강제 주입
+    print("[SYSTEM] 🔱 Imperial Standalone Executable (.exe) detected.", flush=True)
+    print("[SYSTEM] -> Auto-activating Local Agent Mode (Stateless).", flush=True)
+else:
+    # 소스 코드로 실행 중일 때만 .env 설정을 따릅니다.
+    LOCAL_AGENT_MODE = os.environ.get("LOCAL_AGENT_MODE", "false").lower() == "true"
+
+# Imperial Core Modules
+print("[SYSTEM] -> Loading Imperial Core Modules...", flush=True)
+import gdrive_sync
+import cloud_tool
+import phantom_sandbox
+# psycopg2는 설치 환경에서만 활성화 (Stateless 배포 호환). 미설치 시 DB 엔드포인트만 비활성.
+try:
+    import psycopg2
+except ImportError:
+    psycopg2 = None
+from werkzeug.security import check_password_hash, generate_password_hash
+from sovereign_task_manager import task_manager
+import quota_manager
+from quota_manager import ImperialQuotaManager, CopilotUsageManager
+from aegis_link import neural_link
+from sovereign_browser_agent import SovereignMissionAgent
+from imperial_r2_uploader import ImperialR2Uploader
+from compaction import compaction_engine
+print("[SYSTEM] -> Core Modules Loaded.", flush=True)
+
+
+from guardian_monitor import guardian_monitor
+
+from db_util import get_db_connection
+from telegram_service import telegram_bot
+from discord_service import discord_bot
+from tasks_util import execute_shell_command, _post_to_gbb, _analyze_sync_failure
+print("[SYSTEM] -> Loading Blueprints...", flush=True)
+from blueprints.system_routes import system_bp
+from blueprints.auth_gate import auth_gate_bp  # 🔐 공개 배포(HF) 인증 게이트
+from blueprints.media_routes import media_bp
+from blueprints.chat_routes import chat_bp, init_chat_logic
+from blueprints.admin_routes import admin_bp, init_admin_logic
+from blueprints.agent_routes import agent_bp, init_agent_logic
+from blueprints.security_routes import security_bp, init_security_logic
+from blueprints.deployment_routes import deploy_bp, init_deploy_logic
+from blueprints.report_routes import report_bp, init_report_logic
+from blueprints.parallel_routes import parallel_bp, init_parallel_logic
+from blueprints.mcp_routes import mcp_bp
+from blueprints.transcribe_routes import transcribe_bp  # [🔱 미디어 전사 API] /api/transcribe/*
+from blueprints.obsidian_routes import obsidian_bp  # [🗂️ 옵시디언 지식 레이어] /api/obsidian/* — 로컬 전용
+print("[SYSTEM] -> Blueprints Loaded.", flush=True)
+
+from services.brain_service import init_service_logic, get_brain_response_sync
+import services.brain_service as brain_service
+from compaction import CompactionEngine
+from sovereign_memory import memory_write_vector
+
+
+# Imperial Remote Vision Service & Sovereign Engine
+try:
+    import remote_vision_service
+    import sovereign_engine
+    import sovereign_memory
+except ImportError as e:
+    print(f"[SYSTEM] -> Remote services loading warning: {e}")
+except Exception as e:
+    print(f"[SYSTEM] -> Core engine initialization warning: {e}")
+
+# --- [🔱 Imperial Log Monitoring] ---
+class ImperialLogHandler(logging.Handler):
+    """[🔱 Imperial] Custom logging handler to pipe all logs into the global buffer."""
+    def __init__(self, log_queue):
+        super().__init__()
+        self.log_queue = log_queue
+        # Use a consistent timestamp format same as StreamTee
+        self.setFormatter(logging.Formatter('[%(asctime)s] %(name)s: %(message)s', datefmt='%H:%M:%S'))
+
+    def emit(self, record):
+        try:
+            msg = self.format(record)
+            self.log_queue.append(msg)
+        except Exception:
+            self.handleError(record)
+
+class StreamTee:
+    """[🔱 Imperial] Custom stream wrapper to copy stdout/stderr to the global log buffer."""
+    def __init__(self, stream, log_queue):
+        self.stream = stream
+        self.log_queue = log_queue
+
+    def write(self, data):
+        self.stream.write(data)
+        if isinstance(data, bytes):
+            try:
+                data = data.decode('utf-8', errors='replace')
+            except:
+                return # Skip if undecodable
+        
+        text = data.strip()
+        if text:
+            # [🔱 Imperial] Mute periodic polling logs from stdout capture
+            # Stricter filtering to prevent polling loop visibility
+            poll_keywords = [
+                "/api/logs/shadow-brain", 
+                "/api/system/monitor/r2", 
+                "/api/status", 
+                "/health",
+                "GET /api/logs", 
+                "method: get"
+            ]
+            if any(kw in text for kw in poll_keywords):
+                return
+
+            # Format like a log message for consistency
+            timestamp = datetime.now().strftime('%H:%M:%S')
+            self.log_queue.append(f"[{timestamp}] [STDOUT] {text}")
+
+    def flush(self):
+        self.stream.flush()
+
+SHADOW_BRAIN_LOGS = deque(maxlen=30)
+
+def capture_logs():
+    """Starts capturing stdout, stderr, and root logger into the global log queue."""
+    # 1. Pipe Standard Output
+    sys.stdout = StreamTee(sys.stdout, SHADOW_BRAIN_LOGS)
+    sys.stderr = StreamTee(sys.stderr, SHADOW_BRAIN_LOGS)
+    
+    # 2. Add Logging Handler to Root
+    root_logger = logging.getLogger()
+    handler = ImperialLogHandler(SHADOW_BRAIN_LOGS)
+    root_logger.addHandler(handler)
+    logger.info("🔱 Imperial Neural Log Stream Integrated.")
+
+capture_logs()
+
+# --- [Imperial Core] Dynamic R2 Resolution Helpers ---
+def get_imperial_r2_account_name():
+    """Returns the R2 account name based on the current environment."""
+    app_env = os.environ.get("APP_ENV", "local").lower()
+    return "live" if app_env == "live" else "vault"
+
+
+# --- [🔱 Imperial Health Check] ---
+# --- [🔱 Imperial Context Caching] ---
+_IMPERIAL_CONTEXT_CACHE = {
+    "agents": "",
+    "run_info": "",
+    "last_mtime_agents": 0,
+    "last_mtime_run": 0
+}
+
+def _load_imperial_context():
+    """
+    [🔱 High-Performance Comprehensive Context Loader - V3 Final]
+    제국의 모든 정체성(수호자, 아키텍처, 바하무트 등 시스템 자산)을 AGENTS.md와 JARVISRUN.md에서 정밀하게 추출합니다.
+    """
+    global _IMPERIAL_CONTEXT_CACHE
+    
+    agents_path = os.path.join(project_root, "AGENTS.md")
+    run_info_path = os.path.join(project_root, "JARVISRUN.md")
+    
+    try:
+        # 1. AGENTS.md 파싱 (수호자 명단 ~ 하단 전체)
+        # 9번 섹션(수호자 연합) 이후의 모든 규약과 정보를 포함하여 시스템 정체성 확립
+        if os.path.exists(agents_path):
+            current_mtime = os.path.getmtime(agents_path)
+            if current_mtime > _IMPERIAL_CONTEXT_CACHE["last_mtime_agents"]:
+                with open(agents_path, 'r', encoding='utf-8') as f:
+                    content = f.read()
+                    match = re.search(r'## 9\. 🔱 황실 수호자 연합.*', content, re.DOTALL)
+                    if match:
+                        _IMPERIAL_CONTEXT_CACHE["agents"] = match.group(0).strip()
+                    else:
+                        _IMPERIAL_CONTEXT_CACHE["agents"] = content[:3000] # 최후의 수단
+                _IMPERIAL_CONTEXT_CACHE["last_mtime_agents"] = current_mtime
+        
+        # 2. JARVISRUN.md 파싱 (아키텍처 ~ 끝까지)
+        # 3번 섹션(아키텍처) 및 4번(수호자 위계)을 포함하여 바하무트 등 시스템 엔티티 인지
+        if os.path.exists(run_info_path):
+            current_mtime = os.path.getmtime(run_info_path)
+            if current_mtime > _IMPERIAL_CONTEXT_CACHE["last_mtime_run"]:
+                with open(run_info_path, 'r', encoding='utf-8') as f:
+                    content = f.read()
+                    # '## 3. 아키텍처' 이후의 모든 내용을 가져와서 시스템 구조를 학습시킴
+                    match = re.search(r'## 3\. 아키텍처.*', content, re.DOTALL)
+                    if match:
+                        _IMPERIAL_CONTEXT_CACHE["run_info"] = match.group(0).strip()
+                    else:
+                        _IMPERIAL_CONTEXT_CACHE["run_info"] = ""
+                _IMPERIAL_CONTEXT_CACHE["last_mtime_run"] = current_mtime
+                
+    except Exception as e:
+        print(f"[🔱 Context Loader] Error loading context: {e}")
+        
+    return f"{_IMPERIAL_CONTEXT_CACHE['agents']}\n\n{_IMPERIAL_CONTEXT_CACHE['run_info']}"
+
+def get_imperial_r2_root():
+    """Returns the base public URL for the current environment's R2 account."""
+    from cloud_tool import R2_ACCOUNTS
+    acc_name = get_imperial_r2_account_name()
+    return R2_ACCOUNTS.get(acc_name, {}).get("public_url", "")
+
+def resolve_imperial_profile_url(p_url):
+    """Resolves a relative profile URL to a full R2 URL based on environment."""
+    if not p_url: return p_url
+    if p_url.startswith('/images/portraits/'):
+        root = get_imperial_r2_root()
+        return f"{root.rstrip('/')}/{p_url.lstrip('/')}"
+    return p_url
+
+# --- [🔱 Imperial Core] Shadow Brain Engine Port 18700 ---
+# Logic and Management have been consolidated to Django (Port 18701).
+# This engine now focuses on High-Performance Chat, Agency, and GPU tasks.
+
+# get_db_connection is imported from db_util
+
+# [REMOVED] ensure_db_schema_ready is now handled by Django Fortress (18701)
+
+
+
+# Global store for GitHub OAuth sessions
+
+GITHUB_AUTH_SESSIONS = {}
+
+
+
+# Load environment variables from .env if it exists
+
+load_dotenv()
+
+
+
+# Disable Flask logging for a cleaner terminal
+
+log = logging.getLogger('werkzeug')
+
+log.setLevel(logging.ERROR)
+
+
+
+# --- GitHub Copilot Logic ---
+
+# Caches the session-based Copilot token
+
+GITHUB_COPILOT_CACHE = {
+
+    "token": None,
+
+    "expires_at": 0,
+
+    "base_url": "https://api.individual.githubcopilot.com"
+
+}
+
+
+
+# Caches the last few proactive messages to avoid repetition
+
+
+
+
+
+# Phase 41: Graceful Reintegration State
+# RAG Management Globals (Debounce & Thread Safety)
+rag_timers = {}    # {guardian: Timer}
+rag_pending = {}   # {guardian: (content, metadata)}
+rag_lock = threading.Lock()
+
+is_syncing = False
+
+
+
+# Proactive Chat Configuration Default
+
+CHAT_CONTEXT_LIMIT = 50 # [IMPERIAL] Default context window size (overridden by JARVIS_SHADOW_SETTINGS.json)
+
+def _get_chat_context_limit(project_root_path: str) -> int:
+    """JARVIS_SHADOW_SETTINGS.json에서 context_limit을 읽어 Hot-Reload로 반환합니다."""
+    try:
+        settings_file = os.path.join(project_root_path, "docker_data", "shared_workspace", "JarvisRun", "JARVIS_SHADOW_SETTINGS.json")
+        if os.path.exists(settings_file):
+            with open(settings_file, 'r', encoding='utf-8') as f:
+                data = json.load(f)
+            val = data.get("context_limit")
+            if isinstance(val, int) and 5 <= val <= 500:
+                return val
+    except Exception:
+        pass
+    return CHAT_CONTEXT_LIMIT
+
+
+
+PROACTIVE_CONFIG_FILE = None # Will be set in setup_project
+
+# Proactive Topic Mapping (Korean to English Key)
+
+# --- [🔱 Imperial Media Services] (Moved to media_service.py) ---
+from services.media_service import fetch_global_trends, fetch_youtube_trending, fetch_youtube_by_keyword, fetch_joke
+
+def get_imperial_context(project_root, guardian_id="jarvis"):
+    """Gathers situational and hierarchical awareness for the Shadow Brain, supporting OpenClaw-style Soul logic."""
+    context = f"\n[IMPERIAL SITUATIONAL AWARENESS - TARGET: {guardian_id.upper()}]\n"
+
+    # 1. Determine Workspace Path
+    # Default: OpenClaw workspace for Aizen/Gilgamesh, or specialized folders for others
+    workspace_map = {
+        "jarvis": os.path.join(project_root, "docker_data", "shared_workspace", "JarvisRun"),
+        "aris": os.path.join(project_root, "docker_data", "shared_workspace", "Aris")
+    }
+    
+    workspace_path = workspace_map.get(guardian_id.lower())
+    
+    # If no specific workspace, check if a folder exists in docker_data/shared_workspace
+    if not workspace_path:
+        potential_path = os.path.join(project_root, "docker_data", "shared_workspace", guardian_id.capitalize())
+        if os.path.exists(potential_path):
+            workspace_path = potential_path
+
+    # 2. Inject Soul Bridge (IDENTITY.md, SOUL.md, USER.md)
+    if workspace_path and os.path.exists(workspace_path):
+        soul_files = ["IDENTITY.md", "SOUL.md", "USER.md", "AGENTS.md", "BOOTSTRAP.md", "TOOLS.md"]
+        context += f"--- [{guardian_id.upper()}] SOUL CLUSTER (Workspace: {os.path.basename(workspace_path)}) ---\n"
+        for soul_file in soul_files:
+            file_path = os.path.join(workspace_path, soul_file)
+            if os.path.exists(file_path):
+                try:
+                    with open(file_path, 'r', encoding='utf-8') as f:
+                        context += f"\n[{soul_file}]\n{f.read()}\n"
+                except Exception as e:
+                    context += f"\n[{soul_file} LOAD ERROR: {e}]\n"
+    else:
+        # Fallback to legacy SKILL.md if no workspace found
+        persona_file = os.path.join(project_root, ".agent", "skills", "jarvis_guardian_personas", "SKILL.md")
+        if os.path.exists(persona_file):
+            try:
+                with open(persona_file, 'r', encoding='utf-8') as f:
+                    context += "--- 제국 수호자 페르소나 (LEGACY SKILL.md) ---\n" + f.read() + "\n"
+            except Exception as e:
+                context += f"--- 제국 수호자 페르소나 로드 실패: {e} ---\n"
+
+    # 3. Inject Imperial Permanent Rules (AGENTS.md from root)
+    # This is a global rule file that applies to all
+    agents_rules_file = os.path.join(project_root, "AGENTS.md")
+    if os.path.exists(agents_rules_file):
+        try:
+            with open(agents_rules_file, 'r', encoding='utf-8') as f:
+                context += "--- 제국 영구 운영 규약 (GLOBAL AGENTS.md) ---\n" + f.read() + "\n"
+        except Exception as e:
+            context += f"--- 제국 운영 규약 로드 실패: {e} ---\n"
+
+    context += f"현재 서버 시간: {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
+    return context
+
+
+def load_guardian_emails():
+    path = os.path.join(project_root, 'docker_data', 'shared_workspace', 'JarvisRun', 'JARVIS_GUARDIAN_EMAILS.json')
+    if not os.path.exists(path):
+        return {}
+    try:
+        with open(path, 'r', encoding='utf-8') as f:
+            return json.load(f)
+    except Exception as e:
+        logger.error(f"[Emails] Load error: {e}")
+        return {}
+
+def save_guardian_emails(emails):
+    path = os.path.join(project_root, 'docker_data', 'shared_workspace', 'JarvisRun', 'JARVIS_GUARDIAN_EMAILS.json')
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    try:
+        with open(path, 'w', encoding='utf-8') as f:
+            json.dump(emails, f, indent=4, ensure_ascii=False)
+        return True
+    except Exception as e:
+        logger.error(f"[Emails] Save error: {e}")
+        return False
+
+