context.py

"""System context: DULUS.md, git info, cwd injection.

NOTE on prompt caching: this module is the source of the system prompt sent
to every provider call. To get prefix caching (Anthropic explicit + OpenAI-
compat automatic), the rendered prompt MUST be byte-stable across turns of
the same session. Anything that changes per turn (date with sub-day grain,
`git status` modified-file counts, `datetime.now()`, etc.) belongs OUTSIDE
this prompt. Disk reads (DULUS.md, MEMORY.md) are cached by mtime so a
turn that doesn't touch those files re-uses the prior bytes verbatim.
"""
import os
import subprocess
from pathlib import Path

SYSTEM_PROMPT_TEMPLATE = """\
You are Dulus, an AI coding agent. Think in English; reply to {user_name} in {reply_language}.
# Identity: Your name is Dulus. Do NOT proactively declare this — only if the user asks "quién eres" or "qué modelo eres".
# Forbidden: Do NOT claim to be Qwen, Llama, GPT, Claude, Gemini, DeepSeek, or any underlying model. Do NOT mention Ollama or your runtime stack.
# Env: {cwd} | {platform} | auto_show={auto_show}
{isolate_rule}
# Autonomy: Background scripts (nohup/&) allowed | Never refuse monitoring/long tasks | Always wait for tool results before replying
# Tools: SearchLastOutput → for [TRUNCATED] | WebFetch/WebSearch → web | TmuxOffload → tasks > 5s | ReadJob → background results
# EFFICIENCY LAW (hard rule, zero exceptions): Tool calls cost round-trips + tokens — MINIMIZE them ruthlessly. NEVER drip-feed Grep/Glob/Read calls to hunt something: ONE Python() call that os.walk()s the tree and filters in kernel memory replaces 10+ search calls. If you're about to fire your 3rd search/read call for the SAME investigation, STOP — switch to Python and finish the whole hunt there (walk + regex + parse, print only the final slice). INDEPENDENT calls (reads, searches, fetches) → emit ALL of them in ONE response, never one per turn. DEPENDENT steps (scan → filter → drill down) → chain them INSIDE a single Python()/Bash() call, not across turns. The Python console is your working memory: load once, query forever.
# Reminder: ONLY for user-facing reminders/notifications (e.g. "remind me in 10min"). NEVER use it to wait between your own tool calls — the countdown is deferred until your turn ends but you should still pause inside a command sequence using `sleep N` INSIDE the Bash command itself (e.g. Bash('cmd1 && sleep 2 && cmd2')).
# Long-running tools: any tool whose `description` ends in `[long-running — wrap in TmuxOffload]` MUST be invoked via TmuxOffload (not directly), so the REPL stays responsive while it runs.
# Multi-agent: Agent(subagent_type=...) | isolation="worktree" runs parallel | wait=false + name=... for fire-and-forget
# Rules: Edit > Write | Use absolute paths + line numbers | Surface errors immediately, do not retry blindly
# Input: "🎙 Transcribed:" prefix = voice input — tolerate typos/misspellings
# REPL: /help /batch /auto_show /isolate /verbose /soul /memory /schema /thinking /config
{platform_hints}{git_info}{dulus_md}"""

_THINKING_LABELS = {1: "minimal", 2: "moderate", 3: "deep"}


def get_git_info(config: dict | None = None) -> str:
    """Return ONLY the branch name — stable across turns within a session.

    Previous versions also embedded `git status --short` modified-file count
    and the last commit hash; both change as the user works, which trashed
    prefix caching on every turn. The agent can call `git status` itself
    when it actually needs current state.
    """
    if config and not config.get("git_status", True):
        return ""
    try:
        branch = subprocess.check_output(
            ["git", "rev-parse", "--abbrev-ref", "HEAD"],
            stderr=subprocess.DEVNULL, text=True,
        ).strip()
        return f"Git:{branch}\n" if branch else ""
    except Exception:
        return ""


# ── mtime-based caches for DULUS.md / MEMORY.md ──────────────────────────
# Re-reading these files on every turn is wasteful disk I/O. More importantly,
# the *content* is the same most of the time — caching it keeps the rendered
# system prompt byte-stable, which is what providers need to grant prefix
# cache hits. Invalidation key = (path, mtime_ns) tuple of the resolved files.

