shadow_brain_core/brain/coding_tools.py

# -*- coding: utf-8 -*-
"""
🔱 [Imperial Coding Arsenal] 쉐도우 브레인 에이전틱 코딩 도구 세트

Claude Code 방식의 코딩 워크플로(탐색 → 읽기 → 수정 → 검증 → 반복)를
쉐도우 브레인에 이식하기 위한 도구 모음입니다.

- Gemini AFC 경로: CodingToolkit의 public 메서드를 callable로 직접 등록
- OpenAI 호환 경로(Atlas/OpenRouter): OPENAI_SCHEMAS + execute() 디스패처
"""
import fnmatch
import hashlib
import json
import os
import re
import shutil
import subprocess
import logging
import threading
import time

logger = logging.getLogger("ShadowBrain.Coding")

# 작업 루트 — 이 바깥의 파일은 건드리지 않는다 (안전 가드)
DEFAULT_ROOT = r"D:\Git_Work\jarvis_taemin"

# 🔱 [Per-Request Root Override] 헤임달 '태미니 코드'가 고른 프로젝트 폴더(cwd)를
# 해당 요청 스레드에서만 작업 루트로 쓰기 위한 스레드 로컬 오버라이드.
# think() 도구 루프는 요청 스레드 안에서 동기적으로 돌므로 thread-local이 정확히 격리한다.
_ROOT_OVERRIDE = threading.local()


def set_coding_root_override(path: str):
    """현재 요청 스레드의 코딩 작업 루트를 지정한다 (요청 종료 시 clear 필수)."""
    _ROOT_OVERRIDE.value = os.path.abspath(path) if path else None


def clear_coding_root_override():
    _ROOT_OVERRIDE.value = None


# ─────────────────────────────────────────────────────────────────
# 🧭 [Per-Turn Agent State] 태미니 코딩 턴 상태 (스레드 로컬, 루트 오버라이드와 동일 수명)
#   - read_files : 이번 턴에 code_read_file 로 읽은 파일 집합 → read-before-edit 강제
#   - plan       : 📐 플랜 모드 — 파일 수정/생성·상태 변경 명령을 '도구 레벨'에서 차단
#   - changes    : 이번 턴에 수정/생성한 파일 목록 [{path, action}] → CLI 변경 요약
#   - session    : 세션 키 → 체크포인트(undo) 디렉토리 귀속
# 릴레이(relays/taemini.py)가 턴 시작에 begin_coding_turn, 종료에 end_coding_turn을
# 호출한다. 이 상태가 없으면(타 채널의 코딩 모드) 가드는 전부 비활성 — 기존 동작 유지.
# ─────────────────────────────────────────────────────────────────
_TURN_STATE = threading.local()

CHECKPOINT_BASE = os.path.join(os.path.expanduser("~"), ".shadow_brain", "checkpoints")
_CHECKPOINT_KEEP = 20  # 세션당 보관하는 턴 체크포인트 수


def begin_coding_turn(session_key: str = "", plan_mode: bool = False, user_id: str = "anonymous"):
    _TURN_STATE.value = {
        "session": (session_key or "").strip(),
        "plan": bool(plan_mode),
        "uid": (user_id or "anonymous").strip() or "anonymous",
        "read_files": set(),
        "changes": [],
        "ckpt_dir": None,
        "backed_up": set(),
    }


def end_coding_turn():
    """턴 상태를 걷고 이번 턴 변경 파일 목록 [{path, action}] 을 반환한다.
    같은 파일 다회 수정은 1건으로 합치되 'create'가 있으면 그것을 우선 표기."""
    st = getattr(_TURN_STATE, "value", None)
    _TURN_STATE.value = None
    seen = {}
    for ch in (st or {}).get("changes") or []:
        prev = seen.get(ch["path"])
        if prev is None or ch["action"] == "create":
            seen[ch["path"]] = ch["action"]
    return [{"path": p, "action": a} for p, a in seen.items()]


def _turn():
    return getattr(_TURN_STATE, "value", None)


def _session_ckpt_root(session_key: str) -> str:
    h = hashlib.sha1((session_key or "default").encode("utf-8")).hexdigest()[:16]
    return os.path.join(CHECKPOINT_BASE, h)


def _checkpoint_before_change(full_path: str, root: str):
    """수정/생성 직전 원본 백업 — /undo(턴 단위 롤백)의 재료.
    턴당 파일별 최초 1회만 백업(원본 보존). 실패는 조용히 무시(편집을 막지 않는다)."""
    st = _turn()
    if st is None or not st.get("session"):
        return
    try:
        if full_path in st["backed_up"]:
            return
        st["backed_up"].add(full_path)
        if st["ckpt_dir"] is None:
            sess_root = _session_ckpt_root(st["session"])
            os.makedirs(sess_root, exist_ok=True)
            # 오래된 턴 체크포인트 정리 (세션당 최신 _CHECKPOINT_KEEP개 유지)
            turns = sorted(d for d in os.listdir(sess_root)
                           if os.path.isdir(os.path.join(sess_root, d)))
            if len(turns) >= _CHECKPOINT_KEEP:
                for old in turns[:len(turns) - (_CHECKPOINT_KEEP - 1)]:
                    shutil.rmtree(os.path.join(sess_root, old), ignore_errors=True)
            st["ckpt_dir"] = os.path.join(sess_root, str(int(time.time() * 1000)))
            os.makedirs(st["ckpt_dir"], exist_ok=True)
        manifest_path = os.path.join(st["ckpt_dir"], "manifest.json")
        try:
            with open(manifest_path, "r", encoding="utf-8") as f:
                manifest = json.load(f) or []
        except Exception:
            manifest = []
        entry = {"path": full_path, "backup": None}
        if os.path.isfile(full_path):
            if os.path.getsize(full_path) > 5 * 1024 * 1024:
                return  # 5MB 초과 파일은 백업 생략 (undo 대상에서 제외)
            bak = f"{len(manifest):04d}.bak"
            shutil.copy2(full_path, os.path.join(st["ckpt_dir"], bak))
            entry["backup"] = bak
        manifest.append(entry)
        with open(manifest_path, "w", encoding="utf-8") as f:
            json.dump(manifest, f, ensure_ascii=False)
    except Exception as ck_err:
        logger.warning(f"[Coding Checkpoint] 백업 실패(편집은 계속): {ck_err}")


def _record_change(full_path: str, root: str, action: str):
    st = _turn()
    if st is None:
        return
    try:
        rel = os.path.relpath(full_path, root)
    except Exception:
        rel = full_path
    st["changes"].append({"path": rel, "action": action})


def undo_last_coding_turn(session_key: str):
    """세션의 가장 최근 턴 체크포인트를 복원한다 (Claude Code /rewind 등가).
    반환: 복원된 항목 리스트 [{path, action(restored|deleted)}]. 없으면 빈 리스트."""
    sess_root = _session_ckpt_root(session_key)
    if not os.path.isdir(sess_root):
        return []
    turns = sorted((d for d in os.listdir(sess_root)
                    if os.path.isdir(os.path.join(sess_root, d))), reverse=True)
    for turn in turns:
        turn_dir = os.path.join(sess_root, turn)
        manifest_path = os.path.join(turn_dir, "manifest.json")
        try:
            with open(manifest_path, "r", encoding="utf-8") as f:
                manifest = json.load(f) or []
        except Exception:
            shutil.rmtree(turn_dir, ignore_errors=True)
            continue
        restored = []
        for entry in reversed(manifest):
            path = entry.get("path") or ""
            bak = entry.get("backup")
            try:
                if bak:
                    bak_full = os.path.join(turn_dir, os.path.basename(bak))
                    if os.path.isfile(bak_full):
                        shutil.copy2(bak_full, path)
                        restored.append({"path": path, "action": "restored"})
                elif os.path.isfile(path):
                    os.remove(path)  # 이번 턴에 새로 만든 파일 → 삭제로 복원
                    restored.append({"path": path, "action": "deleted"})
            except Exception as undo_err:
                logger.warning(f"[Coding Undo] 복원 실패 {path}: {undo_err}")
        shutil.rmtree(turn_dir, ignore_errors=True)
        if restored:
            return restored
    return []


