Изменение с подтверждением и защитой от повтора

Подготовка к этой главе

Нужны Linux, Python 3.12 и активное окружение из первого сервера. Продолжайте свой diag_server.py после предыдущей главы. Если начинаете здесь, скачайте исходное состояние — стадия 6, распакуйте в новую папку и создайте данные через python3 make_lab.py. Не перезаписывайте свою работу: сначала сохраните её отдельно.

Ситуация

Журнал разросся, его нужно ротировать: переименовать app.log в архив и начать новый. Инструмент rotate_log() одним вызовом опасен трижды:

• человек не видит заранее, что именно будет сделано;
• после обрыва связи хост может повторить вызов — и ротировать новый, почти пустой журнал;
• между «посмотрел» и «сделал» журнал меняется, и подтверждение относится уже к другому состоянию.

Главная мысль

Подтверждение человека — это решение о конкретном действии над конкретным состоянием. Сервер хранит план с отпечатком состояния и выполняет только его, один раз. Удобство SDK (декоратор, автоматическая схема, механизмы протокола для вопросов пользователю) не заменяет эти проверки: разрешение действовать проверяет ваш код.

План

log_rotate_plan() ничего не меняет. Он:

• проверяет, что журнал есть, и выбирает имя архива — первое свободное из app.log.2 … app.log.6; если все пять заняты, отказывает: удалять архивы инструмент не умеет, это решение человека;
• снимает отпечаток журнала: номер inode, размер, время изменения в наносекундах;
• создаёт случайный plan_id (secrets.token_hex(8) — 16 шестнадцатеричных символов) и сохраняет план в app/state/plans.json с временем создания;
• возвращает: plan_id, размер журнала, имя архива, словами — что будет сделано, сколько секунд живёт план.

Хост показывает это человеку. Человек соглашается с этим планом.

Выполнение

log_rotate_apply(plan_id) принимает только plan_id (схема: ^[0-9a-f]{16}$) и по порядку:

1. Если план уже выполнен — возвращает прежний результат со статусом already_applied и ничего не делает.
2. План не найден — отказ. План старше PLAN_TTL (600 с, переменная MCP_LAB_PLAN_TTL для проверок) — отказ.
3. Отпечаток журнала сейчас не равен отпечатку в плане — отказ: «журнал изменился после подготовки плана; подтверждение относилось к другому состоянию».
4. Архив с таким именем уже есть — отказ.
5. Переименовывает журнал, создаёт пустой, записывает результат в app/state/applied.json.

Файлы состояния пишите через временный файл и replace() — замена в пределах одной папки атомарна, и при сбое посередине не останется полузаписанного JSON.

Одновременные вызовы

Два вызова с одним plan_id могут прийти одновременно — вы видели в главе 2, что сервер обрабатывает запросы параллельно. Оба пройдут проверку «ещё не выполнен» и оба переименуют журнал. Нужна блокировка: файл app/state/apply.lock, создаваемый с os.O_CREAT | os.O_EXCL (создание проваливается, если файл есть), и удаление после выполнения. Второй вызов получит «другая ротация выполняется прямо сейчас» — честный отказ вместо второй ротации.

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

Как это проверить, не надеясь на удачу? Если просто запустить два вызова одновременно, они почти всегда пройдут по очереди, и сервер без блокировки «пройдёт» проверку. check.py поэтому управляет порядком сам: в отдельном процессе он импортирует ваш сервер, запускает выполнение плана из двух потоков и останавливает первый вызов перед его первым переименованием файла — то есть после всех проверок и чтения состояния. Пока первый стоит, второй вызов работает до конца; потом первый отпускается. Ожидание ограничено событиями с пределом времени, случайных задержек нет.

Без блокировки результат такой (воспроизведено): оба вызова отвечают applied, а второе переименование кладёт новый пустой журнал на место архива — архив потерян. С блокировкой второй вызов получает отказ занятости, архив один и того же размера, а повтор плана возвращает исходный результат. Проверка принимает и другие правильные способы: если второй вызов дождался первого и вернул already_applied с тем же архивом — это тоже верно.

Сбой посередине

Если сервер переименовал журнал и упал до записи в applied.json, повтор плана не ротирует второй раз: отпечаток нового пустого журнала другой, и проверка 3 откажет. Человек увидит архив и разберётся. Защита от повтора держится не на одном механизме, а на двух.

А как же подтверждение в протоколе?

Ревизия 2026-07-28 позволяет серверу запросить у пользователя данные или подтверждение посреди вызова (elicitation через шаблон Multi Round-Trip Requests: сервер возвращает input_required, клиент повторяет запрос с ответом пользователя). Это удобный канал, но не замена плану: ответ пользователя всё равно нужно привязать к состоянию и защитить от повтора, а наш тестовый клиент без человека такой вопрос задать не может. В курсе подтверждение — ответственность хоста: он показывает план и вызывает выполнение после «да».