_DULUS_MD_CACHE: dict = {"key": None, "value": ""}
_MEMORY_MD_CACHE: dict = {"key": None, "value": ""}


def _resolve_dulus_md_paths() -> list[Path]:
    paths = []
    global_md = Path.home() / ".dulus" / "DULUS.md"
    if global_md.exists():
        paths.append(global_md)
    for p in [Path.cwd()] + list(Path.cwd().parents):
        candidate = p / "DULUS.md"
        if candidate.exists():
            paths.append(candidate)
            break
    return paths


def get_dulus_md() -> str:
    paths = _resolve_dulus_md_paths()
    try:
        key = tuple((str(p), p.stat().st_mtime_ns) for p in paths)
    except OSError:
        key = None
    if key is not None and _DULUS_MD_CACHE["key"] == key:
        return _DULUS_MD_CACHE["value"]

    content_parts = []
    for p in paths:
        try:
            label = "Global DULUS.md" if p == Path.home() / ".dulus" / "DULUS.md" else f"Project DULUS.md:{p.parent}"
            content_parts.append(f"[{label}]\n{p.read_text(encoding='utf-8', errors='replace')}")
        except Exception:
            continue

    value = "\nDULUS.md:\n" + "\n---\n".join(content_parts) + "\n" if content_parts else ""
    _DULUS_MD_CACHE["key"] = key
    _DULUS_MD_CACHE["value"] = value
    return value


def _resolve_memory_index_path() -> Path | None:
    for p in [Path.cwd()] + list(Path.cwd().parents):
        index = p / ".dulus-context" / "memory" / "MEMORY.md"
        if index.exists():
            return index
    return None


def get_project_memory_index() -> str:
    """Auto-load project-scope memories from .dulus-context/memory/MEMORY.md.

    Looks in cwd and parents (first match wins). Returns the index so the model
    knows what memories exist and can Read individual files on demand. Cached
    by mtime so unchanged indexes don't bust the prompt cache.
    """
    path = _resolve_memory_index_path()
    if path is None:
        if _MEMORY_MD_CACHE["key"] != "MISSING":
            _MEMORY_MD_CACHE["key"] = "MISSING"
            _MEMORY_MD_CACHE["value"] = ""
        return ""
    try:
        key = (str(path), path.stat().st_mtime_ns)
    except OSError:
        return ""
    if _MEMORY_MD_CACHE["key"] == key:
        return _MEMORY_MD_CACHE["value"]
    try:
        body = path.read_text(encoding="utf-8", errors="replace").strip()
    except Exception:
        body = ""
    if not body:
        value = ""
    else:
        value = (
            f"\n# Project memories ({path.parent})\n"
            f"# Index below — Read the .md files in that dir for full content.\n"
            f"{body}\n"
        )
    _MEMORY_MD_CACHE["key"] = key
    _MEMORY_MD_CACHE["value"] = value
    return value


def _find_git_bash_path() -> str | None:
    import shutil
    # 1. PATH (skip WSL stubs)
    bash_in_path = shutil.which("bash")
    if bash_in_path:
        low = bash_in_path.lower()
        if "system32" not in low and "sysnative" not in low and "syswow64" not in low:
            if Path(bash_in_path).exists():
                return bash_in_path

    # 2. Check standard system and per-user install directories
    candidates = [
        r"C:\Program Files\Git\bin\bash.exe",
        r"C:\Program Files\Git\usr\bin\bash.exe",
        r"C:\Program Files (x86)\Git\bin\bash.exe",
        r"C:\Program Files (x86)\Git\usr\bin\bash.exe",
        r"C:\msys64\usr\bin\bash.exe",
    ]
    local_app_data = os.environ.get("LOCALAPPDATA", "")
    if local_app_data:
        candidates.extend([
            os.path.join(local_app_data, "Programs", "Git", "bin", "bash.exe"),
            os.path.join(local_app_data, "Programs", "Git", "usr", "bin", "bash.exe"),
        ])
    user_profile = os.environ.get("USERPROFILE", "")
    if user_profile:
        candidates.extend([
            os.path.join(user_profile, "scoop", "apps", "git", "current", "bin", "bash.exe"),
            os.path.join(user_profile, "scoop", "apps", "git", "current", "usr", "bin", "bash.exe"),
        ])
    for c in candidates:
        if c and Path(c).exists():
            return c
    return None


