Практика · Multi-agent systems

Проверяем наследование правил и памяти между AI-агентами в FROST

Потомок не должен получать власть над системой только потому, что родитель передал ему задачу. В этом практикуме мы вынесем правила доступа из промптов в детерминированный слой, разрешим дочернему агенту читать унаследованную память, запретим изменять данные предка и остановим SOP, который требует запрещённого инструмента.

Продвинутый уровень 60 минут Результат: рабочий прототип и набор тестов

Что именно мы построим

AI-агент в этом стенде — субъект с идентификатором, родителем, собственной областью памяти и набором разрешённых возможностей. FROST отвечает за создание и запуск иерархии, а отдельный governance-слой принимает окончательное решение перед чтением, записью или выполнением процесса.

Прототип проверяет три обязательных свойства:

  1. потомок читает разрешённые данные из памяти предков, если у него нет собственного значения;
  2. потомок не может изменить или удалить запись, владельцем которой является предок;
  3. недопустимый SOP отклоняется целиком до выполнения первого шага.

Мы намеренно не поручаем модели решать, можно ли ей выполнить действие. Модель может предложить операцию, но допуск рассчитывается обычным кодом по версии правил, происхождению памяти и возможностям агента.

Конкретный случай: команда подготовки отчёта

Родительский агент report_manager хранит общие ограничения проекта и делегирует сбор материалов агенту researcher. Тот может создать ещё одного потомка — formatter.

У родителя есть такая память:

  • project.client_region = "RU" — доступна потомкам для чтения;
  • policy.external_upload = false — обязательное ограничение;
  • report.retention_days = 30 — настройка, которую может менять только владелец;
  • credentials.storage_token — секрет, не наследуемый потомками.

Исследователь должен прочитать регион и срок хранения, сохранить собственные заметки и выполнить допустимый процесс подготовки черновика. Атакующий сценарий просит его изменить срок хранения в памяти родителя и запустить SOP с выгрузкой исходных документов во внешний сервис.

Если ограничения находятся только в системном промпте, тест проверит главным образом послушность модели. Нам нужен более сильный инвариант: запрещённая операция не достигает хранилища или инструмента независимо от текста ответа модели.

Модель наследования: значение, происхождение и право записи

Память агента нельзя представлять одним общим словарём. После объединения словарей теряется происхождение записи, а вместе с ним — возможность доказать, кто имеет право её менять.

Каждая запись должна содержать как минимум:

{
  "namespace": "report",
  "key": "retention_days",
  "value": 30,
  "owner_agent_id": "report_manager",
  "visibility": "descendants",
  "mutable_by": "owner",
  "revision": 1
}

Для чтения мы идём от текущего агента к корню и берём ближайшую видимую запись. Для записи не применяем такой поиск: операция всегда адресуется конкретной области владельца. Это различие защищает от случайной перезаписи родительского значения.

Наследование правил должно быть монотонным: потомок может получить более узкие права, но не расширить ограничения родителя. Если родитель запрещает external_upload, локальная конфигурация потомка не должна превращать запрет в разрешение.

Эффективные возможности удобно вычислять как пересечение:

effective_capabilities(child) =
    effective_capabilities(parent)
    ∩ declared_capabilities(child)
    ∩ environment_capabilities

Так реализуется принцип минимальных привилегий: делегирование задачи не означает делегирование всех инструментов родителя.

Граница угроз

Перед кодом зафиксируем, от чего именно защищает стенд.

Риск Пример Контроль
Перезапись памяти предка Потомок меняет retention_days Проверка владельца перед записью
Чтение закрытой записи Потомок запрашивает токен родителя Проверка видимости при разрешении памяти
Расширение прав Потомок объявляет себе external_upload Пересечение возможностей
Частичное выполнение плохого SOP Сначала читаются файлы, затем обнаруживается запрещённая отправка Полная предварительная валидация плана
Обход через неизвестный шаг SOP использует новый тип shell_eval Режим запрещено по умолчанию
Гонка между проверкой и записью Запись изменилась после проверки revision Условное обновление по ревизии