Самостоятельно

1. Напишите log_rotate_plan и log_rotate_apply по описанию выше. Модели результата: RotatePlan (plan_id, log_size, archive_name, action, expires_in_s) и RotateResult (plan_id, status, archive_name, archived_bytes).
2. Обновите инструкции сервера: теперь он не только читает.
3. Добавьте в свой клиент сценарий «обрыв связи»: план → выполнение → то же выполнение ещё раз; убедитесь, что второй ответ already_applied, а архив один.
4. Ответьте себе: почему plan_id случайный, а не, например, номер по порядку? Что было бы, если бы выполнение принимало archive_name вместо plan_id?

Проверка

python check.py --chapter 7 diag_server.py

Ожидается Итог: выполнено 50 из 50 проверок, пройдено 50. Проверки главы (8): план ничего не меняет; выполнение даёт архив того же размера и пустой журнал; повтор — already_applied без второго архива; журнал изменился после плана — отказ; неизвестный план — отказ; управляемая гонка — одна ротация, второй вызов — отказ или прежний результат, архив цел; повтор после гонки — полный исходный результат; просроченный план — отказ.

Если результат другой

• «управляемая гонка» не проходит — нет блокировки, она берётся после чтения состояния или снимается до записи результата. В подробностях проверки видны оба ответа и размеры архивов.
• «журнал изменился после плана» не проходит — отпечаток не включает размер или время изменения.
• «просроченный план» не проходит — PLAN_TTL не читается из MCP_LAB_PLAN_TTL.

Решение — answers/07-confirmed-change.md; полный файл — project/stages/07/diag_server.py.

Источники

• Изменения ревизии 2026-07-28: Multi Round-Trip Requests, InputRequiredResult, elicitation.
• Архитектура MCP: elicitation как способ запросить у пользователя подтверждение.
• Документация Python: os.open с O_CREAT | O_EXCL, Path.replace, secrets.token_hex.

def fingerprint() -> dict:
    st = LOG.stat()
    return {"inode": st.st_ino, "size": st.st_size, "mtime_ns": st.st_mtime_ns}


@mcp.tool()
def log_rotate_plan() -> RotatePlan:
    """Готовит план ротации журнала. Ничего не меняет. Выполнить план можно только после подтверждения человеком."""
    if not LOG.exists():
        raise ToolError("журнал app.log не найден")
    archive = next_archive()
    plan_id = secrets.token_hex(8)
    plans = load("plans.json")
    plans[plan_id] = {"fingerprint": fingerprint(), "archive": archive, "created": time.time()}
    save("plans.json", plans)
    size = plans[plan_id]["fingerprint"]["size"]
    return RotatePlan(plan_id=plan_id, log_size=size, archive_name=archive,
                      action=f"переименовать app.log ({size} байт) в {archive} и создать пустой app.log",
                      expires_in_s=PLAN_TTL)


@mcp.tool()
def log_rotate_apply(plan_id: Annotated[str, Field(pattern=r"^[0-9a-f]{16}$")]) -> RotateResult:
    """Выполняет подтверждённый план ротации. Повторный вызов с тем же plan_id не ротирует второй раз."""
    with StateLock():
        return apply_plan(plan_id)


def apply_plan(plan_id: str) -> RotateResult:
    applied = load("applied.json")
    if plan_id in applied:                    # защита от повтора: тот же ответ, без второго действия
        return RotateResult(**applied[plan_id], status="already_applied")
    plan = load("plans.json").get(plan_id)
    if plan is None:
        raise ToolError("план не найден; подготовьте новый через log_rotate_plan")
    if time.time() - plan["created"] > PLAN_TTL:
        raise ToolError("план устарел; подготовьте новый и подтвердите его заново")
    if not LOG.exists() or fingerprint() != plan["fingerprint"]:
        raise ToolError("журнал изменился после подготовки плана; подтверждение относилось к другому состоянию")
    archive = LOGS / plan["archive"]
    if archive.exists():
        raise ToolError(f"{plan['archive']} уже существует; подготовьте новый план")
    size = plan["fingerprint"]["size"]
    os.rename(LOG, archive)
    LOG.touch()
    record = {"plan_id": plan_id, "archive_name": plan["archive"], "archived_bytes": size}
    applied[plan_id] = record
    save("applied.json", applied)
    return RotateResult(**record, status="applied")

        plan = (await client.call_tool("log_rotate_plan")).structured_content
        first = await client.call_tool("log_rotate_apply", {"plan_id": plan["plan_id"]})
        again = await client.call_tool("log_rotate_apply", {"plan_id": plan["plan_id"]})
        print(first.structured_content["status"], again.structured_content["status"])   # applied already_applied