# ─────────────────────────────────────────────────────────────────
# 📚 [자가 무장] 스킬 자가 생성/장착 — 에이전트가 필요한 전문 지식을 스스로 벼려 입는다.
# 스킬 볼트: <repo_root>/.agent/skills/<name>/SKILL.md (직업 장착 스킬과 동일 보관소).
# 세션 장착: local_guardian_store 'session_skills:<sessionKey>' — 릴레이가 턴마다 읽어
# 시스템 지시에 주입한다. 직업(job)의 장착 목록은 절대 건드리지 않는다(직업=마왕님
# 소관, 폴백 금지 헌장 보존) — 세션 장착은 그 위에 얹히는 자율 계층이다.
# ─────────────────────────────────────────────────────────────────
_SKILL_NAME_RE = re.compile(r"^[\w가-힣][\w가-힣 .-]{0,60}$")
_SESSION_SKILLS_CAP = 6  # 세션당 장착 상한 (초과 시 오래된 것부터 밀려남)


def _skills_dir():
    override = os.environ.get("SHADOW_SKILLS_DIR")
    if override:
        return override
    repo_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
    return os.path.join(repo_root, ".agent", "skills")


def list_vault_skills(limit: int = 40):
    """볼트에 존재하는 스킬 이름 목록 (SKILL.md가 실재하는 폴더만)."""
    try:
        d = _skills_dir()
        return sorted(n for n in os.listdir(d)
                      if os.path.isfile(os.path.join(d, n, "SKILL.md")))[:limit]
    except Exception:
        return []


def session_skill_names(user_id: str, session_key: str):
    """세션에 장착된 스킬 이름 목록 (스토어 기록 순서 유지)."""
    if not session_key:
        return []
    try:
        from local_guardian_store import get_setting
        names = json.loads(get_setting(user_id or "anonymous", "local_taemini",
                                       f"session_skills:{session_key}", "[]") or "[]")
        return [str(n) for n in names] if isinstance(names, list) else []
    except Exception:
        return []


def _equip_to_session(st, safe_name: str):
    """세션 장착 기록 공용 — 성공 시 None, 실패 시 오류 문자열 (forge/equip 공유)."""
    try:
        from local_guardian_store import get_setting, set_setting
        key = f"session_skills:{st['session']}"
        uid = st.get("uid") or "anonymous"
        names = json.loads(get_setting(uid, "local_taemini", key, "[]") or "[]")
        if not isinstance(names, list):
            names = []
        if safe_name in names:
            names.remove(safe_name)
        names.append(safe_name)
        names = names[-_SESSION_SKILLS_CAP:]
        set_setting(uid, "local_taemini", key, json.dumps(names, ensure_ascii=False))
        return None
    except Exception as e:
        return str(e)


def load_session_skills(user_id: str, session_key: str, max_chars_each: int = 1500):
    """세션 장착 스킬 [(name, content), ...] — 릴레이가 턴마다 호출해 프롬프트에 주입.
    파일을 매번 새로 읽는다(자가 갱신 즉시 반영 — 직업 스킬 캐시와 달리 무캐시)."""
    out = []
    for name in session_skill_names(user_id, session_key):
        safe = os.path.basename(name.strip())
        path = os.path.join(_skills_dir(), safe, "SKILL.md")
        try:
            with open(path, "r", encoding="utf-8") as f:
                text = f.read().strip()
            if len(text) > max_chars_each:
                text = text[:max_chars_each] + "\n…(이하 생략)"
            out.append((safe, text))
        except Exception:
            continue
    return out


# 📐 플랜 모드에서 code_run_command 로 허용하는 '읽기성' 명령 접두 (상태 불변경)
_PLAN_SAFE_CMD_PREFIXES = (
    "git status", "git diff", "git log", "git show", "git branch", "git blame",
    "ls", "dir", "cat ", "head ", "tail ", "grep ", "rg ", "find ", "wc ",
    "tree", "pwd", "file ", "stat ", "du ", "type ", "where ", "which ",
    "python -m py_compile", "python3 -m py_compile",
)

_PLAN_BLOCK_MSG = ("📐 플랜 모드: 지금은 계획 수립 단계라 {what}이(가) 도구 레벨에서 "
                   "차단되어 있습니다. 이 시도는 실행되지 않았습니다 — 절대 '수정/실행 완료'라고 "
                   "보고하지 마십시오(거짓 보고 금지). 계획(단계별 todos)만 정리해 보고하고, "
                   "실행은 마왕님이 일반 모드로 전환한 뒤에 진행합니다.")


def _loopback_headers():
    """자기 자신(127.0.0.1 데몬)으로의 HTTP 루프백용 인증 헤더 — 설치별 로컬 키.
    (auth_gate가 로컬 요청에도 X-Taemini-Key를 요구하는 강화 환경 대응 —
    2026-07-14 전문가 소환 401 실사고, delegate_to_guardian도 동일 경로)"""
    try:
        with open(os.path.join(os.path.expanduser("~"), ".taemini", "shadow_local_token"),
                  "r", encoding="utf-8") as f:
            tok = f.read().strip()
        return {"X-Taemini-Key": tok} if tok else {}
    except Exception:
        return {}

# 검색/탐색 시 무시할 디렉토리
_SKIP_DIRS = {
    ".git", "node_modules", "venv", ".venv", "__pycache__", "dist", "build",
    ".next", ".dart_tool", "site-packages", ".pub-cache", "docker_data",
}

# 명백히 파괴적인 명령 차단 (마왕님 PC 보호)
_BLOCKED_CMD_PATTERNS = [
    r"\bformat\b", r"\brm\s+-rf\s+/", r"\bdel\s+/s\s+/q\s+[a-z]:\\\s*$",
    r"remove-item\s+-recurse\s+-force\s+[a-z]:\\\s*$", r"\bshutdown\b", r"\bmkfs\b",
]

_TEXT_EXT_HINT = {
    ".py", ".dart", ".js", ".ts", ".tsx", ".jsx", ".json", ".yaml", ".yml",
    ".md", ".txt", ".html", ".css", ".sh", ".ps1", ".bat", ".rb", ".erb",
    ".sql", ".toml", ".ini", ".cfg", ".xml", ".csv", ".env", ".mjs", ".cjs",
}


