Контроль действий ИИ-агента через журнал событий

Почему истории диалога недостаточно

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

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

1. агент предлагает изменение;
2. контур разрешает запись;
3. агент сообщает о завершении записи;
4. контур требует проверку;
5. только успешная проверка разрешает завершение задачи.

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

Модель состояний

Используем четыре состояния задачи и небольшой набор команд:

Каждая запись увеличивает ревизию. Успешная проверка относится к конкретной ревизии, поэтому новое изменение автоматически делает старую проверку недостаточной. В нашем минимальном автомате повторное изменение начинается новой задачей; расширение для нескольких файлов рассмотрим в ограничениях.

Шаг 1. Создаем управляющий контур

Нужен Python 3 без сторонних пакетов. Создайте отдельный пустой каталог и сохраните следующий файл как controller.py. Код является учебным примером, а не готовым механизмом изоляции.

#!/usr/bin/env python3
import argparse
import hashlib
import json
import os
from pathlib import Path
from typing import Any

LOG_PATH = Path("agent-events.jsonl")
GENESIS = "0" * 64

def canonical(data: dict[str, Any]) -> bytes:
    return json.dumps(
        data, ensure_ascii=False, sort_keys=True, separators=(",", ":")
    ).encode("utf-8")

def digest(event_without_hash: dict[str, Any]) -> str:
    return hashlib.sha256(canonical(event_without_hash)).hexdigest()

def initial_state() -> dict[str, Any]:
    return {
        "status": "idle",
        "path": None,
        "revision": 0,
        "verified_revision": None
    }

def reduce_event(state: dict[str, Any], event: dict[str, Any]) -> dict[str, Any]:
    kind = event["type"]
    data = event["data"]
    next_state = dict(state)

    if kind == "change_proposed":
        next_state.update(status="planned", path=data["path"])
    elif kind == "change_applied":
        next_state.update(
            status="changed",
            revision=state["revision"] + 1,
            verified_revision=None
        )
    elif kind == "check_recorded":
        if data["ok"]:
            next_state.update(
                status="verified",
                verified_revision=state["revision"]
            )
        else:
            next_state.update(
                status="changed",
                verified_revision=None
            )
    elif kind == "task_finished":
        next_state["status"] = "finished"
    else:
        raise ValueError(f"unknown event type: {kind}")

    return next_state

def load() -> tuple[dict[str, Any], str, int]:
    state = initial_state()
    previous_hash = GENESIS
    sequence = 0

    if not LOG_PATH.exists():
        return state, previous_hash, sequence

    with LOG_PATH.open("r", encoding="utf-8") as stream:
        for line_number, line in enumerate(stream, 1):
            event = json.loads(line)
            supplied_hash = event.pop("hash")

            if event["seq"] != sequence + 1:
                raise ValueError(f"broken sequence at line {line_number}")
            if event["prev_hash"] != previous_hash:
                raise ValueError(f"broken hash chain at line {line_number}")
            if digest(event) != supplied_hash:
                raise ValueError(f"invalid event hash at line {line_number}")

            state = reduce_event(state, event)
            previous_hash = supplied_hash
            sequence = event["seq"]

    return state, previous_hash, sequence

def append_event(kind: str, data: dict[str, Any]) -> None:
    state, previous_hash, sequence = load()
    event = {
        "seq": sequence + 1,
        "type": kind,
        "data": data,
        "prev_hash": previous_hash
    }
    event["hash"] = digest(event)

    flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND
    descriptor = os.open(LOG_PATH, flags, 0o600)
    try:
        payload = canonical(event) + b"\n"
        os.write(descriptor, payload)
        os.fsync(descriptor)
    finally:
        os.close(descriptor)

    new_state = reduce_event(state, event)
    print(json.dumps(new_state, ensure_ascii=False, indent=2))

