refactor: split config.py into the beaver_agent package (vault, prompts, skills, policy, hands, agents, frontends, texts, memory, jobs)
This commit is contained in:
@@ -0,0 +1,294 @@
|
||||
"""Граница без permission-промптов: правила PreToolUse и зоны vault.
|
||||
|
||||
Модель везде в `bypassPermissions`; что ей можно, решают `disallowed_tools`
|
||||
и правила отсюда. Правило получает `ToolCall` и возвращает `Deny` с
|
||||
причиной, которую модель читает как результат тулзы, или `None`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import fnmatch
|
||||
import shlex
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from beaver_gateway.agents.policy import Deny, PolicyRule, ToolCall
|
||||
|
||||
from beaver_agent.vault import BEAVER, CHATS, VAULT
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
__all__ = [
|
||||
"DEEP_DISALLOWED",
|
||||
"DISPATCHER_DISALLOWED",
|
||||
"DISTILLER_DISALLOWED",
|
||||
"STRICT_ZONES",
|
||||
"TRIAGE_DISALLOWED",
|
||||
"VAULT_POLICY",
|
||||
"ZONES",
|
||||
"Zones",
|
||||
"bash_zones",
|
||||
"forbid_skills",
|
||||
"requires_skill",
|
||||
"skill_tracker",
|
||||
"vault_zones",
|
||||
]
|
||||
|
||||
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: 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:
|
||||
"""Скиллы из ``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:
|
||||
"""Тулзы из ``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
|
||||
|
||||
|
||||
ZONES = Zones(
|
||||
vault=VAULT,
|
||||
write=(BEAVER,),
|
||||
create=(CHATS,),
|
||||
edit=(VAULT,),
|
||||
protected=(VAULT / ".obsidian", VAULT / "мета"),
|
||||
)
|
||||
"""Vault - дом диспетчера: создавать, править, переносить можно везде, кроме
|
||||
`.obsidian` и `мета` вне `бобер`; удалять - только своё и новые чаты."""
|
||||
|
||||
STRICT_ZONES = Zones(vault=VAULT, write=(BEAVER,), create=(CHATS,))
|
||||
"""Дистиллятор и куратор: писать только в своей зоне."""
|
||||
|
||||
FIREFLY_WRITES = ("mcp__firefly__store_*", "mcp__firefly__update_*")
|
||||
VAULT_POLICY = (
|
||||
skill_tracker(),
|
||||
vault_zones(ZONES),
|
||||
bash_zones(ZONES),
|
||||
requires_skill("firefly", FIREFLY_WRITES),
|
||||
)
|
||||
Reference in New Issue
Block a user