class CodingToolkit:
    """파일 탐색/읽기/수정/검색/명령 실행 도구. 모든 경로는 작업 루트 기준 상대/절대 둘 다 허용."""

    def __init__(self, root: str = None):
        self._root = os.path.abspath(root or os.environ.get("SHADOW_CODING_ROOT", DEFAULT_ROOT))

    @property
    def root(self) -> str:
        """요청 스레드에 오버라이드(cwd)가 있으면 그 폴더, 없으면 기본 루트."""
        override = getattr(_ROOT_OVERRIDE, "value", None)
        return override if override else self._root

    # ── 내부 헬퍼 ──────────────────────────────────────────────
    def _resolve(self, path: str) -> str:
        p = (path or "").strip().strip('"')
        if not p or p in (".", "./"):
            return self.root
        if not os.path.isabs(p):
            p = os.path.join(self.root, p)
        p = os.path.abspath(p)
        if os.path.commonpath([p, self.root]) != self.root:
            raise ValueError(f"작업 루트({self.root}) 바깥 경로는 허용되지 않습니다: {p}")
        return p

    @staticmethod
    def _is_text_file(path: str) -> bool:
        ext = os.path.splitext(path)[1].lower()
        if ext in _TEXT_EXT_HINT:
            return True
        try:
            with open(path, "rb") as f:
                chunk = f.read(2048)
            return b"\x00" not in chunk
        except Exception:
            return False

    def _walk(self):
        for dirpath, dirnames, filenames in os.walk(self.root):
            dirnames[:] = [d for d in dirnames if d not in _SKIP_DIRS and not d.startswith(".")]
            yield dirpath, filenames

    # ── 도구: 탐색 ─────────────────────────────────────────────
    def code_list_dir(self, path: str = ".") -> str:
        """Lists files and subdirectories at the given path (relative to the project root). Use this first to understand project structure.
        Args:
            path: Directory path relative to project root (default: root)
        """
        try:
            target = self._resolve(path)
            if not os.path.isdir(target):
                return f"Error: 디렉토리가 아닙니다: {path}"
            entries = sorted(os.listdir(target))
            lines = []
            for e in entries:
                full = os.path.join(target, e)
                if os.path.isdir(full):
                    lines.append(f"[DIR]  {e}/")
                else:
                    try:
                        size = os.path.getsize(full)
                    except OSError:
                        size = 0
                    lines.append(f"[FILE] {e} ({size:,} bytes)")
            rel = os.path.relpath(target, self.root)
            return f"# {rel} ({len(entries)} entries)\n" + "\n".join(lines[:300])
        except Exception as e:
            return f"Error: {e}"

    def code_glob(self, pattern: str, max_results: int = 100) -> str:
        """Finds files matching a glob pattern (e.g. '**/*.py', 'portal-heimdall/lib/**/*.dart'). Returns matching paths relative to project root.
        Args:
            pattern: Glob pattern. '**' matches any directories.
            max_results: Maximum number of paths to return (default 100)
        """
        try:
            pattern = (pattern or "").replace("\\", "/").lstrip("/")
            results = []
            for dirpath, filenames in self._walk():
                rel_dir = os.path.relpath(dirpath, self.root).replace("\\", "/")
                for fn in filenames:
                    rel = fn if rel_dir == "." else f"{rel_dir}/{fn}"
                    # '**/x' 패턴이 루트 바로 아래 파일도 잡도록 보정
                    if fnmatch.fnmatch(rel, pattern) or (
                        pattern.startswith("**/") and fnmatch.fnmatch(rel, pattern[3:])
                    ):
                        results.append(rel)
                        if len(results) >= max_results:
                            return "\n".join(results) + f"\n... (max {max_results} reached)"
            return "\n".join(results) if results else f"No files match: {pattern}"
        except Exception as e:
            return f"Error: {e}"

    def code_grep(self, pattern: str, file_glob: str = None, max_results: int = 50) -> str:
        """Searches file CONTENTS for a regex pattern across the project. Returns 'path:line: text' matches. Essential for finding where something is defined or used.
        Args:
            pattern: Regular expression to search for
            file_glob: Optional filename filter (e.g. '*.py', '*.dart')
            max_results: Maximum matching lines to return (default 50)
        """
        try:
            rx = re.compile(pattern)
        except re.error as e:
            return f"Error: 잘못된 정규식: {e}"
        results = []
        try:
            for dirpath, filenames in self._walk():
                for fn in filenames:
                    if file_glob and not fnmatch.fnmatch(fn, file_glob):
                        continue
                    full = os.path.join(dirpath, fn)
                    if not self._is_text_file(full):
                        continue
                    rel = os.path.relpath(full, self.root)
                    try:
                        with open(full, "r", encoding="utf-8", errors="ignore") as f:
                            for i, line in enumerate(f, 1):
                                if rx.search(line):
                                    results.append(f"{rel}:{i}: {line.rstrip()[:200]}")
                                    if len(results) >= max_results:
                                        return "\n".join(results) + f"\n... (max {max_results} reached)"
                    except OSError:
                        continue
            return "\n".join(results) if results else f"No matches for: {pattern}"
        except Exception as e:
            return f"Error: {e}"

    # ── 도구: 읽기/쓰기 ────────────────────────────────────────
    def code_read_file(self, path: str, offset: int = 1, limit: int = 400) -> str:
        """Reads a text file with line numbers. ALWAYS read a file before editing it.
        Args:
            path: File path (relative to project root or absolute)
            offset: 1-based line number to start from (default 1)
            limit: Max lines to return (default 400)
        """
        try:
            full = self._resolve(path)
            if not os.path.isfile(full):
                return f"Error: 파일이 없습니다: {path}"
            with open(full, "r", encoding="utf-8", errors="replace") as f:
                lines = f.readlines()
            st = _turn()
            if st is not None:
                st["read_files"].add(full)  # read-before-edit 가드 해제
            total = len(lines)
            start = max(int(offset or 1), 1)
            end = min(start - 1 + int(limit or 400), total)
            body = "".join(
                f"{i:>5}\t{lines[i - 1]}" for i in range(start, end + 1)
            )
            return f"# {path} (lines {start}-{end} of {total})\n{body}"
        except Exception as e:
            return f"Error: {e}"

    def code_write_file(self, path: str, content: str) -> str:
        """Creates or fully overwrites a file with the given content (UTF-8). For small changes to existing files, prefer code_edit_file.
        Args:
            path: File path (relative to project root or absolute)
            content: Full file content to write
        """
        try:
            full = self._resolve(path)
            st = _turn()
            if st is not None and st.get("plan"):
                return _PLAN_BLOCK_MSG.format(what="파일 생성/덮어쓰기(code_write_file)")
            existed = os.path.isfile(full)
            if (existed and st is not None and full not in st["read_files"]
                    and os.path.basename(full) not in _READ_EXEMPT_FILES):
                return ("Error: 기존 파일을 읽지 않고 통째로 덮어쓰는 것은 금지입니다. "
                        f"먼저 code_read_file('{path}') 로 현재 내용을 확인한 뒤, "
                        "부분 수정이면 code_edit_file 을 사용하십시오.")
            _checkpoint_before_change(full, self.root)
            os.makedirs(os.path.dirname(full), exist_ok=True)
            with open(full, "w", encoding="utf-8", newline="") as f:
                f.write(content or "")
            if st is not None:
                st["read_files"].add(full)
            _record_change(full, self.root, "write" if existed else "create")
            verb = "덮어씀" if existed else "새 파일 생성"
            return f"OK: wrote {len(content or '')} chars to {path} ({verb})"
        except Exception as e:
            return f"Error: {e}"

    @staticmethod
    def _edit_failure_hint(lines, old_string) -> str:
        """old_string 불일치 시 자가교정 힌트 — 첫 유효 줄과 가장 비슷한 위치의 실제
        원문(줄번호 포함)을 보여줘 모델이 정확한 원문을 다시 복사하게 한다.
        정확/포함 일치 → difflib 유사도(≥0.55) 순으로 탐색 (오타·공백 차이 방어)."""
        import difflib
        first = next((l.strip() for l in (old_string or "").splitlines() if l.strip()), "")
        if not first:
            return ""
        best_i, best_r = -1, 0.55
        for i, line in enumerate(lines[:8000]):
            ls = line.strip()
            if not ls:
                continue
            if ls == first or (len(first) > 12 and first in line):
                best_i, best_r = i, 1.0
                break
            r = difflib.SequenceMatcher(None, first, ls).ratio()
            if r > best_r:
                best_i, best_r = i, r
        if best_i >= 0:
            lo, hi = max(0, best_i - 2), min(len(lines), best_i + 7)
            snippet = "".join(f"{n + 1:>5}\t{lines[n]}" for n in range(lo, hi))
            return ("\n비슷한 위치의 실제 원문(참고):\n" + snippet
                    + "위 원문을 들여쓰기·공백까지 그대로 old_string에 복사해 다시 시도하십시오.")
        return "\n(파일에서 비슷한 줄도 찾지 못했습니다 — code_grep 으로 위치부터 다시 확인하십시오.)"

    @staticmethod
    def _snippet_around(new_text, pos, added_lines, path) -> str:
        """편집 지점 주변 미리보기 — 모델이 재읽기 없이 결과를 확인하게 한다."""
        lines = new_text.splitlines(keepends=True)
        line_no = new_text[:pos].count("\n")  # 0-based
        lo = max(0, line_no - 2)
        hi = min(len(lines), line_no + added_lines + 3)
        if hi - lo > 14:
            hi = lo + 14
        return "".join(f"{n + 1:>5}\t{lines[n]}" for n in range(lo, hi))

    def code_edit_file(self, path: str, old_string: str, new_string: str, replace_all: bool = False) -> str:
        """Edits a file by exact string replacement. old_string must match the file content EXACTLY (including indentation) and be unique unless replace_all=true. Read the file first to copy the exact text.
        Args:
            path: File path (relative to project root or absolute)
            old_string: Exact text to find (must be unique in the file unless replace_all)
            new_string: Replacement text
            replace_all: Replace every occurrence (default false)
        """
        try:
            full = self._resolve(path)
            if not os.path.isfile(full):
                return f"Error: 파일이 없습니다: {path}"
            st = _turn()
            if st is not None and st.get("plan"):
                return _PLAN_BLOCK_MSG.format(what="파일 수정(code_edit_file)")
            if (st is not None and full not in st["read_files"]
                    and os.path.basename(full) not in _READ_EXEMPT_FILES):
                return (f"Error: 읽지 않은 파일은 수정할 수 없습니다. 먼저 code_read_file('{path}') 로 "
                        "수정할 부분의 실제 원문을 확인하십시오 (이번 턴 기준).")
            with open(full, "r", encoding="utf-8") as f:
                text = f.read()
            if old_string not in text:
                return ("Error: old_string을 파일에서 찾지 못했습니다."
                        + self._edit_failure_hint(text.splitlines(keepends=True), old_string))
            count = text.count(old_string)
            if count > 1 and not replace_all:
                return f"Error: old_string이 {count}번 등장합니다. 더 길고 고유한 문자열을 쓰거나 replace_all=true를 지정하세요."
            _checkpoint_before_change(full, self.root)
            pos = text.index(old_string)
            new_text = text.replace(old_string, new_string) if replace_all else text.replace(old_string, new_string, 1)
            with open(full, "w", encoding="utf-8", newline="") as f:
                f.write(new_text)
            _record_change(full, self.root, "edit")
            n = count if replace_all else 1
            preview = self._snippet_around(new_text, pos, (new_string or "").count("\n"), path)
            return (f"OK: {path} — {n}곳 수정 완료\n수정 결과 미리보기(재읽기 불필요):\n{preview}")
        except Exception as e:
            return f"Error: {e}"

    # ── 도구: 명령 실행(검증) ──────────────────────────────────
    def code_run_command(self, command: str, cwd: str = None, timeout: int = 120) -> str:
        """Runs a shell command (PowerShell on Windows) and returns stdout/stderr/exit code. Use to verify changes: run tests, compile checks (python -m py_compile, flutter analyze), git status, etc.
        Args:
            command: The command line to execute
            cwd: Working directory relative to project root (default: project root)
            timeout: Seconds before the command is killed (default 120, max 600)
        """
        try:
            low = (command or "").lower()
            for pat in _BLOCKED_CMD_PATTERNS:
                if re.search(pat, low):
                    return f"Error: 파괴적 명령이 차단되었습니다: {command}"
            st = _turn()
            if st is not None and st.get("plan"):
                stripped = low.strip()
                if not any(stripped.startswith(p) for p in _PLAN_SAFE_CMD_PREFIXES):
                    return _PLAN_BLOCK_MSG.format(what=f"상태 변경 가능 명령 실행({command[:60]})")
            workdir = self._resolve(cwd) if cwd else self.root
            timeout = min(max(int(timeout or 120), 5), 600)
            if os.name == "nt":
                cmd = ["powershell", "-NoProfile", "-NonInteractive", "-Command", command]
            else:
                cmd = ["bash", "-lc", command]
            proc = subprocess.run(
                cmd, cwd=workdir, capture_output=True, timeout=timeout,
            )
            out = proc.stdout.decode("utf-8", errors="replace")
            err = proc.stderr.decode("utf-8", errors="replace")
            res = f"[exit code: {proc.returncode}]\n"
            if out.strip():
                res += f"--- stdout ---\n{out[-6000:]}\n"
            if err.strip():
                res += f"--- stderr ---\n{err[-3000:]}\n"
            return res.strip()
        except subprocess.TimeoutExpired:
            return f"Error: 명령이 {timeout}초 안에 끝나지 않아 중단했습니다: {command}"
        except Exception as e:
            return f"Error: {e}"

    # ── 도구: 📚 자가 무장 (스킬 자가 생성/장착) ────────────────
    def code_forge_skill(self, name: str, content: str) -> str:
        """Creates or updates a reusable SKILL document in the shared skill vault. Use when the task needs domain knowledge/procedures no equipped skill covers — distill the know-how (commands, pitfalls, steps), then equip it with code_equip_skill.
        Args:
            name: Skill name (Korean/English letters, digits, space, ._- up to 60 chars)
            content: The skill document body (markdown). Min 40 chars.
        """
        st = _turn()
        if st is not None and st.get("plan"):
            return _PLAN_BLOCK_MSG.format(what="스킬 생성(code_forge_skill)")
        safe = (name or "").strip()
        # basename으로 조용히 잘라내지 않는다 — 경로 문자가 섞인 이름은 명시적 거부
        if not safe or safe != os.path.basename(safe) or not _SKILL_NAME_RE.match(safe):
            return "Error: 스킬 이름은 한글/영문/숫자/공백/._- 60자 이내여야 합니다 (경로 문자 금지)."
        body = (content or "").strip()
        if len(body) < 40:
            return "Error: 스킬 내용이 너무 짧습니다(40자 미만) — 재사용 가치가 있는 절차/지식을 담으십시오."
        body = body[:20000]
        try:
            d = os.path.join(_skills_dir(), safe)
            os.makedirs(d, exist_ok=True)
            with open(os.path.join(d, "SKILL.md"), "w", encoding="utf-8") as f:
                f.write(body + "\n")
        except Exception as e:
            return f"Error: 스킬 저장 실패: {e}"
        try:  # 직업 장착 캐시 무효화 — 같은 이름 스킬 갱신이 즉시 반영되게
            from services.brain_service import LOGIC_STATE
            b = LOGIC_STATE.get("jarvis_brain")
            if b is not None and hasattr(b, "_skill_cache"):
                b._skill_cache.pop(safe, None)
        except Exception:
            pass
        # ⚔️ 저장 즉시 이 세션에 자동 장착 — "만들고 장착하겠다"고 말만 하고 equip을
        # 안 부르는 약한 모델 실관측(2026-07-13) 때문에 도구 의미론을 forge=생성+장착으로.
        equip_err = _equip_to_session(st, safe) if (st and st.get("session")) else "세션 없음"
        if equip_err:
            return (f"OK: 📚 스킬 '{safe}' 저장({len(body)}자). "
                    f"(세션 장착은 안 됨: {equip_err} — code_equip_skill 로 장착 가능)")
        return (f"OK: 📚 스킬 '{safe}' 저장({len(body)}자) + 이 세션에 장착 완료 — "
                "다음 턴부터 자동 주입됩니다.")

    def code_equip_skill(self, name: str) -> str:
        """Equips a vault skill onto THIS session — its content is injected into your system context every following turn, and returned right now for immediate use. If the name doesn't match, available skills are listed.
        Args:
            name: Skill name to equip (as stored in the vault)
        """
        st = _turn()
        if st is None or not st.get("session"):
            return "Error: 세션 정보가 없어 장착할 수 없습니다 (태미니 코딩 콘솔 턴에서만 지원)."
        safe = os.path.basename((name or "").strip())
        path = os.path.join(_skills_dir(), safe, "SKILL.md")
        if not os.path.isfile(path):
            avail = list_vault_skills(20)
            return ("Error: 그런 스킬이 없습니다: " + safe
                    + (f"\n사용 가능: {', '.join(avail)}" if avail
                       else "\n(볼트가 비어 있습니다 — code_forge_skill 로 만드십시오.)"))
        equip_err = _equip_to_session(st, safe)
        if equip_err:
            return f"Error: 장착 기록 실패: {equip_err}"
        try:
            with open(path, "r", encoding="utf-8") as f:
                text = f.read().strip()[:1500]
        except Exception:
            text = "(내용 읽기 실패)"
        return (f"OK: 📚 '{safe}' 장착 완료 — 다음 턴부터 자동 주입됩니다. "
                f"지금 바로 활용할 내용:\n{text}")

    # ── 도구: 🔧 자작 도구 (스크립트 제작 → 스킬화) ─────────────
    def code_forge_tool(self, name: str, script: str, usage: str) -> str:
        """Creates a self-made TOOL: a Python script saved in the skill vault, plus a SKILL doc describing exactly how to run it. Auto-equipped. Re-forge the same name to improve it (evolution). Run it later with code_run_command.
        Args:
            name: Tool name (letters/digits/space/._- up to 60 chars)
            script: Python source code (syntax-checked before saving)
            usage: How/when to use it — arguments, examples, caveats
        """
        st = _turn()
        if st is not None and st.get("plan"):
            return _PLAN_BLOCK_MSG.format(what="도구 생성(code_forge_tool)")
        safe = (name or "").strip()
        if not safe or safe != os.path.basename(safe) or not _SKILL_NAME_RE.match(safe):
            return "Error: 도구 이름은 한글/영문/숫자/공백/._- 60자 이내여야 합니다 (경로 문자 금지)."
        src = (script or "").strip()
        if len(src) < 40:
            return "Error: 스크립트가 너무 짧습니다(40자 미만)."
        try:
            compile(src, f"{safe}.py", "exec")  # 문법 검사 — 깨진 도구를 볼트에 넣지 않는다
        except SyntaxError as se:
            return (f"Error: 파이썬 문법 오류(line {se.lineno}): {se.msg}\n"
                    "스크립트를 고쳐 같은 이름으로 다시 forge 하십시오.")
        d = os.path.join(_skills_dir(), safe)
        tool_path = os.path.join(d, "tool.py")
        try:
            os.makedirs(d, exist_ok=True)
            with open(tool_path, "w", encoding="utf-8") as f:
                f.write(src[:30000] + "\n")
            doc = (f"# 🔧 자작 도구: {safe}\n\n"
                   f"실행: `python3 \"{tool_path}\" [인자...]` (code_run_command 로 호출)\n\n"
                   f"## 사용법\n{(usage or '').strip()[:3000]}\n\n"
                   f"## 개선\n부정확하면 code_forge_tool('{safe}', 수정 스크립트, 사용법)으로 덮어써 진화시킬 것.\n")
            with open(os.path.join(d, "SKILL.md"), "w", encoding="utf-8") as f:
                f.write(doc)
        except Exception as e:
            return f"Error: 도구 저장 실패: {e}"
        equip_err = _equip_to_session(st, safe) if (st and st.get("session")) else "세션 없음"
        tail = "" if equip_err else " + 사용법 스킬 장착 완료(다음 턴부터 자동 주입)"
        return (f"OK: 🔧 자작 도구 '{safe}' 저장(문법 검사 통과){tail}.\n"
                f"실행: code_run_command(\"python3 \\\"{tool_path}\\\" ...\")")

    # ── 도구: 🧠 장기기억 (SQLite 벡터 + 옵시디언 볼트 동시 기록) ──
    MEMORY_CONTEXT = "taemini-lessons"
    MEMORY_NOTE = "Knowledge/태미니 장기기억.md"

    def code_remember(self, lesson: str, topic: str = "") -> str:
        """Engraves a reusable lesson/fact into LONG-TERM memory so future sessions recall it: SQLite vector memory (semantic recall, auto-injected as [장기기억 회상]) + the local Obsidian vault note. Use for cross-project lessons; workspace facts go to TAEMINI.md instead.
        Args:
            lesson: The lesson/fact worth remembering (min 20 chars)
            topic: Short topic tag (optional)
        """
        st = _turn()
        body = (lesson or "").strip()
        if len(body) < 20:
            return "Error: 교훈이 너무 짧습니다(20자 미만) — 다음에도 쓸 구체적 내용을 담으십시오."
        body = body[:2000]
        tag = (topic or "").strip()[:40]
        text = (f"[교훈/{tag}] " if tag else "[교훈] ") + body
        uid = (st or {}).get("uid") or "anonymous"
        stores = []
        try:  # ① SQLite 장기기억 (chat_history + vec_memory 임베딩 — recall의 재료)
            import local_memory
            if local_memory.remember(uid, self.MEMORY_CONTEXT, "local_taemini", "assistant", text):
                stores.append("SQLite 벡터기억")
        except Exception as mem_err:
            logger.warning(f"[Coding Memory] SQLite 기록 실패: {mem_err}")
        if os.environ.get("SHADOW_MEMORY_NO_OBSIDIAN") == "1":
            return (f"OK: 🧠 장기기억에 새김({' + '.join(stores)}) — 옵시디언 기록은 "
                    "환경변수로 비활성(검증 모드).") if stores else "Error: 저장 실패."
        try:  # ② 옵시디언 볼트 지식 노트 (사람이 읽는 축적 장부) — 동일 교훈 중복 새김 방지
            from obsidian_jobs import _vault
            v = _vault("local")
            if v is not None:
                try:
                    _, _existing = v.read_note(self.MEMORY_NOTE)
                except Exception:
                    _existing = ""
                if text in _existing:
                    stores.append("옵시디언 볼트(이미 있음)")
                else:
                    import datetime as _dt
                    entry = f"- {_dt.date.today().isoformat()} {text}\n"
                    v.write_note(self.MEMORY_NOTE, entry, mode="append")
                    stores.append("옵시디언 볼트")
        except Exception as obs_err:
            logger.warning(f"[Coding Memory] 옵시디언 기록 실패: {obs_err}")
        if not stores:
            return "Error: 장기기억 저장에 실패했습니다 (SQLite/옵시디언 모두 불가)."
        return (f"OK: 🧠 장기기억에 새김({' + '.join(stores)}) — 다음 세션부터 관련 작업에서 "
                "자동 회상됩니다.")

    # ── 도구: 🧬 자가 진화 직업 (에이전트 전용 직업의 지침/장착 자가 개정) ──
    SELF_JOB_ID = "taemini_self"
    SELF_JOB_LABEL = "자가진화 태미니"
    SELF_JOB_LEARNED_HEADER = "## 스스로 배운 지침"
    SELF_JOB_RULE_CAP = 4000  # 지침 총량 상한 — 약한 모델 컨텍스트 보호

    def code_evolve_job(self, rule: str = "", add_skills: str = "") -> str:
        """Evolves YOUR OWN job 'taemini_self': APPENDS a learned directive (base rules are preserved, never replaced) and/or equips skills, then activates the job. Call with NO args first to see the current state. Other jobs are untouchable (owner-only).
        Args:
            rule: ONE new lesson-directive to append to your learned rules (min 20 chars). Distill what you learned about how to work well.
            add_skills: Comma-separated vault skill names to permanently equip on this job.
        """
        st = _turn()
        try:
            from obsidian_jobs import get_vault_job, save_vault_job
            from local_guardian_store import get_local_guardian_job, set_local_guardian_job
        except Exception as e:
            return f"Error: 직업 시스템 모듈 로드 실패: {e}"
        uid = (st or {}).get("uid") or "anonymous"
        try:
            cur = get_vault_job(self.SELF_JOB_ID, "local") or {}
        except Exception as e:
            return f"Error: 직업 볼트 접근 실패(옵시디언 Vault 비활성?): {e}"
        cur_rule = str(cur.get("rule") or "").strip()
        cur_skills = list(cur.get("equipped_skills") or [])
        rule = (rule or "").strip()
        adds = [s.strip() for s in (add_skills or "").split(",") if s.strip()]
        if not rule and not adds:  # 조회 모드 — 개정 전 현재 상태 확인(read-before-write)
            active = get_local_guardian_job(uid, "local_taemini") == self.SELF_JOB_ID
            return (f"[{self.SELF_JOB_LABEL} 현재 상태] 활성: {'예' if active else '아니오'}\n"
                    f"장착 스킬: {cur_skills or '(없음)'}\n지침:\n{cur_rule or '(아직 없음)'}")
        if st is not None and st.get("plan"):
            return _PLAN_BLOCK_MSG.format(what="직업 개정(code_evolve_job)")
        if rule and len(rule) < 20:
            return "Error: 지침이 너무 짧습니다(20자 미만) — 일하며 배운 원칙을 구체적으로 적으십시오."
        # 🧬 지침은 '누적' — 기반 수칙(자비스가 심은 사고방식)을 약한 모델의 자가 개정이
        # 통째로 대체해 지워버리는 사고 방지. 새 지침은 '스스로 배운 지침' 섹션에 append.
        new_rule = cur_rule
        if rule:
            if rule in cur_rule:
                return f"OK: 이미 지침에 있는 내용입니다 — 중복 추가 생략.\n현재 지침:\n{cur_rule[-600:]}"
            entry = "- " + rule
            if self.SELF_JOB_LEARNED_HEADER in cur_rule:
                new_rule = cur_rule.rstrip() + "\n" + entry
            else:
                new_rule = ((cur_rule.rstrip() + "\n\n") if cur_rule else "") \
                    + self.SELF_JOB_LEARNED_HEADER + " (code_evolve_job이 누적)\n" + entry
            if len(new_rule) > self.SELF_JOB_RULE_CAP:
                return ("Error: 지침이 가득 찼습니다(4000자) — 마왕님께 직업 관리자에서 "
                        "정리를 요청하거나, 세부 내용은 code_remember/스킬로 옮기십시오.")
        missing = [s for s in adds if not os.path.isfile(os.path.join(_skills_dir(), s, "SKILL.md"))]
        if missing:
            return (f"Error: 볼트에 없는 스킬: {missing} — code_forge_skill/tool 로 먼저 만드십시오. "
                    f"사용 가능: {', '.join(list_vault_skills(20))}")
        for s in adds:
            if s in cur_skills:
                cur_skills.remove(s)
            cur_skills.append(s)
        cur_skills = cur_skills[-12:]
        job = {
            "job_id": self.SELF_JOB_ID, "label": self.SELF_JOB_LABEL,
            "description": "태미니가 스스로 개정하는 전용 직업 (지침·장착 스킬 자가 진화)",
            "tool_mode": "all", "equipped_skills": cur_skills,
            "rule": new_rule,
        }
        try:
            save_vault_job(job, "local")
            set_local_guardian_job(uid, "local_taemini", self.SELF_JOB_ID)
        except Exception as e:
            return f"Error: 직업 저장/장착 실패: {e}"
        return (f"OK: 🧬 '{self.SELF_JOB_LABEL}' 개정 + 활성화 — 지침 {len(new_rule)}자(누적), "
                f"장착 스킬 {cur_skills or '(없음)'}. 다음 턴부터 지침이 적용됩니다.")

    # ── 도구: 🧑‍🔬 전문가 소환 (막힘 돌파 — 스스로 전문가를 정의해 2차 의견) ──
    def code_consult_expert(self, specialty: str, question: str, context: str = "") -> str:
        """Summons an on-demand EXPERT persona (a specialist YOU define) and asks it your question — a second brain configuration for when you're stuck. Pass the specialty, a self-contained question, and the relevant context you gathered. Verify its claims before adopting them.
        Args:
            specialty: Expert field, e.g. '파이썬 성능 최적화', '네트워크 보안'
            question: Self-contained question (min 10 chars)
            context: Relevant findings/code excerpts you already gathered
        """
        spec = " ".join((specialty or "").split())[:60]
        q = (question or "").strip()
        if not spec or len(q) < 10:
            return "Error: specialty와 question(10자 이상)이 필요합니다."
        sys_i = (f"너는 '{spec}' 분야 최고 전문가로 소환되었다. 아래 질문에 그 전문가로서 답하라.\n"
                 "- 결론부터, 근거와 함께 5문장 이내로. 확실하지 않은 부분은 '불확실'이라고 표기하라.\n"
                 "- 모르면 모른다고 답하라. 지어내는 것은 금지다. 도구를 호출하지 말고 지식으로 답하라.")
        body = {
            "message": ((f"[의뢰 컨텍스트]\n{(context or '').strip()[:4000]}\n\n" if (context or "").strip() else "")
                        + f"[전문가 자문 질문]\n{q[:2000]}"),
            "sessionKey": f"expert-{hashlib.sha1((spec + str(time.time())).encode()).hexdigest()[:8]}",
            "skipGrounding": True,
            "extraSystem": sys_i,
        }
        try:  # 위임(delegate)과 동일한 자기 HTTP 루프백 — 중첩 think() 상태 오염 방지
            import requests as _rq
            port = (os.environ.get("SHADOW_PORT") or "18700").strip()
            r = _rq.post(f"http://127.0.0.1:{port}/api/shadow/shadow/send", json=body,
                         headers=_loopback_headers(), timeout=300)
            data = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
        except Exception as e:
            return f"Error: 전문가 소환 실패: {e}"
        if not data or data.get("status") == "ERROR":
            return f"Error: 전문가 응답 실패: {(data or {}).get('message') or f'HTTP {r.status_code}'}"
        ans = (data.get("content") or "").strip()[:4000]
        if not ans:
            return "Error: 전문가 응답이 비었습니다 — 질문을 더 구체적으로 바꿔 재시도하십시오."
        return (f"[🧑‍🔬 '{spec}' 전문가 답변]\n{ans}\n"
                "(전문가 의견도 검증 대상이다 — 사실 주장은 도구로 확인한 뒤 채택하라.)")

    # ── OpenAI 호환 디스패처 ──────────────────────────────────
    TOOL_NAMES = (
        "code_list_dir", "code_glob", "code_grep",
        "code_read_file", "code_write_file", "code_edit_file", "code_run_command",
        "code_forge_skill", "code_equip_skill", "code_forge_tool", "code_evolve_job",
        "code_remember", "code_consult_expert",
    )

    def handles(self, name: str) -> bool:
        return name in self.TOOL_NAMES

    def execute(self, name: str, args: dict) -> str:
        if not self.handles(name):
            return f"Error: unknown coding tool '{name}'"
        try:
            return getattr(self, name)(**(args or {}))
        except TypeError as e:
            return f"Error: 잘못된 인자: {e}"

    def callables(self):
        """Gemini AFC 경로용 — bound method 리스트."""
        return [getattr(self, n) for n in self.TOOL_NAMES]