Прототип не пытается защищаться от администратора хоста, который напрямую изменяет базу или код policy engine. Такая угроза требует изоляции процесса, подписанных политик и отдельного доверенного хранилища.

Шаг 1. Создаём проект

Нужны Python 3.11 или новее и pytest. Сам governance-слой не зависит от SDK модели и не требует ключей доступа.

mkdir frost-governance-lab
cd frost-governance-lab
python -m venv .venv
source .venv/bin/activate
python -m pip install pytest
mkdir -p frost_governance tests
touch frost_governance/__init__.py
touch frost_governance/core.py
touch tests/test_governance.py

В Windows PowerShell активация окружения выглядит так:

.\.venv\Scripts\Activate.ps1

Структура после подготовки:

frost-governance-lab/
├── frost_governance/
│   ├── __init__.py
│   └── core.py
└── tests/
    └── test_governance.py

Шаг 2. Реализуем доверенный слой

Поместите следующий код в frost_governance/core.py. Он не имитирует рассуждения модели: это исполняемый контур авторизации, который должен находиться между FROST и реальными хранилищами или инструментами.

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Iterable, Literal


Visibility = Literal["private", "children", "descendants"]
MutableBy = Literal["owner"]


class GovernanceError(RuntimeError):
    pass


class AccessDenied(GovernanceError):
    pass


class ValidationError(GovernanceError):
    pass


class RevisionConflict(GovernanceError):
    pass


@dataclass(frozen=True)
class Agent:
    agent_id: str
    parent_id: str | None
    declared_capabilities: frozenset[str]


@dataclass
class MemoryRecord:
    namespace: str
    key: str
    value: Any
    owner_agent_id: str
    visibility: Visibility = "private"
    mutable_by: MutableBy = "owner"
    revision: int = 1


@dataclass(frozen=True)
class SOPStep:
    step_id: str
    action: str
    arguments: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class SOP:
    sop_id: str
    version: int
    steps: tuple[SOPStep, ...]


@dataclass(frozen=True)
class Decision:
    allowed: bool
    subject: str
    operation: str
    resource: str
    reason: str


