Сбой и неполный результат

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

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

Ситуация

Журнал ротировали, пока агент работал: файла app.log нет. Инструмент поиска, написанный «просто», вернёт пустой список — и агент честно напишет: «Ошибок в журнале нет». Это ложь, построенная из пустоты. В курсе «Агент чинит сервер» такие утверждения без опоры вы ловили в выводе агента; здесь вы не даёте серверу их порождать.

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

У инструмента три разных исхода, и модель должна их различать:

Ожидаемая ошибка и падение

Переименуйте журнал в учебной папке и вызовите log_tail сервером после главы 4:

mv ~/mcp-lab/app/logs/app.log ~/mcp-lab/app/logs/app.log.moved

Клиент получит только Error executing tool log_tail. Для SDK это падение: функция бросила неожиданное исключение (FileNotFoundError). Модель не узнает причину, а трассировка уйдёт в stderr сервера. Верните файл: mv ~/mcp-lab/app/logs/app.log.moved ~/mcp-lab/app/logs/app.log.

Ошибку, которую вы предвидели, бросайте как ToolError со смыслом:

def open_log():
    try:
        return open(LOG, "rb")
    except FileNotFoundError:
        raise ToolError("журнал app.log не найден; возможно, его переименовали или ротировали") from None
    except PermissionError:
        raise ToolError("нет прав на чтение app.log") from None

Теперь модель прочитает: Error executing tool log_tail: журнал app.log не найден; возможно, его переименовали или ротировали. Это повод для следующего наблюдения, а не для вывода «ошибок нет».

Плохие байты

Последняя строка учебного журнала содержит байты, которые не являются UTF-8. Чтение с errors="replace" не падает, а ставит �. Считайте замены и отдавайте число в ответе: replaced_bytes: 2. Модель увидит, что текст строки искажён, а не что пользователь прислал два странных символа.

def decode(raw: bytes) -> tuple[str, int]:
    text = raw.decode("utf-8", errors="replace")
    return text.rstrip("\n"), text.count("�")

Предел — на чтении, а не после него

Журнал может быть огромным, а одна строка — длиной в десятки мегабайт (дамп, бинарный мусор). Предел, который проверяется после чтения, не защищает: for raw in f: сначала выделит всю строку, а потом вы сравните её длину с бюджетом. Ограничивайте сами операции чтения:

• f.readline(n) читает не больше n байт, даже если строка длиннее; остаток строки дочитывается следующими вызовами. Возьмите кусок CHUNK = 64 * 1024.
• Перед каждым чтением считайте, сколько осталось от бюджета, и не просите больше: f.readline(min(CHUNK, room)).
• Подстроку ищите в каждом куске, сохраняя хвост предыдущего куска длиной len(искомое) − 1 байт, — иначе совпадение на стыке двух кусков потеряется. Искомую строку для этого переведите в байты (text.encode()).
• Для ответа хватит начала строки (первый кусок), обрезанного до MAX_LINE; если строка длиннее куска — cut: true.

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

Бюджет и неполный ответ

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

• limit совпадений — stop_reason: "limit";
• больше SEARCH_MS миллисекунд — "time_budget";
• больше SEARCH_BYTES прочитанных байт — "size_budget". Если бюджет кончился ровно на конце файла, это не превышение: попробуйте прочитать ещё байт, и если файл кончился — результат полный.

В ответе: scanned_lines — сколько строк просмотрено до конца, complete — дошли ли до конца файла, stop_reason — почему нет. Номер строки в поиске честный: поиск идёт с начала файла и считает строки по дороге.

Для проверок пределы задаются переменными окружения (MCP_LAB_SEARCH_MS, MCP_LAB_SEARCH_BYTES): check.py запускает сервер с бюджетом 0 мс и 1000 байт и ждёт complete: false с правильной причиной.

Хвост файла и номера строк

log_tail нужен конец журнала. Прочитайте не больше TAIL_BYTES = 256 * 1024 байт с конца — и ничего до них.

Отсюда честное ограничение: номер строки от начала файла неизвестен, если окно не охватило весь файл. Чтобы узнать номер, пришлось бы прочитать весь префикс (или вести индекс) — а мы обещали читать только хвост. Поэтому в модели строки line_no: int | None, а в результате whole_file: bool: номера есть, только когда окно — весь файл (учебный журнал около 250 КБ в окно помещается).