def _schema(name, desc, props, required):
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": desc,
            "parameters": {"type": "object", "properties": props, "required": required},
        },
    }


OPENAI_SCHEMAS = [
    _schema(
        "code_list_dir",
        "Lists files and subdirectories at a path inside the project. Use first to learn the project layout.",
        {"path": {"type": "string", "description": "Directory path relative to project root (default '.')"}},
        [],
    ),
    _schema(
        "code_glob",
        "Finds files by glob pattern, e.g. '**/*.py' or 'portal-heimdall/lib/**/*.dart'. Returns relative paths.",
        {
            "pattern": {"type": "string", "description": "Glob pattern ('**' = any directories)"},
            "max_results": {"type": "integer", "description": "Max paths (default 100)"},
        },
        ["pattern"],
    ),
    _schema(
        "code_grep",
        "Searches file CONTENTS with a regex across the project. Returns 'path:line: text'. Use to find where code is defined or used.",
        {
            "pattern": {"type": "string", "description": "Regular expression"},
            "file_glob": {"type": "string", "description": "Filename filter, e.g. '*.py'"},
            "max_results": {"type": "integer", "description": "Max matching lines (default 50)"},
        },
        ["pattern"],
    ),
    _schema(
        "code_read_file",
        "Reads a text file with line numbers. ALWAYS read before editing.",
        {
            "path": {"type": "string", "description": "File path (relative or absolute)"},
            "offset": {"type": "integer", "description": "1-based start line (default 1)"},
            "limit": {"type": "integer", "description": "Max lines (default 400)"},
        },
        ["path"],
    ),
    _schema(
        "code_write_file",
        "Creates or fully overwrites a file (UTF-8). Prefer code_edit_file for small changes.",
        {
            "path": {"type": "string", "description": "File path"},
            "content": {"type": "string", "description": "Full file content"},
        },
        ["path", "content"],
    ),
    _schema(
        "code_edit_file",
        "Edits a file by EXACT string replacement. old_string must match exactly (indentation included) and be unique unless replace_all.",
        {
            "path": {"type": "string", "description": "File path"},
            "old_string": {"type": "string", "description": "Exact text to find"},
            "new_string": {"type": "string", "description": "Replacement text"},
            "replace_all": {"type": "boolean", "description": "Replace all occurrences (default false)"},
        },
        ["path", "old_string", "new_string"],
    ),
    _schema(
        "code_run_command",
        "Runs a shell command (PowerShell) in the project and returns output + exit code. Use to VERIFY changes: tests, py_compile, flutter analyze, git status.",
        {
            "command": {"type": "string", "description": "Command line to run"},
            "cwd": {"type": "string", "description": "Working dir relative to project root"},
            "timeout": {"type": "integer", "description": "Seconds (default 120, max 600)"},
        },
        ["command"],
    ),
    _schema(
        "code_forge_skill",
        "Creates/updates a reusable SKILL document in the shared vault. Use when the task needs know-how no equipped skill covers — distill it (steps, commands, pitfalls), then equip with code_equip_skill.",
        {
            "name": {"type": "string", "description": "Skill name (letters/digits/space/._- up to 60 chars)"},
            "content": {"type": "string", "description": "Skill body markdown (min 40 chars)"},
        },
        ["name", "content"],
    ),
    _schema(
        "code_equip_skill",
        "Equips a vault skill onto this session — injected into your context every following turn, and returned now for immediate use. Wrong name → lists available skills.",
        {"name": {"type": "string", "description": "Skill name to equip"}},
        ["name"],
    ),
    _schema(
        "code_forge_tool",
        "Creates a self-made TOOL: a Python script (syntax-checked) saved in the vault + a usage SKILL auto-equipped. Re-forge the same name to improve it. Run it via code_run_command.",
        {
            "name": {"type": "string", "description": "Tool name"},
            "script": {"type": "string", "description": "Python source code"},
            "usage": {"type": "string", "description": "How/when to use — args, examples, caveats"},
        },
        ["name", "script", "usage"],
    ),
    _schema(
        "code_evolve_job",
        "Evolves YOUR OWN job 'taemini_self' (self-authored directive + equipped skills) and activates it. Call with no args first to see current state. Other jobs are untouchable.",
        {
            "rule": {"type": "string", "description": "New standing directive (replaces old; min 20 chars)"},
            "add_skills": {"type": "string", "description": "Comma-separated vault skill names to equip on the job"},
        },
        [],
    ),
    _schema(
        "code_consult_expert",
        "Summons an on-demand EXPERT persona you define and asks it a self-contained question — a second opinion when you're stuck. Verify its claims with tools before adopting.",
        {
            "specialty": {"type": "string", "description": "Expert field, e.g. '파이썬 성능 최적화'"},
            "question": {"type": "string", "description": "Self-contained question (min 10 chars)"},
            "context": {"type": "string", "description": "Relevant findings you already gathered"},
        },
        ["specialty", "question"],
    ),
    _schema(
        "code_remember",
        "Engraves a reusable lesson/fact into LONG-TERM memory (SQLite vector recall + Obsidian vault note) so future sessions remember it. Cross-project lessons only; workspace facts go to TAEMINI.md.",
        {
            "lesson": {"type": "string", "description": "The lesson/fact worth remembering (min 20 chars)"},
            "topic": {"type": "string", "description": "Short topic tag"},
        },
        ["lesson"],
    ),
]