class Governance:
    KNOWN_ACTIONS = frozenset({
        "memory.read",
        "memory.write",
        "document.summarize",
        "artifact.save",
        "external.upload",
    })

    def __init__(
        self,
        agents: Iterable[Agent],
        environment_capabilities: Iterable[str],
    ) -> None:
        agent_list = list(agents)
        self.agents = {agent.agent_id: agent for agent in agent_list}
        if len(self.agents) != len(agent_list):
            raise ValidationError("duplicate agent_id")

        self.environment_capabilities = frozenset(
            environment_capabilities
        )
        self.memory: dict[tuple[str, str, str], MemoryRecord] = {}
        self.audit: list[Decision] = []
        self._validate_hierarchy()

    def _validate_hierarchy(self) -> None:
        for agent in self.agents.values():
            if (
                agent.parent_id is not None
                and agent.parent_id not in self.agents
            ):
                raise ValidationError(
                    f"unknown parent: {agent.parent_id}"
                )

            seen: set[str] = set()
            cursor: Agent | None = agent
            while cursor is not None:
                if cursor.agent_id in seen:
                    raise ValidationError("agent hierarchy contains a cycle")
                seen.add(cursor.agent_id)
                cursor = (
                    self.agents[cursor.parent_id]
                    if cursor.parent_id is not None
                    else None
                )

    def lineage(self, agent_id: str) -> list[str]:
        result: list[str] = []
        cursor = self.agents[agent_id]
        while True:
            result.append(cursor.agent_id)
            if cursor.parent_id is None:
                return result
            cursor = self.agents[cursor.parent_id]

    def is_descendant(
        self,
        candidate_id: str,
        ancestor_id: str,
    ) -> bool:
        return ancestor_id in self.lineage(candidate_id)[1:]

    def is_direct_child(
        self,
        candidate_id: str,
        parent_id: str,
    ) -> bool:
        return self.agents[candidate_id].parent_id == parent_id

    def effective_capabilities(
        self,
        agent_id: str,
    ) -> frozenset[str]:
        lineage = self.lineage(agent_id)
        effective = set(self.environment_capabilities)
        for item_id in reversed(lineage):
            effective.intersection_update(
                self.agents[item_id].declared_capabilities
            )
        return frozenset(effective)

    def _record(self, decision: Decision) -> Decision:
        self.audit.append(decision)
        return decision

    def put_initial(self, record: MemoryRecord) -> None:
        storage_key = (
            record.owner_agent_id,
            record.namespace,
            record.key,
        )
        if record.owner_agent_id not in self.agents:
            raise ValidationError("unknown memory owner")
        if storage_key in self.memory:
            raise ValidationError("initial record already exists")
        self.memory[storage_key] = record

    def _visible(
        self,
        reader_id: str,
        record: MemoryRecord,
    ) -> bool:
        if reader_id == record.owner_agent_id:
            return True
        if record.visibility == "private":
            return False
        if record.visibility == "children":
            return self.is_direct_child(
                reader_id,
                record.owner_agent_id,
            )
        return self.is_descendant(
            reader_id,
            record.owner_agent_id,
        )

    def resolve(
        self,
        reader_id: str,
        namespace: str,
        key: str,
    ) -> MemoryRecord:
        resource = f"{namespace}.{key}"

        if "memory.read" not in self.effective_capabilities(reader_id):
            self._record(Decision(
                False,
                reader_id,
                "memory.read",
                resource,
                "capability_not_inherited",
            ))
            raise AccessDenied("memory.read is not allowed")

        for owner_id in self.lineage(reader_id):
            record = self.memory.get((owner_id, namespace, key))
            if record is None:
                continue
            if self._visible(reader_id, record):
                self._record(Decision(
                    True,
                    reader_id,
                    "memory.read",
                    resource,
                    f"visible_from:{owner_id}",
                ))
                return record

            self._record(Decision(
                False,
                reader_id,
                "memory.read",
                resource,
                f"not_visible_from:{owner_id}",
            ))
            raise AccessDenied("record is not visible")

        self._record(Decision(
            False,
            reader_id,
            "memory.read",
            resource,
            "record_not_found",
        ))
        raise KeyError(resource)

    def write(
        self,
        actor_id: str,
        owner_id: str,
        namespace: str,
        key: str,
        value: Any,
        expected_revision: int | None = None,
    ) -> MemoryRecord:
        resource = f"{owner_id}:{namespace}.{key}"

        if "memory.write" not in self.effective_capabilities(actor_id):
            self._record(Decision(
                False,
                actor_id,
                "memory.write",
                resource,
                "capability_not_inherited",
            ))
            raise AccessDenied("memory.write is not allowed")

        if actor_id != owner_id:
            self._record(Decision(
                False,
                actor_id,
                "memory.write",
                resource,
                "actor_is_not_owner",
            ))
            raise AccessDenied("cannot modify another agent's memory")

        storage_key = (owner_id, namespace, key)
        current = self.memory.get(storage_key)

        if current is not None:
            if (
                expected_revision is not None
                and current.revision != expected_revision
            ):
                self._record(Decision(
                    False,
                    actor_id,
                    "memory.write",
                    resource,
                    "revision_conflict",
                ))
                raise RevisionConflict(resource)

            current.value = value
            current.revision += 1
            result = current
        else:
            result = MemoryRecord(
                namespace=namespace,
                key=key,
                value=value,
                owner_agent_id=owner_id,
                visibility="private",
            )
            self.memory[storage_key] = result

        self._record(Decision(
            True,
            actor_id,
            "memory.write",
            resource,
            "owner_write",
        ))
        return result

    def validate_sop(
        self,
        actor_id: str,
        sop: SOP,
    ) -> tuple[Decision, ...]:
        capabilities = self.effective_capabilities(actor_id)
        decisions: list[Decision] = []
        seen_ids: set[str] = set()

        if not sop.steps:
            raise ValidationError("SOP must contain at least one step")

        for step in sop.steps:
            resource = f"{sop.sop_id}:{step.step_id}"

            if step.step_id in seen_ids:
                decision = Decision(
                    False,
                    actor_id,
                    "sop.validate",
                    resource,
                    "duplicate_step_id",
                )
            elif step.action not in self.KNOWN_ACTIONS:
                decision = Decision(
                    False,
                    actor_id,
                    "sop.validate",
                    resource,
                    "unknown_action",
                )
            elif step.action not in capabilities:
                decision = Decision(
                    False,
                    actor_id,
                    "sop.validate",
                    resource,
                    "action_not_in_effective_capabilities",
                )
            elif (
                step.action == "memory.write"
                and step.arguments.get("owner_agent_id") != actor_id
            ):
                decision = Decision(
                    False,
                    actor_id,
                    "sop.validate",
                    resource,
                    "cross_owner_memory_write",
                )
            else:
                decision = Decision(
                    True,
                    actor_id,
                    "sop.validate",
                    resource,
                    "step_allowed",
                )

            decisions.append(decision)
            seen_ids.add(step.step_id)

        self.audit.extend(decisions)

        rejected = [item for item in decisions if not item.allowed]
        if rejected:
            reasons = ", ".join(item.reason for item in rejected)
            raise AccessDenied(f"SOP rejected: {reasons}")

        return tuple(decisions)

