beaver-agent/policy.py

"""§3.7: правила PreToolUse - граница без permission-промптов.

Модель везде в `bypassPermissions`; что ей можно, решают маунты,
`disallowed_tools` и правила отсюда. Правило получает `ToolCall` и
возвращает `Deny` с причиной, которую модель читает как результат тулзы,
или `None`. Маунт `/vault:ro` - первая линия, хук - вторая: он ловит то,
что маунт не различает (зоны внутри одного тома, порядок «скилл - тулза»).
"""

from __future__ import annotations

import fnmatch
import shlex
from dataclasses import dataclass
from typing import TYPE_CHECKING

from beaver_gateway.core.policy import Deny, PolicyRule, ToolCall

if TYPE_CHECKING:
    from collections.abc import Iterator
    from pathlib import Path

__all__ = [
    "DEEP_DISALLOWED",
    "DISPATCHER_DISALLOWED",
    "DISTILLER_DISALLOWED",
    "TRIAGE_DISALLOWED",
    "Zones",
    "bash_zones",
    "forbid_skills",
    "requires_skill",
    "skill_tracker",
    "vault_zones",
]

# §3.7: у диспетчера AskUserQuestion остаётся, планов и ноутбуков нет;
# у глубоких вопросы текстом, сабагентов нет; дистиллятор и триаж - без веба.
DISPATCHER_DISALLOWED = ("ExitPlanMode", "EnterPlanMode", "NotebookEdit")
DEEP_DISALLOWED = (
    "AskUserQuestion",
    "ExitPlanMode",
    "EnterPlanMode",
    "NotebookEdit",
    "Task",
)
DISTILLER_DISALLOWED = ("AskUserQuestion", "Task", "WebSearch", "WebFetch")
TRIAGE_DISALLOWED = DISTILLER_DISALLOWED

FILE_TOOLS = ("Write", "Edit", "MultiEdit", "NotebookEdit")
MUTATING = frozenset(
    {
        "rm",
        "rmdir",
        "unlink",
        "mv",
        "cp",
        "tee",
        "truncate",
        "touch",
        "mkdir",
        "ln",
        "install",
        "rsync",
        "dd",
        "chmod",
        "chown",
        "shred",
    }
)
COPYING = frozenset({"cp", "install", "rsync"})
DELETING = frozenset({"rm", "rmdir", "unlink", "shred", "dd", "truncate"})
INPLACE = frozenset({"sed", "perl"})
REDIRECTS = frozenset({">", ">>", ">|", "&>", "&>>", ">&"})
SEPARATORS = frozenset({";", "&&", "||", "|", "&", "(", ")"})


@dataclass(frozen=True, slots=True)
class Zones:
    """Зоны vault.

    ``write`` - всё, включая удаление; ``create`` - только новые файлы;
    ``edit`` - создавать, править, переносить можно, удалять - нет;
    ``protected`` - не трогать. Приоритет: write > protected > create > edit;
    путь в vault вне всех зон - отказ. Vault целиком в ``edit`` - это
    «дом» диспетчера (решение h, 2026-08-30): заметки, задачи, люди - его;
    удаление чужого и `.obsidian`/`мета` остаются за Бобром.
    """

    vault: Path
    write: tuple[Path, ...]
    create: tuple[Path, ...] = ()
    edit: tuple[Path, ...] = ()
    protected: tuple[Path, ...] = ()

    def verdict(
        self, path: Path, *, creating: bool, deleting: bool = False
    ) -> Deny | None:
        if not _under(path, self.vault):
            return None
        if any(_under(path, zone) for zone in self.write):
            return None
        rel = path.relative_to(self.vault)
        for zone in self.protected:
            if _under(path, zone):
                return Deny(
                    reason=f"{rel}: «{zone.relative_to(self.vault)}» не трогаем - "
                    "это зона Бобра"
                )
        for zone in self.create:
            if _under(path, zone):
                if creating and not path.exists():
                    return None
                return Deny(
                    reason=f"{rel}: в «{zone.relative_to(self.vault)}» можно только "
                    "создавать новые файлы, существующие не трогаем"
                )
        for zone in self.edit:
            if _under(path, zone):
                if deleting:
                    zones = ", ".join(
                        f"«{z.relative_to(self.vault)}»" for z in self.write
                    )
                    return Deny(
                        reason=f"{rel}: удалять можно только в {zones}; чужие "
                        "заметки не удаляем - скажи Бобру, он сам"
                    )
                return None
        zones = ", ".join(f"«{z.relative_to(self.vault)}»" for z in self.write)
        return Deny(reason=f"{rel}: vault только на чтение; писать можно в {zones}")