def _find_powershell_path() -> str | None:
    import shutil
    candidates = [
        shutil.which("pwsh"),
        shutil.which("powershell"),
        r"C:\Program Files\PowerShell\7\pwsh.exe",
        r"C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe",
    ]
    for c in candidates:
        if c and Path(c).exists():
            return c
    return None


def _find_cmd_path() -> str:
    import shutil
    return shutil.which("cmd") or r"C:\Windows\System32\cmd.exe"


def _detect_running_parent_shell() -> str | None:
    """Inspect parent process tree if psutil is available to see what launched Dulus."""
    try:
        import psutil
        proc = psutil.Process()
        for _ in range(4):
            proc = proc.parent()
            if not proc:
                break
            name = proc.name().lower()
            if "pwsh" in name or "powershell" in name:
                return "powershell"
            if "bash" in name or "mintty" in name or "sh.exe" in name:
                return "gitbash"
            if "cmd.exe" in name:
                return "cmd"
    except Exception:
        pass
    return None


def resolve_shell_environment(config: dict | None = None) -> dict[str, str]:
    """Single source of truth for shell detection and execution across Dulus.

    Returns a dict with:
      - kind: 'gitbash' | 'powershell' | 'cmd' | 'wsl' | 'bash' | 'custom'
      - path: path to the executable
      - family: 'bash' | 'powershell' | 'cmd'
    """
    import sys as _sys
    if _sys.platform != "win32":
        shell_path = os.environ.get("SHELL", "/bin/bash")
        return {"kind": "bash", "path": shell_path, "family": "bash"}

    configured = config.get("shell", {}).get("type", "auto") if config else "auto"
    forced_path = config.get("shell", {}).get("path", "") if config else ""
    st = (configured or "auto").lower()

    # 1. Custom shell with explicit path
    if st == "custom" and forced_path and Path(forced_path).exists():
        low = forced_path.lower()
        if "bash" in low or "sh" in low:
            fam = "bash"
        elif "pwsh" in low or "powershell" in low:
            fam = "powershell"
        else:
            fam = "cmd"
        return {"kind": "custom", "path": forced_path, "family": fam}

    # 2. Explicitly configured shells
    if st == "gitbash":
        p = _find_git_bash_path()
        if p:
            return {"kind": "gitbash", "path": p, "family": "bash"}
    elif st == "wsl":
        import shutil
        wsl = shutil.which("wsl")
        if wsl:
            return {"kind": "wsl", "path": wsl, "family": "bash"}
    elif st == "powershell":
        p = _find_powershell_path()
        if p:
            return {"kind": "powershell", "path": p, "family": "powershell"}
    elif st == "cmd":
        return {"kind": "cmd", "path": _find_cmd_path(), "family": "cmd"}

    # 3. Auto-detection:
    # A) Check parent process tree (e.g. launched from Windows Terminal / VS Code)
    parent_shell = _detect_running_parent_shell()
    if parent_shell == "gitbash":
        p = _find_git_bash_path()
        if p:
            return {"kind": "gitbash", "path": p, "family": "bash"}
    elif parent_shell == "powershell":
        p = _find_powershell_path()
        if p:
            return {"kind": "powershell", "path": p, "family": "powershell"}
    elif parent_shell == "cmd":
        p = _find_git_bash_path()
        if p:
            return {"kind": "gitbash", "path": p, "family": "bash"}
        return {"kind": "cmd", "path": _find_cmd_path(), "family": "cmd"}

    # B) Check interactive shell environment variables
    if os.environ.get("MSYSTEM") or "MINGW" in os.environ.get("MSYSTEM", ""):
        p = _find_git_bash_path()
        if p:
            return {"kind": "gitbash", "path": p, "family": "bash"}

    shell_env = os.environ.get("SHELL", "").lower()
    if "bash" in shell_env or "BASH" in os.environ or "BASH_VERSION" in os.environ:
        p = _find_git_bash_path()
        if p:
            return {"kind": "gitbash", "path": p, "family": "bash"}

    if os.environ.get("WSL_DISTRO_NAME") or os.environ.get("WSL_INTEROP"):
        import shutil
        wsl = shutil.which("wsl")
        if wsl:
            return {"kind": "wsl", "path": wsl, "family": "bash"}

    if os.environ.get("PSExecutionPolicyPreference") or os.environ.get("PSCommandPath"):
        p = _find_powershell_path()
        if p:
            return {"kind": "powershell", "path": p, "family": "powershell"}

    # C) Neutral preference (GUI app, background service, etc.):
    # Prefer Git Bash for POSIX tool call compatibility if installed
    git_bash = _find_git_bash_path()
    if git_bash:
        return {"kind": "gitbash", "path": git_bash, "family": "bash"}

    # Next prefer PowerShell
    ps = _find_powershell_path()
    if ps:
        return {"kind": "powershell", "path": ps, "family": "powershell"}

    # Default fallback to CMD
    return {"kind": "cmd", "path": _find_cmd_path(), "family": "cmd"}


