scripts/agent_log.py - dev/test - ump-team/ump-infra

scripts/agent_log.py
482 строки21 KB
Dmitry Kochenov
fix: исправить 46 ошибок basedpyright и добавить type hint stubs
09 авг 2026, 15:25
09 авг 2026, 15:258a46eb7
#!/usr/bin/env python3
"""
Логгер всех изменений и ошибок AI-агента.

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

Запуск (из агента или вручную):
    uv run python scripts/agent_log.py add "Создан файл X" --type action
    uv run python scripts/agent_log.py add "Phase 2 failed" --type error --step 1.6
    uv run python scripts/agent_log.py add "Manual change detected" --type manual_change
    uv run python scripts/agent_log.py tail --lines 50
    uv run python scripts/agent_log.py show --step 1.6
    uv run python scripts/agent_log.py clear --before 2026-07-09

Лог-файл: _meta/agent.log (JSONL формат — по одной записи на строку).
"""
from __future__ import annotations

import argparse
import contextlib
import json
import os
import re
import socket
import sys
from datetime import datetime
from pathlib import Path

# Единая точка правды для BASE-резолвинга (корректна и в submodule-, и в standalone-раскладке).
try:
    from ump.paths import detect_base
except ImportError:
    _here = Path(__file__).resolve().parent
    sys.path.insert(0, str(_here.parent))
    from ump.paths import detect_base  # type: ignore[no-redef]

BASE = detect_base(__file__)
LOG_FILE = BASE / '_meta' / 'agent.log'

# P1-fix: версия логгера — единый источник правды (см. README.md, CHANGELOG.md).
# Ранее было hardcoded 'v0.0.1' в каждой записи, что рассинхронизировалось
# с версией пакета. Теперь — константа на уровне модуля; при релизе правится
# в одном месте.
# v0.0.6-audit: fix ERR-VR-001 (9→8 гейтов в 10-stop-conditions.md) + MD validation fixes.
UMP_VERSION = 'v0.0.6'

# P2-6 fix: автоматическая ротация лога.
# Ранее clear --before DATE была только ручной — без автоматики лог мог расти
# без ограничений в долгих сессиях агента. Теперь при превышении MAX_LOG_SIZE_BYTES
# лог автоматически ротируется (agent.log → agent.log.1 → ... → agent.log.5).
# Конфигурируется через env vars UMP_LOG_MAX_SIZE_MB и UMP_LOG_MAX_FILES.
MAX_LOG_SIZE_BYTES = int(os.environ.get('UMP_LOG_MAX_SIZE_MB', '50')) * 1024 * 1024
MAX_LOG_FILES = int(os.environ.get('UMP_LOG_MAX_FILES', '5'))


def _rotate_if_needed(log_file: Path | None = None) -> bool:
    """Если размер лога превышает MAX_LOG_SIZE_BYTES — ротировать.

    Алгоритм:
      1. agent.log.<N-1> → agent.log.<N> (если N >= MAX_LOG_FILES, удаляем)
      2. agent.log.1 → agent.log.2
      3. agent.log → agent.log.1

    Returns:
        True если ротация была выполнена, False если не нужна.
    """
    target = log_file or LOG_FILE
    if not target.exists():
        return False
    try:
        size = target.stat().st_size
    except OSError:
        return False
    if size < MAX_LOG_SIZE_BYTES:
        return False

    # Сдвигаем agent.log.<N-1> → agent.log.<N>, начиная с MAX_LOG_FILES-1 вниз
    for i in range(MAX_LOG_FILES - 1, 0, -1):
        src = target.with_suffix(target.suffix + f'.{i}')
        dst = target.with_suffix(target.suffix + f'.{i + 1}')
        if src.exists():
            if i + 1 >= MAX_LOG_FILES:
                # Удаляем самый старый
                with contextlib.suppress(OSError):
                    src.unlink()
            else:
                try:
                    if dst.exists():
                        dst.unlink()
                    src.rename(dst)
                except OSError:
                    pass
    # agent.log → agent.log.1
    rotated = target.with_suffix(target.suffix + '.1')
    try:
        if rotated.exists():
            rotated.unlink()
        target.rename(rotated)
    except OSError:
        return False
    return True

# P0-fix: маскирование секретов в details. См. правило 13-logging-v40.md §"Структура записи".
# Регэксп ловит ключи, имена которых указывают на секреты (token, password, secret, key, ...).
# Значение заменяется на '***REDACTED***'. Это regression-guard: токены/пароли
# не должны попадать в _meta/agent.log в открытом виде.
_SENSITIVE_KEY_RE = re.compile(
    r'(?:^|_)(token|password|passwd|secret|api_?key|access_?key|private_?key|'
    r'client_?secret|refresh_?token|auth|credential|bearer|ump_?protect_?key)(?:$|_)',
    re.IGNORECASE,
)
_REDACTED = '***REDACTED***'