# ─────────────────────────────────────────────────────────────────
# 🗺️ [프로젝트 컨텍스트 자동 수집] — Claude Code의 env+CLAUDE.md 주입 등가.
# 코딩 턴 프롬프트 앞에 자동 삽입되어 모델이 도구로 더듬기 전에 방향을 잡는다.
# 전부 로컬 파일/git 로컬 명령만 사용(네트워크 0) — TTFT 부담 최소.
# ─────────────────────────────────────────────────────────────────
_PROJECT_TYPE_MARKERS = (
    "package.json", "pubspec.yaml", "requirements.txt", "pyproject.toml",
    "Cargo.toml", "go.mod", "Gemfile", "CMakeLists.txt", "build.gradle",
    "settings.gradle", "Makefile", "pom.xml", "composer.json", "mix.exs",
)
_AGENT_NOTE_FILES = (  # (파일명, 최대 문자수) — 순서 = 우선순위
    ("TAEMINI.md", 3500),
    ("TAEMINI_NOTES.md", 3500),
    ("CLAUDE.md", 2500),
    ("AGENTS.md", 2500),
)

# 🧬 [워크스페이스 지침서] TAEMINI.md — 이 폴더에서 태미니가 일하는 방식의 단일 진실.
# 없으면 코딩 턴 시작 시 시드를 자동 생성한다(마왕님 요구: "없으면 만들고 스스로 개선").
# 매 턴 [프로젝트 컨텍스트]로 주입되고, 모델이 code_edit_file로 개선을 누적한다.
WORKSPACE_CHARTER = "TAEMINI.md"
_CHARTER_SEED = """# 🧬 TAEMINI.md — 이 워크스페이스의 태미니 지침서 (자가 개선)

이 파일은 태미니(로컬 코딩 에이전트)가 이 프로젝트에서 일하는 방식의 단일 진실이다.
태미니는 매 작업에서 이 지침을 따르고, 작업이 끝나면 배운 것을 여기 반영해 개선한다.

## 지침 (스스로 개선할 것)
- 검증 없이 완료 보고 금지. 같은 실패 2회면 접근을 바꾼다.

## 프로젝트 사실 (구조·빌드·검증 명령 — 확인되면 기록)
- (아직 없음)

## 함정 (겪은 실수 — 반복 금지)
- (아직 없음)

## 자작 도구/스킬 (code_forge_tool/skill 로 만든 것 — 용도 메모)
- (아직 없음)
"""