def _detect_shell_type(config: dict | None = None) -> str:
    """Resolve which shell family to advertise: 'bash', 'powershell', or 'cmd'."""
    info = resolve_shell_environment(config)
    return info.get("family", "bash")


def get_platform_hints(config: dict | None = None) -> str:
    import platform as _plat
    dulus_home = Path.home() / ".dulus"
    skills_dir = dulus_home / "skills"
    if _plat.system() == "Windows":
        info = resolve_shell_environment(config)
        fam = info.get("family", "bash")
        kind = info.get("kind", "gitbash")
        if fam == "bash":
            return f"# Shell:Windows({kind}/POSIX) | cat,grep,ls,curl,&& | Dulus:{dulus_home} | Skills:{skills_dir} | Forward slashes /c/Users/... or relative paths"
        elif fam == "powershell":
            cmds = "Get-Content=cat,Select-String=grep,Get-ChildItem=ls"
            return f"# Shell:Windows(PowerShell) | {cmds} | Dulus:{dulus_home} | Skills:{skills_dir} | WARNING: Use Glob/Read tools instead of raw shell commands for path navigation to avoid Windows path errors."
        else:
            cmds = "type=cat,findstr=grep,dir=ls"
            return f"# Shell:Windows(CMD) | {cmds} | Dulus:{dulus_home} | Skills:{skills_dir} | WARNING: Use Glob/Read tools instead of raw shell commands for path navigation to avoid Windows path errors."
    return f"# Shell:Unix(cat,grep,ls,&&) | Dulus:{dulus_home} | Skills:{skills_dir}"


def _build_ollama_system_prompt(config: dict | None = None) -> str:
    import platform
    auto_show = config.get("auto_show", True) if config else True
    prompt = f"""你是Dulus，AI编程助手。
# 身份: 你的名字是 Dulus。不需要主动向用户声明这一点，除非用户直接问你 "quién eres" 或 "qué modelo eres"。
# 禁止: 不要声称自己是 Qwen、Llama、GPT、Claude、Gemini、DeepSeek 或任何其他底层模型。不要提及 Ollama 或运行你的技术栈。
Dir:{Path.cwd()} | Platform:{platform.system()} | auto_show={'ON' if auto_show else 'OFF'}
# 规则: 读取工具结果后再响应 | >5s操作用TmuxOffload
# 效率(强制): 独立的工具调用必须在同一回复中并行发出，绝不一轮一个 | 多步骤的文件搜索用Python控制台一次完成(os.walk+正则)，不要连续多次Grep/Read
# 工具格式(单行): <tool_call>{{"name":"TOOL","input":{{"param":"val"}}}}</tool_call>
# 可用: Bash,Glob,Read,Write,SearchLastOutput,TmuxOffload,WebSearch,WebFetch,PrintToConsole
# PrintToConsole: auto_show=OFF时输出给用户，展示后不重复
"""
    dulus_md = get_dulus_md()
    if dulus_md: prompt += f"\n{dulus_md}"
    return prompt


def _normalize_thinking_level(config: dict | None) -> int:
    raw = config.get("thinking", 0) if config else 0
    if raw is True:
        return 3
    if raw in (False, None):
        return 0
    try:
        return max(0, min(4, int(raw)))
    except (TypeError, ValueError):
        return 0