Первая строка окна — обычно обрывок. Но не всегда: если байт прямо перед окном — перевод строки, окно начинается с целой строки, и выбрасывать её нельзя. Прочитайте окно на один байт раньше и посмотрите на этот байт. Обрывок отбросьте и скажите об этом полем dropped_partial: true.

log_levels по-прежнему считает все строки — это честно стоит одного прохода по файлу. Но и он читает кусками: память не растёт с длиной строки, а время растёт с размером журнала. Об этом сказано в описании инструмента.

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

1. Добавьте open_log() и используйте его во всех инструментах журнала. Ни один из них не должен возвращать пустой результат, если журнала нет.
2. Добавьте decode() и поле replaced_bytes в результат log_tail.
3. Перепишите log_search: чтение кусками через readline(min(CHUNK, room)), совпадения на стыках, три причины остановки, поля scanned_lines, complete, stop_reason (Literal["limit", "time_budget", "size_budget"] | None). Пределы: SEARCH_BYTES = int(os.environ.get("MCP_LAB_SEARCH_BYTES", 5 * 1024 * 1024)), SEARCH_MS = int(os.environ.get("MCP_LAB_SEARCH_MS", 2000)).
4. Перепишите log_tail на чтение не больше TAIL_BYTES с конца: line_no — None, если окно не весь файл; поля whole_file и dropped_partial; граница окна по переводу строки.
5. Перепишите log_levels на чтение кусками.
6. Ответьте себе: агент получил complete: false, stop_reason: "limit" и 1 совпадение по запросу INFO. Можно ли ему написать «в журнале одна строка INFO»? Что он должен сделать, если ему нужно число?

Проверка

python check.py --chapter 5 diag_server.py

Ожидается Итог: выполнено 35 из 35 проверок, пройдено 35. Проверки главы (11): два заменённых байта, пустой поиск как полный результат, limit, отсутствие журнала для log_tail и log_search, бюджет времени и объёма — и четыре проверки объёма чтения. Для них check.py в отдельном процессе вызывает функции сервера напрямую и меряет выделения памяти Python (tracemalloc) — результат не зависит от скорости процессора:

• log_tail на разреженном файле 64 МиБ: память меньше 1 МиБ, последняя строка на месте, номер строки — None;
• log_search по журналу, где первая строка — 64 МиБ без перевода: память меньше 1 МиБ, stop_reason: size_budget;
• совпадение на стыке двух кусков внутри строки в 3 МиБ найдено, строка помечена cut;
• граница окна: строка сразу после перевода строки сохранена, обрывок отброшен с dropped_partial: true.

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

• «журнала нет — понятная ошибка» не проходит, а в ответе только Error executing tool log_tail — исключение не перехвачено как ToolError.
• «нет совпадений — полный пустой результат» не проходит — complete не true, когда журнал просмотрен до конца.
• Бюджет времени не срабатывает — проверка времени стоит после чтения строки, а не перед ней, или SEARCH_MS читается не из окружения.
• Проверки памяти не проходят, а остальные проходят — где-то файл или строка читается целиком: f.read() без предела, for raw in f:, read_text(), подсчёт номеров через чтение префикса.
• «граница окна» не проходит — первая строка окна выбрасывается всегда, без проверки байта перед окном.

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

Источники

• Пакет mcp 2.3.0, mcp/server/mcpserver/exceptions.py: ToolError — ожидаемая ошибка, текст уходит модели; любое другое исключение — падение, модель видит только Error executing tool <имя>.

TAIL_BYTES = 256 * 1024        # log_tail читает не больше этого с конца файла
CHUNK = 64 * 1024              # больше этого за одно чтение строки не берём
SEARCH_BYTES = int(os.environ.get("MCP_LAB_SEARCH_BYTES", 5 * 1024 * 1024))
SEARCH_MS = int(os.environ.get("MCP_LAB_SEARCH_MS", 2000))


class LogLine(BaseModel):
    line_no: int | None = Field(description="номер строки от начала файла; None — неизвестен (читался только хвост)")
    level: str
    text: str
    cut: bool = Field(description="строка обрезана до MAX_LINE символов")


class TailResult(BaseModel):
    lines: list[LogLine]
    whole_file: bool = Field(description="окно охватило весь файл: только тогда номера строк известны")
    dropped_partial: bool = Field(description="окно началось с середины строки, этот обрывок отброшен")
    replaced_bytes: int = Field(description="байт с неверной кодировкой, заменённых на �")