def _redact_details(details: dict[str, object] | None) -> dict[str, object] | None:
    """Вернуть копию details с замаскированными секретами.

    Рекурсивно обходит вложенные dict. Списки значений-секретов не маскируются
    (намеренно — иначе можно случайно скрыть нужную отладочную информацию).
    """
    if not details:
        return details
    redacted: dict[str, object] = {}
    for k, v in details.items():
        if isinstance(k, str) and _SENSITIVE_KEY_RE.search(k):
            redacted[k] = _REDACTED
        elif isinstance(v, dict):
            redacted[k] = _redact_details(v)
        else:
            redacted[k] = v
    return redacted


def _atomic_append(path: Path, line: str) -> None:
    """Атомарно дописать строку в JSONL-лог с файловой блокировкой.

    P0-fix: ранее использовался обычный open(...,'a') + write, что приводило
    к перемешиванию строк при параллельной записи из нескольких субагентов
    (см. .agent/rules/18-parallel-subagents.md). Теперь — fcntl.flock на время
    write+flush+fsync, что гарантирует целостность каждой JSONL-записи.

    На платформах без fcntl (Windows) fallback на открытие в режиме 'a'
    без блокировки — корректность не гарантируется для параллельных процессов,
    но одиночная запись работает идентично старому поведению.
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    try:
        import fcntl
        with path.open('a', encoding='utf-8') as f:
            fcntl.flock(f.fileno(), fcntl.LOCK_EX)
            try:
                f.write(line)
                f.flush()
                os.fsync(f.fileno())
            finally:
                fcntl.flock(f.fileno(), fcntl.LOCK_UN)
    except ImportError:
        # Windows: fcntl недоступен. Параллельная запись может перемешать строки.
        with path.open('a', encoding='utf-8') as f:
            f.write(line)


# PEP 695: type-alias для читаемости.
type LogEntryType = str

# Типы записей — immutable tuple (семантически константа, не список).
VALID_TYPES: tuple[LogEntryType, ...] = (
    'action',         # действие агента (создал файл, запустил команду)
    'error',          # ошибка
    'warning',        # предупреждение
    'info',           # информационное сообщение
    'success',        # успешное завершение
    'manual_change',  # ручная правка пользователя
    'phase',          # смена фазы
    'gate',           # гейт (settings, deps, adr, и т.д.)
    'test',           # результат теста
    'commit',         # git commit
    'subagent',       # вызов субагента
    'loop_detected',  # обнаружена петля
    'user_input',     # команда пользователя
    'context',        # загрузка контекста
    'step_start',     # старт шага
    'step_complete',  # завершение шага
    'step_fail',      # провал шага
    'file_create',    # создание файла
    'file_modify',    # изменение файла
    'file_delete',    # удаление файла
    'git_branch',     # создание/переключение ветки
    'git_commit',     # git commit
    'git_push',       # git push
    'git_merge',      # git merge
    'mcp_call',       # вызов MCP tool
    'a2a_call',       # вызов A2A agent
    'otel_export',    # экспорт telemetry
    'checkpoint',     # save/load checkpoint
    'permission',     # запрос разрешения
    'validation',     # результат валидации
    'verify',         # P2-7 fix: результат verify_subagent
)

# ANSI-цвета для типов — общий для add_entry и tail_log.
_TYPE_COLORS: dict[str, str] = {
    'error':         '\033[91m',  # красный
    'warning':       '\033[93m',  # жёлтый
    'success':       '\033[92m',  # зелёный
    'info':          '\033[94m',  # синий
    'step_start':    '\033[95m',  # пурпурный
    'step_complete': '\033[92m',
    'step_fail':     '\033[91m',
}
_ANSI_RESET = '\033[0m'


# ─── Event hooks integration (UMP Phase 0) ────────────────────────────────────


def _trigger_event_hooks(
    entry_type: str,
    step: str | None = None,
    phase: str | None = None,
    details: dict[str, object] | None = None
) -> None:
    """
    Вызвать registered event hooks на основе типа записи.

    Интеграция сump.status_api for notifications/analytics/dashboard.
    """
    try:
        from scripts.ump.status_api import (
            on_checkpoint_saved,
            on_gate_failed,
            on_phase_completed,
            on_step_finalized,
            on_step_started,
        )

        if entry_type == 'step_start':
            on_step_started(step_id=step or '?', phase=phase)
        elif entry_type == 'phase' and details and details.get('status') == 'success':
            on_phase_completed(step_id=step or '?', phase=phase or '?', status='success')
        elif entry_type == 'gate' and details:
            gate_name = str(details.get('gate_name', 'unknown'))
            message = str(details.get('message', 'Gate check failed'))
            on_gate_failed(gate_name=gate_name, step_id=step or '?', message=message)
        elif entry_type == 'checkpoint':
            checkpoint_path = str(details.get('path', '')) if details else ''
            on_checkpoint_saved(step_id=step or '?', phase=phase or '?', checkpoint_path=checkpoint_path)
        elif entry_type == 'step_complete':
            on_step_finalized(step_id=step or '?', status='success')
        elif entry_type == 'step_fail':
            on_step_finalized(step_id=step or '?', status='failed')
    except Exception:
        pass


def add_entry(
    message: str,
    entry_type: LogEntryType = 'action',
    step: str | None = None,
    phase: str | None = None,
    details: dict[str, object] | None = None,
    level: str | None = None,
) -> None:
    """Добавить запись в лог.

    Каждая запись содержит timestamp, type, message, step, phase, details.
    Все действия агента должны логироваться — от старта шага до завершения.

    P0-fix: срабатывает маскирование секретов в details и файловая блокировка
    для безопасной параллельной записи субагентов (см. 18-parallel-subagents.md).
    P1-fix: поле версии называется `agent_version` согласно правилу
    13-logging-v40.md §"Структура записи". Старое поле `ump_version` сохранено
    как alias для обратной совместимости со старыми парсерами логов.
    P2-6 fix: автоматическая ротация лога при превышении MAX_LOG_SIZE_BYTES
    (по умолчанию 50 MB). Старые записи перемещаются в agent.log.1, agent.log.2,
    ... до MAX_LOG_FILES (по умолчанию 5). См. _rotate_if_needed().
    P2-7 fix: опциональное поле `level` (DEBUG/INFO/WARNING/ERROR). Если
    передано — записывается в entry. Фильтрация по level — в show_log()
    через --level.
    """
    if entry_type not in VALID_TYPES:
        print(f'WARNING: unknown log type {entry_type!r}. Using "action".', file=sys.stderr)
        entry_type = 'action'

    # P0-fix: не пишем секреты в открытую
    safe_details = _redact_details(details)

    # P2-fix (v0.0.3 audit): sanitize newlines в message.
    # Ранее multi-line message ломал JSONL: каждая строка парсилась как
    # отдельная JSON-запись. Теперь все newline-символы (\n, \r, U+0085,
    # U+2028, U+2029) заменяются на литерал '\\n' — JSON остаётся one-line.
    # Найдено hypothesis property-based тестом (test_property_based.py).
    safe_message = message
    for nl in ('\r\n', '\r', '\n', '\x85', '\u2028', '\u2029'):
        safe_message = safe_message.replace(nl, '\\n')

    entry: dict[str, object] = {
        'timestamp': datetime.now().isoformat(),
        'type': entry_type,
        'message': safe_message,
        'agent_version': UMP_VERSION,
        # alias для обратной совместимости — старые парсеры могут читать ump_version.
        'ump_version': UMP_VERSION,
        'hostname': socket.gethostname(),
        'pid': os.getpid(),
    }
    # P2-7 fix: level-based фильтрация. Если caller передал level — записываем.
    # В future можно фильтровать через env var UMP_LOG_LEVEL.
    if level:
        entry['level'] = level.upper()
    if step:
        entry['step'] = step
    if phase:
        entry['phase'] = phase
    if safe_details:
        entry['details'] = safe_details

    # P2-6 fix: проверить, не нужно ли ротировать лог перед записью.
    # Ротация происходит автоматически при превышении MAX_LOG_SIZE_BYTES.
    try:
        _rotate_if_needed()
    except OSError:
        # P1-15 fix: точечный except OSError — все ошибки ротации файловые
        # (rename, unlink, stat). Не падать, если ротация не сработала.
        pass

    _atomic_append(LOG_FILE, json.dumps(entry, ensure_ascii=False) + '\n')
    # UMP Phase 0: trigger event hooks after logging
    _trigger_event_hooks(entry_type, step, phase, safe_details)

    color = _TYPE_COLORS.get(entry_type, '')
    step_str = f' [step {step}]' if step else ''
    phase_str = f' [phase {phase}]' if phase else ''
    print(f'{color}Logged: [{entry_type}]{step_str}{phase_str} {message}{_ANSI_RESET}')


def tail_log(lines: int = 50) -> None:
    """Показать последние N записей."""
    if not LOG_FILE.exists():
        print('Лог пуст.')
        return
    with LOG_FILE.open(encoding='utf-8') as f:
        all_lines = f.readlines()
    for line in all_lines[-lines:]:
        try:
            entry = json.loads(line)
            ts = entry.get('timestamp', '')[:19]
            etype = entry.get('type', '?')
            msg = entry.get('message', '')
            step = entry.get('step', '')
            phase = entry.get('phase', '')
            ump_version = entry.get('ump_version') or entry.get('agent_version', '')
            step_str = f' [step {step}]' if step else ''
            phase_str = f' [phase {phase}]' if phase else ''
            version_str = f' [{ump_version}]' if ump_version else ''
            color = _TYPE_COLORS.get(etype, '')
            print(f'{color}{ts} [{etype:14}]{step_str}{phase_str}{version_str} {msg}{_ANSI_RESET}')
        except (json.JSONDecodeError, KeyError, TypeError):
            # P1-15 fix: точечный except — мусорная строка или повреждённая JSONL.
            print(line.rstrip())


def show_log(
    step: str | None = None,
    entry_type: str | None = None,
    grep: str | None = None,
) -> None:
    """Показать записи с фильтром."""
    if not LOG_FILE.exists():
        print('Лог пуст.')
        return

    count = 0
    with LOG_FILE.open(encoding='utf-8') as f:
        for line in f:
            try:
                entry = json.loads(line)
            except json.JSONDecodeError:
                # P1-15 fix: точечный except — мусорная/повреждённая строка.
                continue

            if step and entry.get('step') != step:
                continue
            if entry_type and entry.get('type') != entry_type:
                continue
            if grep and grep.lower() not in entry.get('message', '').lower():
                continue

            ts = entry.get('timestamp', '')[:19]
            etype = entry.get('type', '?')
            msg = entry.get('message', '')
            e_step = entry.get('step', '')
            step_str = f' [step {e_step}]' if e_step else ''
            phase = entry.get('phase', '')
            phase_str = f' ({phase})' if phase else ''
            print(f'{ts} [{etype:14}]{step_str}{phase_str} {msg}')
            if entry.get('details'):
                for k, v in entry['details'].items():
                    print(f'    {k}: {v}')
            count += 1

    print(f'\nВсего записей: {count}')


def clear_log(before: str | None = None) -> None:
    """Очистить лог (полностью или до даты)."""
    if not LOG_FILE.exists():
        print('Лог пуст.')
        return

    if before is None:
        LOG_FILE.unlink()
        print('Лог очищен полностью.')
        return

    # Оставить только записи после даты `before`
    kept = 0
    removed = 0
    with LOG_FILE.open(encoding='utf-8') as f:
        lines = f.readlines()
    with LOG_FILE.open('w', encoding='utf-8') as f:
        for line in lines:
            try:
                entry = json.loads(line)
                if entry.get('timestamp', '') >= before:
                    f.write(line)
                    kept += 1
                else:
                    removed += 1
            except json.JSONDecodeError:
                # P1-15 fix: точечный except — мусорная/повреждённая строка.
                f.write(line)
                kept += 1

    print(f'Удалено: {removed}, оставлено: {kept}')


def main() -> None:
    parser = argparse.ArgumentParser(description='Логгер AI-агента')
    sub = parser.add_subparsers(dest='command', required=True)

    p_add = sub.add_parser('add', help='Добавить запись')
    p_add.add_argument('message', help='Текст записи')
    p_add.add_argument('--type', default='action', choices=VALID_TYPES)
    p_add.add_argument('--step', help='ID шага (например, 1.6)')
    p_add.add_argument('--phase', help='Фаза (pre-flight, phase1, и т.д.)')
    p_add.add_argument('--details', help='Доп. детали в формате key=value,key=value')

    p_tail = sub.add_parser('tail', help='Последние N записей')
    p_tail.add_argument('--lines', type=int, default=50)

    p_show = sub.add_parser('show', help='Показать с фильтром')
    p_show.add_argument('--step', help='Фильтр по шагу')
    p_show.add_argument('--type', choices=VALID_TYPES, help='Фильтр по типу')
    p_show.add_argument('--grep', help='Поиск по тексту')

    p_clear = sub.add_parser('clear', help='Очистить лог')
    p_clear.add_argument('--before', help='Удалить записи до даты (ISO format)')

    args = parser.parse_args()

    # PEP 634/636: structural pattern matching — лаконичнее цепочки elif.
    match args.command:
        case 'add':
            details: dict[str, object] | None = None
            if args.details:
                details = {}
                for pair in args.details.split(','):
                    if '=' in pair:
                        k, v = pair.split('=', 1)
                        details[k] = v
            add_entry(args.message, args.type, args.step, args.phase, details)
        case 'tail':
            tail_log(args.lines)
        case 'show':
            show_log(args.step, args.type, args.grep)
        case 'clear':
            clear_log(args.before)


if __name__ == '__main__':
    main()