# ── Reply-language resolution ─────────────────────────────────────────────
#
# `config["lang"]` lets the user steer what language Dulus replies in
# without touching the prompt template. Two kinds of values are accepted:
#
#   • ISO-639 (with optional region):   "en", "es", "es-DO", "zh", "zh-Hant",
#                                        "pt-BR", "ja", "fr", "de", "it",
#                                        "ko", "ru", "ar", "tr", "hi", "id"…
#     Mapped to a human-readable instruction by _LANG_NAMES.
#   • Free-form natural string:         "very formal British English",
#                                        "dominicano callejero", "pirate".
#     Passed through verbatim so power users can role-play any voice.
#
# Default = "es-DO" (Dominican Spanish — the founder's tongue) to keep
# the existing identity untouched for existing users.

_LANG_NAMES: dict[str, str] = {
    # Spanish (default + regions)
    "es":     "Dominican Spanish",
    "es-do":  "Dominican Spanish",
    "es-mx":  "Mexican Spanish",
    "es-es":  "Castilian Spanish",
    "es-ar":  "Argentinian Spanish",
    "es-co":  "Colombian Spanish",
    # Global big ones
    "en":     "English",
    "en-us":  "American English",
    "en-gb":  "British English",
    "zh":     "Simplified Chinese (Mandarin)",
    "zh-cn":  "Simplified Chinese (Mandarin)",
    "zh-tw":  "Traditional Chinese",
    "zh-hant":"Traditional Chinese",
    "pt":     "Portuguese (Brazilian)",
    "pt-br":  "Brazilian Portuguese",
    "pt-pt":  "European Portuguese",
    "ja":     "Japanese",
    "ko":     "Korean",
    "fr":     "French",
    "de":     "German",
    "it":     "Italian",
    "ru":     "Russian",
    "ar":     "Arabic",
    "tr":     "Turkish",
    "hi":     "Hindi",
    "id":     "Indonesian",
    "vi":     "Vietnamese",
    "th":     "Thai",
    "nl":     "Dutch",
    "pl":     "Polish",
    "sv":     "Swedish",
    "uk":     "Ukrainian",
    "he":     "Hebrew",
    "fa":     "Persian (Farsi)",
}


def _resolve_reply_language(config: dict | None) -> str:
    raw = (config.get("lang", "") if config else "") or ""
    raw = raw.strip()
    if not raw:
        return "Dominican Spanish"
    # ISO code shortcut.
    code = raw.lower().replace("_", "-")
    if code in _LANG_NAMES:
        return _LANG_NAMES[code]
    # Free-form descriptor — return verbatim so the user can role-play
    # voice ("Shakespeare-era English", "callejero dominicano", "pirate").
    return raw



def _append_baseline_fragments(prompt: str) -> str:
    """Attach soul + gold system fragments (always-on baseline)."""
    try:
        from memory import soul_system_fragment, gold_system_fragment
        _soul = soul_system_fragment()
        if _soul:
            prompt += "\n\n" + _soul
        _gold = gold_system_fragment()
        if _gold:
            prompt += "\n\n" + _gold
    except Exception:
        pass
    return prompt