def decide(command: str, args: argparse.Namespace) -> tuple[str, dict[str, Any]]:
    state, _, _ = load()
    status = state["status"]

    if command == "propose_change" and status == "idle":
        path = args.path.strip()
        if not path or Path(path).is_absolute() or ".." in Path(path).parts:
            raise ValueError("path must be a safe relative path")
        return "change_proposed", {"path": path}

    if command == "apply_change" and status == "planned":
        if args.path != state["path"]:
            raise ValueError("path differs from approved proposal")
        return "change_applied", {"path": args.path}

    if command == "record_check" and status == "changed":
        return "check_recorded", {
            "name": args.name,
            "ok": args.result == "pass"
        }

    if command == "finish" and status == "verified":
        if state["verified_revision"] != state["revision"]:
            raise ValueError("current revision is not verified")
        return "task_finished", {}

    raise ValueError(f"transition {status} -> {command} is forbidden")

def main() -> None:
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="command", required=True)

    propose = subparsers.add_parser("propose_change")
    propose.add_argument("--path", required=True)

    apply_change = subparsers.add_parser("apply_change")
    apply_change.add_argument("--path", required=True)

    check = subparsers.add_parser("record_check")
    check.add_argument("--name", required=True)
    check.add_argument("--result", choices=("pass", "fail"), required=True)

    subparsers.add_parser("finish")
    subparsers.add_parser("status")

    args = parser.parse_args()

    try:
        if args.command == "status":
            state, last_hash, sequence = load()
            print(json.dumps({
                "state": state,
                "last_hash": last_hash,
                "events": sequence
            }, ensure_ascii=False, indent=2))
            return

        kind, data = decide(args.command, args)
        append_event(kind, data)
    except (ValueError, json.JSONDecodeError, KeyError) as error:
        parser.exit(2, f"rejected: {error}\n")

if __name__ == "__main__":
    main()

Управляющий контур разделен на три части:

• decide проверяет команду относительно восстановленного состояния;
• append_event добавляет принятое событие и принудительно передает запись операционной системе;
• load проверяет цепочку хешей и заново строит состояние функцией reduce_event.

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

Шаг 2. Воспроизводим разрешенный маршрут

Следующие команды работают только с файлами в текущем учебном каталоге. Они не запускают сетевые запросы, не требуют секретов и не изменяют реальный проект:

mkdir agent-control-demo
cd agent-control-demo
# Сохраните controller.py в этом каталоге.
python3 controller.py propose_change --path src/example.py
python3 controller.py apply_change --path src/example.py
python3 controller.py record_check --name syntax --result pass
python3 controller.py finish
python3 controller.py status

В финальном выводе ожидаются "status": "finished", "revision": 1, "verified_revision": 1 и "events": 4. Значение last_hash зависит от содержимого событий; сравнивать его с заранее заданной строкой не нужно.

Обратите внимание: команда apply_change в этом прототипе подтверждает факт изменения, но сама не редактирует файл. В реальной архитектуре именно доверенный исполнитель должен выполнить изменение и только после успешного завершения записать change_applied. Нельзя позволять агенту самостоятельно утверждать, что операция состоялась.

Шаг 3. Проверяем запрещенные переходы

Для чистого сценария используйте новый каталог с копией controller.py. Не удаляйте журнал рабочей системы ради теста.

Завершение без изменения

python3 controller.py finish

Контур должен завершиться с кодом 2 и сообщением transition idle -> finish is forbidden. Событие при этом не добавляется.

Подмена согласованного пути

python3 controller.py propose_change --path src/example.py
python3 controller.py apply_change --path src/other.py

Вторая команда должна быть отклонена сообщением path differs from approved proposal. Состояние останется planned.

Обход обязательной проверки

python3 controller.py apply_change --path src/example.py
python3 controller.py finish

Вторая команда должна быть отклонена: после применения изменения состояние равно changed, а переход к finish разрешен только из verified.

Неуспешная проверка

python3 controller.py record_check --name syntax --result fail
python3 controller.py finish

Неуспешная проверка сохраняется как факт, но состояние остается changed. Завершение снова отклоняется.