Ключевой участок — validate_sop. Он просматривает весь процесс до исполнения. Если последний шаг запрещён, система не должна успеть выполнить первые разрешённые шаги.

Список KNOWN_ACTIONS является списком разрешённых операций. Неизвестное действие отклоняется, даже если оно выглядит безобидно или модель убедительно объясняет его необходимость.

Шаг 3. Добавляем воспроизводимые тесты

Поместите код ниже в tests/test_governance.py. Тесты проверяют фактическое состояние и журнал решений, а не текст, который вернул агент.

import pytest

from frost_governance.core import (
    AccessDenied,
    Agent,
    Governance,
    MemoryRecord,
    RevisionConflict,
    SOP,
    SOPStep,
)


ALL = frozenset({
    "memory.read",
    "memory.write",
    "document.summarize",
    "artifact.save",
    "external.upload",
})


@pytest.fixture
def governance() -> Governance:
    root = Agent(
        agent_id="report_manager",
        parent_id=None,
        declared_capabilities=ALL,
    )
    researcher = Agent(
        agent_id="researcher",
        parent_id="report_manager",
        declared_capabilities=frozenset({
            "memory.read",
            "memory.write",
            "document.summarize",
            "artifact.save",
        }),
    )
    formatter = Agent(
        agent_id="formatter",
        parent_id="researcher",
        declared_capabilities=frozenset({
            "memory.read",
            "artifact.save",
        }),
    )

    service = Governance(
        agents=[root, researcher, formatter],
        environment_capabilities=ALL,
    )

    service.put_initial(MemoryRecord(
        namespace="project",
        key="client_region",
        value="RU",
        owner_agent_id="report_manager",
        visibility="descendants",
    ))
    service.put_initial(MemoryRecord(
        namespace="report",
        key="retention_days",
        value=30,
        owner_agent_id="report_manager",
        visibility="descendants",
    ))
    service.put_initial(MemoryRecord(
        namespace="credentials",
        key="storage_token",
        value="test-only-placeholder",
        owner_agent_id="report_manager",
        visibility="private",
    ))
    return service


def test_child_inherits_visible_parent_memory(governance):
    record = governance.resolve(
        "researcher",
        "project",
        "client_region",
    )

    assert record.value == "RU"
    assert record.owner_agent_id == "report_manager"
    assert governance.audit[-1].allowed is True
    assert governance.audit[-1].reason == (
        "visible_from:report_manager"
    )


def test_grandchild_inherits_descendant_memory(governance):
    record = governance.resolve(
        "formatter",
        "report",
        "retention_days",
    )

    assert record.value == 30
    assert record.owner_agent_id == "report_manager"


def test_private_parent_memory_is_hidden(governance):
    with pytest.raises(AccessDenied):
        governance.resolve(
            "researcher",
            "credentials",
            "storage_token",
        )

    assert governance.audit[-1].allowed is False
    assert governance.audit[-1].reason.startswith(
        "not_visible_from:"
    )