def build_system_prompt(config: dict | None = None) -> str:
    import platform
    model_lower = (config.get("model", "") if config else "").lower()
    is_deepseek_r1 = "deepseek-r1" in model_lower or "deepseek-reasoner" in model_lower
    if is_deepseek_r1 and config and config.get("deep_override", False):
        return _append_baseline_fragments(_build_ollama_system_prompt(config))

    auto_show = "ON" if (not config or config.get("auto_show", True)) else "OFF"
    lite = bool(config and config.get("lite_mode"))

    # Isolate fragment is empty when OFF so the prompt stays byte-stable for
    # prompt-cache across toggles-off sessions; when ON it injects a hard rule.
    try:
        from dulus_tools.isolate import prompt_fragment as _isolate_prompt_fragment
        isolate_rule = _isolate_prompt_fragment(config)
    except Exception:
        isolate_rule = ""

    # In LITE mode: drop the optional context blocks (platform hints, git info,
    # DULUS.md, project memory index, batch/thinking/plan/tmux hints). The
    # core identity + tool rules stay. This is what the /lite toggle was
    # supposed to do all along — previously the flag flipped a config bit
    # that nothing actually consumed.
    user_name = (config.get("user_name") if config else None) or "KevRojo"
    reply_language = _resolve_reply_language(config)
    prompt = SYSTEM_PROMPT_TEMPLATE.format(
        cwd=str(Path.cwd()),
        platform=platform.system(),
        auto_show=auto_show,
        isolate_rule=isolate_rule,
        user_name=user_name,
        reply_language=reply_language,
        platform_hints="" if lite else get_platform_hints(config),
        git_info="" if lite else get_git_info(config),
        dulus_md="" if lite else get_dulus_md(),
    )

    # Soul + Gold baseline ALWAYS (even in lite) — model source of truth.
    # Display copies in chat are GUI-only and get stripped before the API call.
    prompt = _append_baseline_fragments(prompt)

    if lite:
        # Bail early — minimal prompt only (baseline already attached).
        return prompt

    try:
        from tmux_tools import tmux_available
        if tmux_available():
            prompt += "\n# Tmux: available"
    except Exception:
        pass

    prompt += (
        "\n# Batch: /batch list|status|fetch (suggest when 3+ similar tasks) | "
        # Both `dulus` (when pip-installed) and `python dulus.py` work — the
        # entry-point shim is registered in pyproject.toml [project.scripts].
        'In agents: Bash(\'dulus -c "batch status|fetch ID"\')'
    )

    thk_label = _THINKING_LABELS.get(_normalize_thinking_level(config))
    if thk_label:
        prompt += f"\n# Thinking: {thk_label}"

    if config and config.get("_plan_mode"):
        prompt += f"\n# Plan mode: read-only (except {config.get('_plan_file', 'PLAN.md')})"

    # Hint: pip-installed users can run `dulus` directly (no .py path).
    prompt += (
        "\n# CLI: 'dulus' command works after `pip install dulus` — "
        "no need for `python dulus.py`. Same flags: --print, --accept-all, -c, etc."
    )

    # Skills proactivity hint — make the agent reach for skills instead of
    # writing one-off code when a topic comes up that no current tool/plugin
    # covers. The dump file lets us grep ~1000 skills instantly without
    # paging through them interactively.
    try:
        skills_dump = Path.home() / ".dulus" / "skills_catalog.txt"
        if skills_dump.exists():
            # Do NOT embed file size/mtime — that changes on every catalog
            # refresh and would bust the entire system-prompt prefix cache
            # (xAI/OpenAI/Anthropic all key on exact bytes).
            prompt += (
                f"\n# Skills catalog: {skills_dump} (tab-separated "
                "source\\tid\\tdescription). Before writing custom code or "
                "saying 'I can't do that', Grep this file for the topic — "
                "there's often an awesome/composio/local skill that fits. "
                "Install with `dulus -c \"skill get <id>\"`. Refresh the "
                "dump anytime with `dulus -c \"skill list dump\"`."
            )
        else:
            prompt += (
                "\n# Skills tip: run `dulus -c \"skill list dump\"` once "
                "to write ~/.dulus/skills_catalog.txt — then Grep it for any "
                "topic you don't have a tool for, before writing custom code."
            )
    except Exception:
        pass

    project_mem = get_project_memory_index()
    if project_mem:
        prompt += project_mem

    # ── Reply-language HARD OVERRIDE ──────────────────────────────────────────
    # The soul (soul.md) and gold memories are injected as conversation messages
    # and often assert a fixed voice ("I speak Dominican Spanish"). Those pin the
    # language and quietly beat the single line at the top of the prompt, so
    # /lang appeared to "not work". We re-assert the chosen language HERE, at the
    # very end of the system prompt (highest authority / most recent), but only
    # when the user has explicitly set config["lang"] to something other than the
    # Dominican-Spanish default — otherwise the soul stays in charge.
    if config:
        raw_lang = (config.get("lang", "") or "").strip()
        if raw_lang:
            resolved = _resolve_reply_language(config)
            if resolved and resolved != "Dominican Spanish":
                prompt += (
                    f"\n\n# ⚠ LANGUAGE OVERRIDE (set via /lang — HIGHEST PRIORITY): "
                    f"Reply to {user_name} in {resolved}. This OVERRIDES any language "
                    f"stated in your soul, identity essence, or golden memories. "
                    f"Keep your personality and tone, but the OUTPUT LANGUAGE must be "
                    f"{resolved}, no exceptions, every single turn."
                )

    return prompt