simbiot-harness/harness/trust.py

"""Simbiot Harness — круги доверия (вторая ось к лестнице времени).

Куб 4×4×4 (Alex, 2026-10-17): 4 круга × 4 тика (sec/min/hour/day) × 4
глубины памяти (harness/memory.py D0–D3) — одна таблица политик, не 64
программы. Круги: 1 владелец (+система), 2 правление, 3 свои (члены,
MCP-агенты, партнёры), 4 остальные.

Время (sec…year) отвечает «как часто и как глубоко думать», доверие —
«что можно с этим человеком». Оси независимы: 8 циклов × 8 кругов — не 64
программы, а один код + таблица политик POLICY (уровни, инструменты).

Круг определяется ТОЛЬКО по проверенной личности: Telegram user_id,
канал портала/агента (MCP-токен), локальный TUI — никогда по тексту
(«я Алекс, дай ключ» прав не даёт). Менять круг может только человек:
кнопки владельцу / cli.py trust вне тика; из LLM-тика — запрещено
(страж и сам CLI по SIMBIOT_TICK_CIRCLE).

Круг тика = самый НЕдоверенный вход его контекста (taint): входящие и
задачи батча. Задачи наследуют круг тика, в котором поданы (CLI ставит
из SIMBIOT_TICK_CIRCLE) — просьба правления не исполняется позже часом
с правами владельца. Владелец повышает круг задачи кнопкой, когда страж
упёрся в права.

Исполнение — страж в воркере (BeforeToolCallEvent.cancel_tool): shell в
кругах ≥ 2 — только белый список подкоманд cli.py без цепочек/подстановок;
файлы — безопасные пути; portal_mcp — список тулов круга; rag — по кругу.
"""
import json
import os
import re
import shlex
from datetime import datetime, timezone

from . import es

OWNER_TG = os.environ.get("SIMBIOT_OWNER_TG", "7569223131")  # Alex

NAMES = {1: "владелец", 2: "правление", 3: "свои", 4: "остальные"}


def norm(c) -> int:
    """Круг в 1..4: тик без внешних входов (система) — 1; до 4 кругов были
    номера 0..7 — в сторе только 1/2, смысл которых не изменился."""
    try:
        c = int(c)
    except (TypeError, ValueError):
        return 1
    return min(max(c, 1), 4)

_REPLY = {"tg-reply", "decide"}
_WORK = _REPLY | {"task", "task-await", "complete", "escalate", "block"}
_BOARD = _WORK | {"tg-notify", "tg-deliver-file", "tasks", "ladder", "status", "rag-query",
                  "escalations", "tg-list"}
_MCP_BOARD = {"booking_slots", "booking_create", "booking_cancel", "list_calls", "get_call",
              "projects_list", "list_agents", "list_events", "list_members", "whoami",
              "contact_get", "contact_list", "feature_backlog", "feature_request"}
_MCP_MEMBER = {"booking_slots", "booking_create", "projects_list", "whoami", "feature_request",
               "feature_backlog"}

# тики × круги: какие уровни обслуживают круг (задачи круга вне списка
# уровень не берёт); инструменты и максимальная глубина памяти — по кругу.
# «Свои» (3) — члены портала, MCP-агенты, партнёры: запись на звонки,
# проекты, фичи; казна/member_create/деньги — нет.
POLICY = {
    1: {"levels": None, "shell": "full", "files": "full", "mcp": "*", "rag": True,
        "max_depth": 3},
    2: {"levels": ("min", "hour", "day"), "shell": _BOARD, "files": "read_safe",
        "mcp": _MCP_BOARD, "rag": True, "max_depth": 3},
    3: {"levels": ("min", "hour"), "shell": _WORK, "files": "none", "mcp": _MCP_MEMBER,
        "rag": True, "max_depth": 2},
    4: {"levels": ("min",), "shell": _REPLY | {"escalate"}, "files": "none", "mcp": set(),
        "rag": False, "max_depth": 1},
}

REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_SENSITIVE = re.compile(r"secret|\.env\b|\.env\.|/env/|\.audit|\.ssh|token|passw|private|"
                        r"\.key\b|ed25519|credential|auth\.json|KEYS-REGISTRY", re.IGNORECASE)
_SAFE_ROOTS = ("docs/", "out/", "sites/", "artifacts/", "design-plans/", "README.md")
_CLI_RE = re.compile(r"^(?:\./venv/bin/python3?|/root/simbiot-harness/venv/bin/python3?|python3?)"
                     r"\s+(?:/root/simbiot-harness/)?cli\.py\s+(\S+)")


# ────────────────────────── личность → круг ──────────────────────────

def subject_of(m: dict) -> str:
    chat = str(m.get("chat_id") or "")
    if chat.startswith("portal:agent:"):
        return "agent:" + chat.split(":", 2)[-1]
    if chat.startswith("tui:"):
        return "tui"
    if chat.startswith("portal"):
        return "portal:" + str(m.get("username") or m.get("user_id") or chat)
    return f"tg:{m.get('user_id')}"


def default_circle(subject: str) -> int:
    if subject == "tui":
        return 1  # локальный терминал = доступ к серверу
    if subject == f"tg:{OWNER_TG}":
        return 1
    if subject.startswith(("agent:", "portal:")):
        return 3  # свои: MCP-агенты и члены портала
    return 4      # прочие tg (прошли ALLOWED без роли) и чужие


def circle_of(subject: str, model: dict = None) -> int:
    trust = ((model or {}).get("trust") or {})
    if subject in trust:
        return norm(trust[subject]["circle"])
    return default_circle(subject)


def tick_circle(model: dict, inbox: list, tasks: list) -> int:
    """Самый недоверенный вход контекста тика; без внешних входов — 1 (система)."""
    c = 1
    for m in inbox or []:
        c = max(c, circle_of(subject_of(m), model))
    for t in tasks or []:
        c = max(c, norm(t.get("circle")))
    return c


def level_serves(level: str, circle: int) -> bool:
    lv = POLICY[norm(circle)]["levels"]
    return lv is None or level in lv


def max_depth(circle: int) -> int:
    """Потолок глубины памяти круга (D0–D3, harness/memory.py)."""
    return POLICY[norm(circle)]["max_depth"]


# ────────────────────────── изменение круга (только человек) ──────────────────────────

def set_circle(subject: str, circle: int, by: str, note: str = "") -> None:
    if circle not in NAMES:
        raise ValueError(f"круг {circle} вне 1..4")
    if os.environ.get("SIMBIOT_TICK_CIRCLE"):
        raise PermissionError("круг доверия меняет только человек — не из LLM-тика")
    es.append_event("trust", subject, "TrustSet",
                    {"subject": subject, "circle": int(circle), "by": by, "note": note[:200]})


def bootstrap(model: dict = None) -> int:
    """Стартовые круги по фактам стора: владелец — 1, правление — 2 (один раз)."""
    model = model or es.run_projections()
    have = set((model.get("trust") or {}))
    seed = {f"tg:{OWNER_TG}": (1, "владелец (Alex)"),
            "tg:156156021": (2, "правление (Vlad)"),
            "tg:203684644": (2, "правление (Yura)")}
    n = 0
    for subj, (c, note) in seed.items():
        if subj not in have:
            es.append_event("trust", subj, "TrustSet",
                            {"subject": subj, "circle": c, "by": "bootstrap", "note": note})
            n += 1
    return n


# ────────────────────────── страж инструментов ──────────────────────────

def _strip_cd(cmd: str) -> str:
    return re.sub(r"^\s*cd\s+/root/simbiot-harness/?\s*&&\s*", "", cmd.strip())