def test_child_can_write_only_its_own_memory(governance):
    created = governance.write(
        actor_id="researcher",
        owner_id="researcher",
        namespace="notes",
        key="draft",
        value="Локальные заметки",
    )

    assert created.owner_agent_id == "researcher"
    assert created.value == "Локальные заметки"


def test_child_cannot_modify_parent_memory(governance):
    original = governance.resolve(
        "researcher",
        "report",
        "retention_days",
    )

    with pytest.raises(AccessDenied):
        governance.write(
            actor_id="researcher",
            owner_id="report_manager",
            namespace="report",
            key="retention_days",
            value=0,
            expected_revision=original.revision,
        )

    unchanged = governance.memory[
        ("report_manager", "report", "retention_days")
    ]
    assert unchanged.value == 30
    assert unchanged.revision == 1
    assert governance.audit[-1].reason == "actor_is_not_owner"


def test_stale_revision_is_rejected(governance):
    governance.write(
        actor_id="report_manager",
        owner_id="report_manager",
        namespace="report",
        key="retention_days",
        value=45,
        expected_revision=1,
    )

    with pytest.raises(RevisionConflict):
        governance.write(
            actor_id="report_manager",
            owner_id="report_manager",
            namespace="report",
            key="retention_days",
            value=60,
            expected_revision=1,
        )

    assert governance.memory[
        ("report_manager", "report", "retention_days")
    ].value == 45


def test_child_cannot_expand_parent_capabilities(governance):
    effective = governance.effective_capabilities("researcher")

    assert "external.upload" not in effective
    assert "document.summarize" in effective


def test_grandchild_is_narrower_than_parent(governance):
    effective = governance.effective_capabilities("formatter")

    assert effective == frozenset({
        "memory.read",
        "artifact.save",
    })


def test_valid_sop_is_accepted(governance):
    sop = SOP(
        sop_id="prepare-report",
        version=1,
        steps=(
            SOPStep(
                "read-region",
                "memory.read",
                {"namespace": "project", "key": "client_region"},
            ),
            SOPStep(
                "summarize",
                "document.summarize",
                {"source": "workspace"},
            ),
            SOPStep(
                "save-draft",
                "artifact.save",
                {"target": "draft/report.md"},
            ),
        ),
    )

    decisions = governance.validate_sop("researcher", sop)

    assert len(decisions) == 3
    assert all(item.allowed for item in decisions)


def test_sop_with_external_upload_is_rejected(governance):
    sop = SOP(
        sop_id="unsafe-export",
        version=1,
        steps=(
            SOPStep(
                "summarize",
                "document.summarize",
                {"source": "workspace"},
            ),
            SOPStep(
                "upload",
                "external.upload",
                {"destination": "untrusted-service"},
            ),
        ),
    )

    with pytest.raises(
        AccessDenied,
        match="action_not_in_effective_capabilities",
    ):
        governance.validate_sop("researcher", sop)

    rejected = [
        item for item in governance.audit
        if item.resource == "unsafe-export:upload"
    ]
    assert len(rejected) == 1
    assert rejected[0].allowed is False


def test_sop_cannot_write_parent_memory(governance):
    sop = SOP(
        sop_id="rewrite-policy",
        version=1,
        steps=(
            SOPStep(
                "rewrite-retention",
                "memory.write",
                {
                    "owner_agent_id": "report_manager",
                    "namespace": "report",
                    "key": "retention_days",
                    "value": 0,
                },
            ),
        ),
    )

    with pytest.raises(
        AccessDenied,
        match="cross_owner_memory_write",
    ):
        governance.validate_sop("researcher", sop)


def test_unknown_sop_action_is_rejected(governance):
    sop = SOP(
        sop_id="unknown-action",
        version=1,
        steps=(
            SOPStep(
                "run-code",
                "shell_eval",
                {"code": "do_something()"},
            ),
        ),
    )

    with pytest.raises(
        AccessDenied,
        match="unknown_action",
    ):
        governance.validate_sop("researcher", sop)