Шаг 4. Восстанавливаем состояние из журнала

После разрешенного маршрута запустите:

python3 controller.py status

Процесс стартует без памяти о предыдущих командах. Функция load читает agent-events.jsonl, проверяет номера и хеш-цепочку, а затем применяет события по порядку. Полученное состояние и есть восстановление агента в рамках управляющего автомата.

Посмотреть структурные поля журнала можно безопасной командой:

python3 -m json.tool --json-lines agent-events.jsonl

Каждая строка содержит:

• seq — монотонный номер события;
• type и data — принятый факт и его параметры;
• prev_hash — хеш предыдущей записи;
• hash — контрольное значение текущей записи.

Как встроить контур вокруг реального агента

Схема интеграции должна отделять предложение от исполнения:

1. агент отправляет структурированную команду контроллеру;
2. контроллер восстанавливает состояние и проверяет переход;
3. доверенный исполнитель применяет разрешенное изменение;
4. контроллер записывает событие с фактическим результатом;
5. проверки запускаются отдельным доверенным процессом;
6. их код завершения преобразуется в событие, а не пересказывается агентом.

Минимальный контракт команды может выглядеть так:

{
  "task_id": "local-example",
  "command": "record_check",
  "arguments": {
    "name": "syntax",
    "observed_exit_code": 0
  },
  "expected_revision": 1
}

Это пример формата. Поле expected_revision позволяет отклонять устаревшие команды при конкуренции. Поле observed_exit_code должен заполнять исполнитель проверки, а не модель.

Для производственного варианта добавьте идемпотентный command_id, идентификатор задачи, автора решения, версию политики и хеш входных артефактов. Событие должно отвечать на вопрос «что надежно произошло», а не «что агент собирался сделать».

Проверка результата

Прототип выполняет заявленную задачу, если вручную подтверждены все пункты:

• разрешенный маршрут создает четыре события и состояние finished;
• finish до успешной проверки возвращает код 2;
• отклоненная команда не увеличивает счетчик событий;
• новый процесс восстанавливает то же состояние командой status;
• нарушение последовательности или хеш-цепочки останавливает загрузку журнала;
• абсолютный путь и путь с компонентом .. отклоняются на этапе предложения.

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

Типовые ошибки

Записывать только успешные решения

Такой журнал объясняет состояние, но не показывает попытки обхода. Отказы полезно сохранять в отдельном аудиторском потоке с причиной и идентификатором команды. Не смешивайте их с доменными событиями, иначе reducer может принять отказ за изменение состояния.

Доверять полю result: pass от агента

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

Сначала выполнить, потом спросить политику

Проверка после изменения годится для аудита, но не для контроля. Решение о допустимости пути, инструмента и перехода принимается до побочного эффекта.

Считать O_APPEND полной защитой

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

Игнорировать частичную запись

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

Не учитывать параллельные команды

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

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

• Журнал локальный и не защищен от владельца файла.
• Один вызов os.write не является универсальной гарантией атомарности для файлов любого размера и любой файловой системы.
• Нет блокировки нескольких процессов, дедупликации команд и привязки событий к отдельным задачам.
• Контур фиксирует подтверждение изменения, но не выполняет его и не сверяет содержимое файла.
• Хеш SHA-256 здесь служит контролем целостности цепочки, а не подписью автора.
• Снимки состояния не реализованы; восстановление занимает время, пропорциональное числу событий.
• Политика допускает один путь и одну ревизию. Пакетные изменения потребуют явной модели change set.

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

Итог

Журнал диалога отвечает на вопрос, что модель сказала. Управляющий журнал отвечает на более строгий вопрос: какие переходы доверенный контур разрешил и какие факты наблюдал исполнитель. В прототипе запрещенное завершение отклоняется до записи события, обязательная проверка привязана к ревизии, а состояние полностью восстанавливается повторным применением append-only последовательности.

Для продолжения изучите практические руководства Agent Lab Journal и сверяйтесь с глоссарием терминов при проектировании политики агента.