# 지침서류는 매 턴 [프로젝트 컨텍스트]로 전체 주입되므로 read-before-edit 면제
# (재읽기 강요는 순수 마찰 — 모델이 개선을 포기하고 거짓 보고하는 실관측의 원인)
_READ_EXEMPT_FILES = (WORKSPACE_CHARTER, "TAEMINI_NOTES.md")


def ensure_workspace_charter(cwd: str) -> bool:
    """TAEMINI.md가 없으면 시드 생성(생성 시 True). 구관례 TAEMINI_NOTES.md가
    이미 있으면 그것을 지침서로 인정하고 새로 만들지 않는다(이중 장부 방지)."""
    try:
        cwd = os.path.abspath(cwd or ".")
        path = os.path.join(cwd, WORKSPACE_CHARTER)
        if os.path.isfile(path) or os.path.isfile(os.path.join(cwd, "TAEMINI_NOTES.md")):
            return False
        with open(path, "w", encoding="utf-8") as f:
            f.write(_CHARTER_SEED)
        with _PROJECT_CTX_LOCK:  # 이번 턴 주입에 바로 반영되게 캐시 무효화
            _PROJECT_CTX_CACHE.pop(cwd, None)
        return True
    except Exception:
        return False
_PROJECT_CTX_CACHE = {}  # cwd → (ts, text)
_PROJECT_CTX_LOCK = threading.Lock()
_PROJECT_CTX_TTL = 15  # 초 — 연속 턴에서 재수집 생략