def test_duplicate_step_ids_are_rejected(governance):
    sop = SOP(
        sop_id="ambiguous",
        version=1,
        steps=(
            SOPStep("save", "artifact.save", {}),
            SOPStep("save", "artifact.save", {}),
        ),
    )

    with pytest.raises(
        AccessDenied,
        match="duplicate_step_id",
    ):
        governance.validate_sop("researcher", sop)

Шаг 4. Запускаем и проверяем

Из корня проекта выполните:

python -m pytest -q

Не фиксируйте в документации заранее придуманное число успешных тестов. Считайте запуск подтверждённым только по фактическому коду возврата 0 и отчёту pytest в вашей среде.

Для подробной диагностики решений:

python -m pytest -vv
python -m pytest -k "parent_memory or sop" -vv

Убедитесь, что стенд доказывает не только возникновение исключения, но и отсутствие запрещённого эффекта:

  • значение родителя осталось 30;
  • ревизия запрещённой записи не увеличилась;
  • в журнале есть решение allowed=false с машинно-читаемой причиной;
  • неизвестная операция SOP не трактуется как разрешённая;
  • возможности потомка являются подмножеством возможностей предка.

Именно проверка состояния отличает тест безопасности от проверки формулировки ответа. Фраза «я не буду изменять память» ничего не доказывает, если запись уже изменилась.

Шаг 5. Подключаем слой к FROST

В интеграции должны существовать две независимые фазы: планирование и выполнение. FROST или модель формирует структурированное предложение, governance-слой проверяет его, а исполнитель получает только уже разрешённую операцию.

async def run_frost_agent(agent_id, task, frost_runtime, governance):
    proposal = await frost_runtime.plan(
        agent_id=agent_id,
        task=task,
    )

    if proposal["kind"] == "sop":
        sop = parse_sop(proposal)
        governance.validate_sop(agent_id, sop)
        return await execute_validated_sop(
            actor_id=agent_id,
            sop=sop,
            governance=governance,
        )

    if proposal["kind"] == "memory_read":
        return governance.resolve(
            agent_id,
            proposal["namespace"],
            proposal["key"],
        ).value

    if proposal["kind"] == "memory_write":
        return governance.write(
            actor_id=agent_id,
            owner_id=proposal["owner_agent_id"],
            namespace=proposal["namespace"],
            key=proposal["key"],
            value=proposal["value"],
            expected_revision=proposal.get("expected_revision"),
        )

    raise AccessDenied("unsupported proposal kind")

parse_sop должен проверять схему до создания объекта: типы полей, обязательный version, уникальность идентификаторов, максимальное число шагов и допустимый размер аргументов. Разбор JSON не является авторизацией: синтаксически корректный документ всё ещё может содержать запрещённую операцию.

Не передавайте агенту прямую ссылку на драйвер базы, файловую систему или HTTP-клиент. Иначе он сможет обойти проверенный шлюз другим инструментом. В реестре FROST должны остаться только обёртки:

tools:
  - name: governed_memory_read
    handler: app.tools:memory_read
  - name: governed_memory_write
    handler: app.tools:memory_write
  - name: governed_sop_submit
    handler: app.tools:sop_submit

agents:
  report_manager:
    parent: null
    capabilities:
      - memory.read
      - memory.write
      - document.summarize
      - artifact.save
      - external.upload

  researcher:
    parent: report_manager
    capabilities:
      - memory.read
      - memory.write
      - document.summarize
      - artifact.save

  formatter:
    parent: researcher
    capabilities:
      - memory.read
      - artifact.save

Это иллюстративная конфигурация границы, а не обещание точного синтаксиса вашей версии FROST. Перенесите поля в её фактический формат, не меняя семантику пересечения прав.

Почему SOP нужно валидировать целиком

Рассмотрим процесс из трёх шагов:

  1. прочитать внутренний документ;
  2. создать краткое резюме;
  3. отправить документ и резюме во внешний сервис.

Если проверять шаг непосредственно перед исполнением, первые два действия произойдут до отказа на третьем. Иногда это допустимо, но тогда система уже создала промежуточные артефакты, изменила память или раскрыла содержание модели с другим контуром обработки.