@mcp.tool()
def log_tail(
    lines: Annotated[int, Field(ge=1, le=200, description="сколько последних строк")] = 20,
    level: Level | None = None,
) -> TailResult:
    """Последние строки журнала (из последних 256 КиБ файла), при желании только одного уровня."""
    with open_log() as f:
        size = f.seek(0, os.SEEK_END)
        start = max(0, size - TAIL_BYTES)
        if start == 0:
            f.seek(0)
            buf, first_complete = f.read(size), True
        else:
            f.seek(start - 1)                   # на байт раньше окна: был ли там перевод строки
            buf = f.read(size - start + 1)
            first_complete, buf = buf[:1] == b"\n", buf[1:]
    parts = buf.split(b"\n")
    dropped = not first_complete
    if dropped:                                 # окно началось с середины строки — обрывок не показываем
        parts = parts[1:]
    whole = start == 0                          # номера строк знаем, только если видели весь файл
    replaced, out = 0, []
    for i, raw in enumerate(parts):
        if not raw:
            continue
        text, bad = decode(raw)
        replaced += bad
        line = parse_line(i + 1 if whole else None, text)
        if level is None or line.level == level:
            out.append(line)
    return TailResult(lines=out[-lines:], whole_file=whole, dropped_partial=dropped, replaced_bytes=replaced)


def line_heads(f):
    """Строки файла по одной: (начало строки — не больше CHUNK байт, длина строки). Память — O(CHUNK)."""
    while True:
        piece = f.readline(CHUNK)
        if not piece:
            return
        head, size = piece, len(piece)
        while not piece.endswith(b"\n"):
            piece = f.readline(CHUNK)
            if not piece:
                break
            size += len(piece)
        yield head, size


@mcp.tool()
def log_levels() -> dict[str, int]:
    """Сколько строк каждого уровня в журнале. Читает весь файл кусками; время растёт с размером журнала."""
    counts: dict[str, int] = {}
    with open_log() as f:
        for head, _ in line_heads(f):
            level = parse_line(None, decode(head)[0]).level
            counts[level] = counts.get(level, 0) + 1
    return counts


@mcp.tool()
def log_search(
    text: Annotated[str, Field(min_length=1, max_length=100, description="искомая подстрока, не регулярное выражение")],
    level: Level | None = None,
    limit: Annotated[int, Field(ge=1, le=50)] = 20,
) -> SearchResult:
    """Ищет подстроку в журнале. Возвращает не больше limit совпадений и честно сообщает, если просмотрел не всё."""
    needle = text.encode("utf-8")
    keep = len(needle) - 1                      # хвост куска, чтобы найти совпадение на стыке кусков
    started, scanned, read_bytes, no = time.monotonic(), 0, 0, 0
    matches: list[LogLine] = []
    stop = None
    with open_log() as f:
        while stop is None:
            head, size, found, carry, ended = b"", 0, False, b"", False
            while True:                         # одна строка, кусками не больше CHUNK
                if (time.monotonic() - started) * 1000 > SEARCH_MS:
                    stop = "time_budget"
                    break
                room = SEARCH_BYTES - read_bytes
                if room <= 0:                   # бюджет исчерпан; ровно конец файла — не превышение
                    if f.read(1):
                        stop = "size_budget"
                    else:
                        ended = True
                    break
                piece = f.readline(min(CHUNK, room))
                if not piece:
                    ended = True
                    break
                read_bytes += len(piece)
                head = head or piece
                size += len(piece)
                window = carry + piece
                found = found or needle in window
                carry = window[-keep:] if keep else b""
                if piece.endswith(b"\n"):
                    ended = True
                    break
            if size == 0:                       # конец файла или бюджет до начала строки
                break
            no += 1
            if ended:
                scanned = no
            line_text, _ = decode(head)
            line = parse_line(no, line_text.rstrip("\ufffd") if not ended else line_text)
            line.cut = line.cut or size > len(head)
            if found and (level is None or line.level == level):
                matches.append(line)
                if len(matches) == limit and stop is None:
                    stop = "limit"
    # summary — как в главе 3 (в главе 6 его не станет)
    return SearchResult(summary=summary, matches=matches, scanned_lines=scanned, complete=stop is None,
                        stop_reason=stop)