def _git_lines(cwd, args, timeout=2.0):
    try:
        p = subprocess.run(["git", "-C", cwd] + args, capture_output=True,
                           timeout=timeout, text=True, errors="replace")
        if p.returncode != 0:
            return []
        return [l for l in p.stdout.splitlines() if l.strip()]
    except Exception:
        return []


def build_project_context(cwd: str, max_chars: int = 7000) -> str:
    """작업 루트의 오리엔테이션 블록 — 최상위 구성·프로젝트 타입·git 상태·에이전트 노트."""
    cwd = os.path.abspath(cwd or ".")
    now = time.time()
    with _PROJECT_CTX_LOCK:
        cached = _PROJECT_CTX_CACHE.get(cwd)
        if cached and now - cached[0] < _PROJECT_CTX_TTL:
            return cached[1]
    parts = []
    try:
        entries = sorted(os.listdir(cwd))
        dirs = [e + "/" for e in entries
                if os.path.isdir(os.path.join(cwd, e)) and e not in _SKIP_DIRS and not e.startswith(".")]
        files = [e for e in entries
                 if os.path.isfile(os.path.join(cwd, e)) and not e.startswith(".")]
        top = (dirs + files)[:40]
        if top:
            parts.append("최상위 구성: " + "  ".join(top))
        markers = [m for m in _PROJECT_TYPE_MARKERS if m in files]
        if markers:
            parts.append("프로젝트 마커: " + ", ".join(markers))
    except Exception:
        pass
    branch = _git_lines(cwd, ["rev-parse", "--abbrev-ref", "HEAD"])
    if branch:
        git_bits = [f"git 브랜치: {branch[0]}"]
        status = _git_lines(cwd, ["status", "--porcelain"])
        if status:
            shown = status[:10]
            more = f" (+{len(status) - 10}건 더)" if len(status) > 10 else ""
            git_bits.append("변경 파일:\n  " + "\n  ".join(shown) + more)
        else:
            git_bits.append("작업 트리 깨끗함")
        log = _git_lines(cwd, ["log", "-5", "--format=%h %s (%cr)"])
        if log:
            git_bits.append("최근 커밋:\n  " + "\n  ".join(log))
        parts.append("\n".join(git_bits))
    for fname, cap in _AGENT_NOTE_FILES:
        fp = os.path.join(cwd, fname)
        try:
            if os.path.isfile(fp):
                with open(fp, "r", encoding="utf-8", errors="replace") as f:
                    body = f.read(cap + 1)
                if len(body) > cap:
                    body = body[:cap] + "\n…(이하 생략 — 필요 시 code_read_file로 전체 열람)"
                parts.append(f"### {fname} (프로젝트 노트/지침 — 자동 포함)\n{body.strip()}")
        except Exception:
            continue
    text = "\n\n".join(p for p in parts if p).strip()[:max_chars]
    with _PROJECT_CTX_LOCK:
        _PROJECT_CTX_CACHE[cwd] = (now, text)
        if len(_PROJECT_CTX_CACHE) > 32:  # 무한 성장 방지
            oldest = min(_PROJECT_CTX_CACHE, key=lambda k: _PROJECT_CTX_CACHE[k][0])
            _PROJECT_CTX_CACHE.pop(oldest, None)
    return text