def _under(path: Path, root: Path) -> bool:
    return path == root or root in path.parents


def vault_zones(zones: Zones) -> PolicyRule:
    """Write/Edit/NotebookEdit только в зонах; Write в create-зону - новый файл."""

    def rule(call: ToolCall) -> Deny | None:
        if call.tool not in FILE_TOOLS:
            return None
        path = call.path()
        if path is None:
            return None
        return zones.verdict(path, creating=call.tool == "Write")

    return rule


def bash_zones(zones: Zones) -> PolicyRule:
    """`rm`/`mv`/`cp`/`tee`/`sed -i`/редирект с путём в vault вне зон - отказ."""

    def rule(call: ToolCall) -> Deny | None:
        if call.tool != "Bash":
            return None
        command = call.input.get("command")
        if not isinstance(command, str):
            return None
        for target, creating, deleting in _bash_targets(command):
            deny = zones.verdict(
                call.resolve(target), creating=creating, deleting=deleting
            )
            if deny is not None:
                return Deny(reason=f"Bash: {deny.reason}")
        return None

    return rule


def _bash_targets(command: str) -> Iterator[tuple[str, bool, bool]]:
    """``(путь, создаёт, удаляет)`` для каждого пути, который команда трогает."""
    for segment in _segments(command):
        plain: list[str] = []
        i = 0
        while i < len(segment):
            word = segment[i]
            if word in REDIRECTS:
                if i + 1 < len(segment) and _pathlike(segment[i + 1]):
                    yield segment[i + 1], True, False
                i += 2
                continue
            plain.append(word)
            i += 1
        for j, word in enumerate(plain):
            rest = plain[j + 1 :]
            inplace = word in INPLACE and any(a.startswith("-i") for a in rest)
            if word not in MUTATING and not inplace:
                continue
            args = [a for a in rest if _pathlike(a)]
            if word in COPYING:
                args = args[-1:]
            for arg in args:
                yield arg, False, word in DELETING
            break


def _segments(command: str) -> Iterator[list[str]]:
    lexer = shlex.shlex(command, posix=True, punctuation_chars=True)
    lexer.whitespace_split = True
    try:
        tokens = list(lexer)
    except ValueError:
        tokens = command.split()
    segment: list[str] = []
    for token in tokens:
        if token in SEPARATORS:
            if segment:
                yield segment
            segment = []
        else:
            segment.append(token)
    if segment:
        yield segment


def _pathlike(word: str) -> bool:
    return bool(word) and not (
        word.startswith(("-", "&", "$", "/dev/"))
        or word.isdigit()
        or ("=" in word and "/" not in word.split("=", 1)[0])
    )


def _skill_name(call: ToolCall) -> str:
    """Имя скилла из `Skill({"skill": "vault:firefly", ...})`, без плагина."""
    name = str(call.input.get("skill") or call.input.get("name") or "")
    return name.rsplit(":", 1)[-1]


def skill_tracker() -> PolicyRule:
    """Запоминает открытые скиллы в состоянии сессии (для `requires_skill`)."""

    def rule(call: ToolCall) -> None:
        if call.tool != "Skill":
            return
        opened: set[str] = call.state.setdefault("skills", set())
        opened.add(_skill_name(call))

    return rule


def forbid_skills(names: tuple[str, ...], kinds: tuple[str, ...]) -> PolicyRule:
    """§4.3: скиллы из ``names`` (fnmatch, без учёта регистра) в ``kinds`` закрыты.

    Мастер их не открывает - для этого есть ветка; правило отвечает так,
    чтобы модель сама её открыла.
    """
    patterns = tuple(p.lower() for p in names)

    def rule(call: ToolCall) -> Deny | None:
        if call.tool != "Skill" or call.kind not in kinds:
            return None
        name = _skill_name(call)
        if not any(fnmatch.fnmatchcase(name.lower(), p) for p in patterns):
            return None
        return Deny(reason=f"скилл {name} недоступен в {call.kind}: открой ветку")

    return rule


def requires_skill(skill: str, tools: tuple[str, ...]) -> PolicyRule:
    """§4.3: жёсткий протокол - тулзы из ``tools`` только после `Skill(skill)`."""

    def rule(call: ToolCall) -> Deny | None:
        if not any(fnmatch.fnmatchcase(call.tool, pattern) for pattern in tools):
            return None
        if skill in call.state.get("skills", ()):
            return None
        return Deny(
            reason=f"сначала открой скилл «{skill}» (тулза Skill), потом {call.tool}"
        )

    return rule