simbiot-harness/harness/cc_backend.py

"""Simbiot Harness — движок LLM-тика «claude_code»: тик исполняет Claude Code (`claude -p`).

Альтернатива Strands + 9router (_llm_worker). Выбор движка и модели — на уровень:
  SIMBIOT_LLM_BACKEND_<LEVEL> > SIMBIOT_LLM_BACKEND > strands   (strands | claude_code)
  SIMBIOT_CC_MODEL_<LEVEL>    > дефолт: min — Haiku 5.5, hour — Sonnet 5.5, day — Opus 5.5
  SIMBIOT_CC_API_KEY          — ключ Anthropic API для тиков (без него — вход Claude Code на сервере)
Запускается внутри того же изолированного воркера (isolated_runner): таймаут уровня
убивает группу процессов вместе с claude.

Что сохраняется от Strands-тика:
  - мандат уровня → --append-system-prompt; промпт тика → stdin;
  - круг доверия: хук cc_guard.py на каждый инструмент (trust.check_tool), MCP портала —
    только наш сервер (--strict-mcp-config) с токеном харнеса и лимитом вызовов за тик;
  - бюджет шагов — --max-turns (resolve_tool_budget уровня);
  - трасса: InvocationStarted здесь, ToolCalled/ToolResult — хук;
  - BudgetSpent инкрементально — по каждому ответу модели из stream-json (kill по
    таймауту теряет только хвост, не весь тик).
Сессии Claude Code не сохраняются (--no-session-persistence): тики не засоряют панель.
"""
import json
import os
import shutil
import subprocess
import tempfile

from .levels import _env_cfg

HARNESS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFAULT_MODELS = {"min": "claude-haiku-5-5", "hour": "claude-sonnet-5-5", "day": "claude-opus-5-5"}
RAG_HINT = ("\n\nДвижок тика — Claude Code. Поиск по базе знаний кооператива: "
            "./venv/bin/python cli.py rag-query \"вопрос целиком\" (1 запрос за тик). "
            "Портал кооператива — MCP-инструменты mcp__portal__* (до 4 вызовов за тик). "
            "Команды харнеса (tg-reply, task, escalate, …) — ./venv/bin/python cli.py … из корня репо.")


def backend_for(level: str) -> str:
    v = (_env_cfg(f"SIMBIOT_LLM_BACKEND_{level.upper()}") or _env_cfg("SIMBIOT_LLM_BACKEND") or "strands").strip().lower()
    return "claude_code" if v in ("claude_code", "claude-code", "claude", "cc") else "strands"


def model_for(level: str) -> str:
    return _env_cfg(f"SIMBIOT_CC_MODEL_{level.upper()}") or DEFAULT_MODELS.get(level, DEFAULT_MODELS["min"])


def claude_bin() -> str:
    return _env_cfg("SIMBIOT_CLAUDE_BIN") or shutil.which("claude") or "/usr/bin/claude"


def _mcp_config(tmpdir: str):
    """Только MCP портала с токеном харнеса; без токена — без MCP."""
    from . import portal_mcp
    tok = portal_mcp._load_token()
    if not tok:
        return None
    p = os.path.join(tmpdir, "mcp.json")
    with open(p, "w") as f:
        json.dump({"mcpServers": {"portal": {"type": "http", "url": portal_mcp._url(),
                                              "headers": {"Authorization": f"Bearer {tok}"}}}}, f)
    os.chmod(p, 0o600)
    return p


def _usage_delta(u: dict) -> dict:
    p = sum(int(u.get(k) or 0) for k in ("input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens"))
    return {"prompt_tokens": p, "completion_tokens": int(u.get("output_tokens") or 0), "model_calls": 1}


def run(payload: dict, emit_budget) -> dict:
    """Тик на Claude Code. Возвращает {"ok", "text", "usage", "trace_id"} как Strands-воркер."""
    from . import es
    from ._llm_worker import resolve_tool_budget
    level = payload.get("level") or "min"
    trace_id = payload.get("trace_prefix") or "tick"
    model = model_for(level)
    es.append_event("trace", f"{trace_id}:cc0", "InvocationStarted",
                    {"trace_id": trace_id, "level": level, "model": model, "engine": "claude_code"})
    tmpdir = tempfile.mkdtemp(prefix="simbiot_cc_")
    try:
        state = os.path.join(tmpdir, "state.json")
        guard = f"{os.path.join(HARNESS_DIR, 'venv', 'bin', 'python')} {os.path.join(HARNESS_DIR, 'harness', 'cc_guard.py')}"
        settings = {"hooks": {ev: [{"matcher": "*", "hooks": [{"type": "command", "command": guard, "timeout": 30}]}]
                              for ev in ("PreToolUse", "PostToolUse")}}
        cmd = [claude_bin(), "-p", "--model", model, "--output-format", "stream-json", "--verbose",
               "--no-session-persistence", "--max-turns", str(resolve_tool_budget(level)),
               "--append-system-prompt", (payload.get("instructions") or "") + RAG_HINT,
               "--settings", json.dumps(settings),
               # права решает страж (cc_guard → trust.check_tool), не интерактивные промпты
               "--dangerously-skip-permissions"]
        if not payload.get("can_shell"):
            cmd += ["--disallowedTools", "Bash", "Write", "Edit", "MultiEdit", "NotebookEdit"]
        mcp = _mcp_config(tmpdir)
        cmd += (["--mcp-config", mcp] if mcp else []) + ["--strict-mcp-config"]
        # IS_SANDBOX=1: харнес работает от root, а без него Claude Code отказывается от
        # --dangerously-skip-permissions (инцидент 18.10 21:25, e-261018212529-a21a91);
        # права тика всё равно решает страж cc_guard, а не интерактивные подтверждения
        env = dict(os.environ, SIMBIOT_CC_STATE=state, SIMBIOT_TRACE_ID=trace_id, SIMBIOT_TICK_LEVEL=level,
                   IS_SANDBOX="1")
        # отдельный ключ Anthropic API для харнеса (иначе — вход Claude Code на сервере)
        api_key = _env_cfg("SIMBIOT_CC_API_KEY")
        if api_key:
            env["ANTHROPIC_API_KEY"] = api_key
        proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                cwd=HARNESS_DIR, env=env, text=True)
        proc.stdin.write(payload.get("prompt") or "")
        proc.stdin.close()
        seen, usage, final = set(), {"prompt_tokens": 0, "completion_tokens": 0, "model_calls": 0}, None
        for line in proc.stdout:
            try:
                ev = json.loads(line)
            except ValueError:
                continue
            if ev.get("type") == "assistant":
                msg = ev.get("message") or {}
                mid = msg.get("id")
                if mid and mid not in seen and msg.get("usage"):
                    seen.add(mid)  # один ответ модели = несколько событий по блокам — считаем раз
                    d = _usage_delta(msg["usage"])
                    for k in usage:
                        usage[k] += d[k]
                    emit_budget(level, trace_id, {**d, "engine": "claude_code"}, model)
            elif ev.get("type") == "result":
                final = ev
        err = proc.stderr.read()[-500:]
        proc.wait()
        if not final:
            return {"ok": False, "error": f"claude_code: нет результата (rc={proc.returncode}) {err}"}
        if final.get("is_error"):
            return {"ok": False, "error": f"claude_code: {str(final.get('result') or final.get('subtype'))[:400]}"}
        usage["cost_usd"] = final.get("total_cost_usd")
        usage["engine"] = "claude_code"
        usage["model"] = model
        return {"ok": True, "text": final.get("result") or "", "usage": usage, "trace_id": trace_id}
    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)