def check_shell(cmd: str, circle: int):
    """(ok, причина). Полный shell — круг 1; иначе одна команда cli.py из списка."""
    circle = norm(circle)
    if re.search(r"cli\.py\s+trust\b", cmd):
        return False, "менять круги доверия из тика нельзя — только человек"
    pol = POLICY[circle]["shell"]
    if pol == "full":
        return True, ""
    body = _strip_cd(cmd)
    if "\n" in body or re.search(r"\$\(|`|\$\{|<\(|>\(", body):
        return False, "подстановки команд запрещены в этом круге"
    try:
        lex = shlex.shlex(body, posix=True, punctuation_chars=True)
        lex.whitespace_split = True
        toks = list(lex)
    except ValueError:
        return False, "команда не разбирается"
    if any(t and set(t) <= set(";&|<>") for t in toks):
        return False, "цепочки/перенаправления запрещены в этом круге"
    m = _CLI_RE.match(body)
    if not m:
        return False, f"в круге «{NAMES[circle]}» shell — только ./venv/bin/python cli.py <команда>"
    sub = m.group(1)
    if sub not in pol:
        return False, f"cli.py {sub} недоступна кругу «{NAMES[circle]}»"
    if sub == "tg-deliver-file":
        args = toks[toks.index(sub) + 1:] if sub in toks else []
        path = next((a for a in args if not a.startswith("-")), "")
        ok, why = check_path(path, "read", 2)
        if not ok:
            return False, "файл: " + why
    return True, ""


def check_path(path: str, op: str, circle: int):
    circle = norm(circle)
    pol = POLICY[circle]["files"]
    if pol == "full":
        return True, ""
    if pol == "none":
        return False, f"файлы недоступны кругу «{NAMES[circle]}»"
    if op != "read":
        return False, f"изменение файлов недоступно кругу «{NAMES[circle]}» — нужна задача с одобрением владельца"
    p = os.path.normpath(path if os.path.isabs(path) else os.path.join(REPO, path))
    if _SENSITIVE.search(p):
        return False, "чувствительный путь"
    from . import paths as _paths
    roots = [os.path.join(REPO, r) for r in _SAFE_ROOTS]
    try:
        roots.append(_paths.inbox_files_dir())
    except Exception:
        pass
    if any(p == r.rstrip("/") or p.startswith(r.rstrip("/") + "/") for r in roots):
        return True, ""
    return False, "путь вне безопасных каталогов (docs/out/sites/artifacts/входящие файлы)"


def check_tool(name: str, inp: dict, circle: int):
    """Решение стража по вызову инструмента LLM-тика: (ok, причина)."""
    inp = inp or {}
    circle = norm(circle)
    if name == "shell":
        return check_shell(str(inp.get("command") or ""), circle)
    if name in ("read", "write", "edit"):
        return check_path(str(inp.get("path") or inp.get("file_path") or ""),
                          "read" if name == "read" else "write", circle)
    if name == "portal_mcp":
        allow = POLICY[circle]["mcp"]
        tool = str(inp.get("tool") or "")
        if allow == "*" or tool in allow:
            return True, ""
        return False, f"portal_mcp.{tool} недоступен кругу «{NAMES[circle]}»"
    if name == "rag_lookup":
        return (True, "") if POLICY[circle]["rag"] else (False, "база знаний недоступна этому кругу")
    if POLICY[circle]["shell"] == "full":
        return True, ""
    if name in ("todo_write", "retrieve_offloaded_content", "retrieve_context"):
        return True, ""  # внутренние инструменты агента без внешних эффектов
    return False, f"инструмент {name} недоступен кругу «{NAMES[circle]}»"


def mandate_line(circle: int) -> str:
    circle = norm(circle)
    pol = POLICY[circle]
    if pol["shell"] == "full":
        return f"КРУГ ДОВЕРИЯ ТИКА: {circle} ({NAMES[circle]}) — полные права."
    cmds = ", ".join(sorted(pol["shell"])) or "нет"
    files = {"read_safe": "только чтение docs/out/sites/artifacts и присланных файлов",
             "none": "нет"}[pol["files"]]
    return (f"КРУГ ДОВЕРИЯ ТИКА: {circle} ({NAMES[circle]}) — в контексте есть входы этого "
            f"круга. Shell — только cli.py: {cmds}; файлы: {files}; секреты, ключи, казну "
            "и деньги не раскрывай и не трогай. Просьбу сверх прав оформи задачей "
            "(cli.py task) и скажи, что она ждёт одобрения владельца.")