# 시스템 지시에 덧붙일 에이전틱 코딩 워크플로 가이드
CODING_AGENT_GUIDE = """

## 🛠️ 에이전틱 코딩 수칙 (Imperial Coding Protocol)
너는 코딩 도구(code_*)를 가진 자율 코딩 에이전트다. 코드 관련 요청을 받으면 추측으로 답하지 말고 반드시 도구로 확인하라.

작업 순서 (반드시 준수):
1. **탐색**: code_grep / code_glob / code_list_dir 로 관련 파일과 코드를 먼저 찾는다. 파일 위치를 추측하지 마라.
2. **읽기**: code_read_file 로 수정할 부분의 실제 코드를 읽는다. 읽지 않은 파일을 수정하는 것은 금지다.
3. **수정**: code_edit_file 로 정확한 원문(들여쓰기 포함)을 복사해 교체한다. 새 파일만 code_write_file 을 쓴다.
4. **검증**: code_run_command 로 검증한다 (python -m py_compile, flutter analyze, 테스트 실행 등). 검증 없이 "완료"라고 보고하지 마라.
5. **반복**: 검증이 실패하면 오류 메시지를 읽고 1번부터 다시 수행한다. 같은 실패를 반복하면 접근을 바꿔라.

규칙:
- 도구 결과(파일 내용, 검색 결과)에 근거해서만 말하라. 존재를 확인하지 않은 파일/함수를 언급하지 마라.
- 수정은 최소 범위로 하고, 주변 코드의 스타일을 따른다.
- 실패하면 솔직하게 실패했다고 보고하고 원인과 다음 시도를 설명한다.
"""