Поэтому прототип сначала выполняет статическую проверку всех шагов. Для процессов с динамическими ветвлениями используйте двухуровневую схему:

  • до старта проверьте граф, типы шагов и верхнюю границу разрешений;
  • перед каждым фактическим действием повторите проверку конкретных аргументов и актуальной версии политики.

Предварительная проверка не заменяет runtime-контроль. Между планированием и выполнением могла измениться политика, владелец ресурса или классификация данных.

Шаг 6. Расширяем негативную матрицу

Базовый набор стоит дополнить случаями, которые часто обходят поверхностные проверки.

Сценарий Ожидаемый результат Что проверять после отказа
Потомок указывает себя владельцем родительской записи Создаётся отдельная локальная запись либо запрос отклоняется по политике namespace Родительская запись не изменилась
Потомок передаёт owner_agent_id брата Отказ actor_is_not_owner В памяти брата нет изменений
Приватная запись существует у ближайшего предка Отказ, без поиска одноимённой записи выше Закрытое значение не попало в ответ
SOP содержит неизвестное поле действия Ошибка схемы Ни один шаг не запущен
SOP превышает лимит шагов Ошибка до авторизации Нет промежуточных эффектов
Политика меняется между проверкой и выполнением Повторная проверка отклоняет действие Исполнитель не получил команду
Повтор одного разрешённого SOP Поведение определяется ключом операции Нет непредусмотренных дублей

Для внешних эффектов добавьте идемпотентность: стабильный operation_id, уникальное ограничение в хранилище и возврат прежнего результата при повторе. Права доступа и защита от дублей решают разные задачи.

Журнал решений как часть доказательства

Журнал аудита должен отвечать на вопросы «кто», «что», «над каким ресурсом», «по какой версии политики» и «почему разрешено или запрещено».

Для рабочего контура расширьте Decision такими полями:

{
  "event_id": "uuid",
  "timestamp": "UTC timestamp",
  "trace_id": "one agent run",
  "actor_agent_id": "researcher",
  "actor_lineage": ["researcher", "report_manager"],
  "operation": "memory.write",
  "resource": "report_manager:report.retention_days",
  "allowed": false,
  "reason_code": "actor_is_not_owner",
  "policy_version": "governance-7",
  "sop_id": "rewrite-policy",
  "sop_version": 1,
  "arguments_digest": "sha256:...",
  "previous_revision": 1
}

Чувствительные аргументы в журнале лучше заменять хешем или безопасным описанием. Нельзя исправлять утечку памяти, копируя секрет целиком в лог отказа.

Причина должна быть стабильным кодом, а человекочитаемое сообщение — отдельным полем. Тогда тесты и мониторинг не сломаются после редакторского изменения текста.

Что обычно ломается

Правило существует только в промпте

Фраза «никогда не меняй память родителя» полезна для планирования, но не является границей безопасности. Prompt injection, ошибка модели или другой системный промпт могут изменить предложенное действие. Хранилище всё равно обязано проверить владельца.

Память наследуется копированием

Если при создании потомка родительские записи копируются в его словарь без метаданных, появляются две конкурирующие версии. Нельзя определить, является ли изменение локальным переопределением или попыткой изменить общий факт.

Разрешение родителя автоматически становится разрешением потомка

Это превращает делегирование в эскалацию полномочий. У каждого потомка должен быть собственный декларативный набор, а итог вычисляется пересечением по всей цепочке.

Валидатор разрешает неизвестное действие

Правило «если действие не упомянуто, пропустить» опасно при обновлении FROST или подключении нового инструмента. Неизвестная операция должна перейти в отказ или ручное подтверждение.

Проверяется имя инструмента, но не аргументы

artifact.save может быть допустимым для draft/report.md и недопустимым для системного пути. После проверки возможности нужна проверка ресурса: namespace, пути, адресата, размера и классификации данных.

Тест утверждает только наличие исключения

Исключение могло возникнуть после записи. Негативный тест обязан повторно прочитать состояние, проверить ревизию и убедиться, что исполнитель внешнего действия не вызывался.

Проверка и действие разделены слишком большой паузой