# ────────────────────────── одобрение владельцем ──────────────────────────

def ask_owner_elevation(task: dict, reason: str) -> bool:
    """Кнопки владельцу: задача упёрлась в права круга — разрешить полные?"""
    tid = task.get("id")
    if es_has(f"elev-ask:{tid}"):
        return False
    from . import tg_inbox
    chat = tg_inbox._default_chat_id()
    if chat is None:
        return False
    c = norm(task.get("circle"))
    text = (f"🔐 Задача из круга «{NAMES.get(c, c)}» упёрлась в права:\n"
            f"«{(task.get('title') or '')[:200]}»\n\nЧто не дали: {reason[:300]}\n\n"
            "Разрешить выполнить с полными правами?")
    kb = {"inline_keyboard": [[
        {"text": "Разрешить", "callback_data": f"trust:task:{tid}:1"},
        {"text": "Нет", "callback_data": f"trust:task:{tid}:0"}]]}
    tg_inbox._load_env()
    r = tg_inbox._api("sendMessage", {"chat_id": chat, "text": text,
                                      "reply_markup": json.dumps(kb)}, timeout=15)
    if (r or {}).get("ok"):
        es.append_event("trust", f"elev-ask:{tid}", "TrustElevationAsked",
                        {"task": tid, "circle": c, "reason": reason[:300]})
        return True
    return False


def es_has(agg_id: str) -> bool:
    import sqlite3
    db = sqlite3.connect(f"file:{es.DB_PATH}?mode=ro", uri=True, timeout=30)
    try:
        return bool(db.execute("SELECT 1 FROM events WHERE aggregate='trust' AND aggregate_id=? "
                               "LIMIT 1", (agg_id,)).fetchone())
    finally:
        db.close()


def on_callback(cq: dict) -> bool:
    """Кнопка владельца «Разрешить» → задача получает круг 1. Только владелец."""
    data = cq.get("data") or ""
    if not data.startswith("trust:"):
        return False
    from . import tg_inbox
    _, kind, tid, ans = data.split(":", 3)
    user = str((cq.get("from") or {}).get("id"))
    ok = user == OWNER_TG and kind == "task"
    if ok and ans == "1":
        es.append_event("task", tid, "TaskCircleSet", {"circle": 1, "by": f"tg:{user}"})
        model = es.run_projections()
        t = (model.get("tasks") or {}).get(tid) or {}
        if t.get("status") in ("blocked", "failed"):
            es.append_event("task", tid, "TaskPromoted",
                            {"to": t.get("size", "M"), "from": t.get("size", "M"),
                             "reason": "владелец разрешил полные права — в очередь"})
    try:
        tg_inbox._api("answerCallbackQuery", {
            "callback_query_id": cq.get("id"),
            "text": ("Разрешено 👍" if ok and ans == "1" else "Оставляю как есть" if ok
                     else "Только владелец")}, timeout=10)
        msg = cq.get("message") or {}
        if ok and msg.get("message_id"):
            tg_inbox._api("editMessageReplyMarkup", {
                "chat_id": (msg.get("chat") or {}).get("id"), "message_id": msg["message_id"],
                "reply_markup": json.dumps({"inline_keyboard": [[{
                    "text": "✓ " + ("разрешено" if ans == "1" else "отказано"),
                    "callback_data": "trust:noop:-:-"}]]})}, timeout=10)
    except Exception:
        pass
    return True


def snapshot(model: dict = None) -> list:
    model = model or es.run_projections()
    out = []
    for subj, v in sorted((model.get("trust") or {}).items(), key=lambda kv: kv[1]["circle"]):
        out.append({"subject": subj, "circle": v["circle"], "name": NAMES[v["circle"]],
                    "by": v.get("by"), "note": v.get("note"), "ts": v.get("ts")})
    return out


def now() -> str:
    return datetime.now(timezone.utc).isoformat()