Это создаёт ошибку типа time-of-check/time-of-use. Перед записью используйте условное обновление по ревизии, а перед внешним вызовом — короткоживущий подписанный допуск с привязкой к аргументам.

Что добавить перед рабочим использованием

  1. Версионирование политик. Каждое решение и запуск SOP должны ссылаться на точную версию набора правил.
  2. Политику namespace. Например, потомок может писать только в notes.* и artifacts.*, даже в собственной области.
  3. Контроль типов. Срок хранения должен быть целым числом в допустимом диапазоне, а не произвольной строкой.
  4. Неизменяемый журнал. Отказы и разрешения записываются append-only и не редактируются самим агентом.
  5. Ограничение ресурсов. Максимальная глубина иерархии, число потомков, шагов SOP, повторов и длительность запуска.
  6. Повторную авторизацию. Долгий SOP проверяет актуальную политику перед каждым эффектом.
  7. Разделение ролей. Компонент, предлагающий операцию, не должен сам выдавать себе разрешение.
  8. Изоляцию инструментов. Непроверенные HTTP-, shell- и файловые операции не регистрируются в runtime агента.
  9. Тесты миграций. Обновление схемы памяти не должно удалять владельца, видимость или ревизию.
  10. Аварийный запрет. Общая политика окружения должна уметь немедленно убрать возможность у всей иерархии.

Итоговая проверка стенда

Работу можно считать воспроизводимой, если другой инженер способен выполнить следующие действия без доступа к вашей модели или секретам:

  • создать окружение и установить только pytest;
  • запустить тесты одной командой;
  • увидеть, от какого предка унаследована разрешённая запись;
  • подтвердить отказ при чтении приватной записи;
  • подтвердить неизменность данных предка после запрещённой записи;
  • подтвердить сужение возможностей на каждом уровне;
  • подтвердить отклонение SOP с внешней загрузкой;
  • подтвердить отклонение неизвестного типа шага;
  • сопоставить каждый отказ с записью журнала;
  • заменить адаптер FROST, не переписывая правила и тестовые ожидания.

Отдельно проведите интеграционный тест с фальшивым исполнителем:

class SpyExecutor:
    def __init__(self):
        self.calls = []

    async def execute(self, action, arguments):
        self.calls.append((action, arguments))
        return {"status": "ok"}

После отклонения SOP список calls должен оставаться пустым. Такой spy доказывает, что валидатор остановил процесс до границы эффекта.

Ограничения прототипа

Учебная реализация хранит состояние в памяти одного процесса. Она не обеспечивает транзакции между несколькими worker, долговечность журнала, криптографическую целостность решений и распределённую блокировку.

Прототип также не анализирует смысл документов. Разрешение external.upload может зависеть не только от агента и адресата, но и от классификации конкретного файла. Для этого нужен отдельный атрибут ресурса и политика уровня объекта.

Проверка SOP работает с линейным списком шагов. Для графа с условиями, циклами и подпрограммами потребуется:

  • проверка достижимости запрещённых узлов;
  • лимиты циклов и общей стоимости;
  • валидация входов и выходов каждой ветви;
  • правила передачи данных между шагами;
  • повторная проверка динамически созданных подзадач.

Наконец, тесты подтверждают свойства приведённого governance-слоя, а не любой установки FROST автоматически. После подключения необходимо повторить те же сценарии через реальные точки входа runtime и убедиться, что обходных инструментов нет.

Практический вывод

Безопасное наследование — это не копирование промпта родителя. Оно требует явной модели происхождения памяти, монотонного сужения возможностей и обязательного шлюза перед каждым эффектом.

Минимальный надёжный контракт выглядит так: потомок может читать только видимые записи, писать только в собственную область, получать не больше прав, чем есть у всей цепочки предков, а SOP проходит полную предварительную проверку и повторную проверку перед выполнением. Тесты должны доказывать отсутствие эффекта, а не только наличие корректного сообщения об отказе.

Следующий разумный шаг — заменить словарь памяти транзакционным хранилищем, добавить версию политики в журнал и прогнать негативную матрицу через настоящий адаптер FROST.

← Все практические гайды · Лабораторный глоссарий →