scripts/agent_log.py - main - kochenov/uni

scripts/agent_log.py
267 строк11 KB
Dmitry Kochenov
fix(step-1.0): финализация шага 1.0 и поддержка FINAL_TESTING/FINAL_REVIEW
16 июл 2026, 23:41
16 июл 2026, 23:41dfac4cd
#!/usr/bin/env python3
"""
Логгер всех изменений и ошибок AI-агента (аудит v8).

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

Запуск (из агента или вручную):
    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 формат — по одной записи на строку).
"""
import argparse
import json
import sys
from datetime import datetime
from pathlib import Path

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

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


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

    Аудит v40 (2026-07-16): расширено для требования "всё в лог".
    Каждая запись содержит timestamp, type, message, step, phase, details.
    Все действия агента должны логироваться — от старта шага до завершения.
    """
    if entry_type not in VALID_TYPES:
        print(f'WARNING: unknown log type "{entry_type}". Using "action".', file=sys.stderr)
        entry_type = 'action'

    LOG_FILE.parent.mkdir(parents=True, exist_ok=True)

    # Аудит v40: добавить agent_version и hostname для трассировки
    import os
    import socket
    entry = {
        'timestamp': datetime.now().isoformat(),
        'type': entry_type,
        'message': message,
        'agent_version': 'v40',
        'hostname': socket.gethostname(),
        'pid': os.getpid(),
    }
    if step:
        entry['step'] = step
    if phase:
        entry['phase'] = phase
    if details:
        entry['details'] = details

    with open(LOG_FILE, 'a', encoding='utf-8') as f:
        f.write(json.dumps(entry, ensure_ascii=False) + '\n')

    # Аудит v40: вывод в чат с цветовым кодированием по типу
    type_colors = {
        'error': '\033[91m',       # красный
        'warning': '\033[93m',     # жёлтый
        'success': '\033[92m',     # зелёный
        'info': '\033[94m',        # синий
        'step_start': '\033[95m',  # пурпурный
        'step_complete': '\033[92m',
        'step_fail': '\033[91m',
    }
    reset = '\033[0m'
    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}{reset}')


def tail_log(lines: int = 50) -> None:
    """Показать последние N записей.

    Аудит v40: улучшенный вывод с цветовым кодированием и детальной информацией.
    """
    if not LOG_FILE.exists():
        print('Лог пуст.')
        return
    with open(LOG_FILE, 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', '')
            agent_version = entry.get('agent_version', '')
            step_str = f' [step {step}]' if step else ''
            phase_str = f' [phase {phase}]' if phase else ''
            version_str = f' [v{agent_version}]' if agent_version else ''

            # Аудит v40: цветовое кодирование
            type_colors = {
                'error': '\033[91m',
                'warning': '\033[93m',
                'success': '\033[92m',
                'info': '\033[94m',
                'step_start': '\033[95m',
                'step_complete': '\033[92m',
                'step_fail': '\033[91m',
            }
            reset = '\033[0m'
            color = type_colors.get(etype, '')
            print(f'{color}{ts} [{etype:14}]{step_str}{phase_str}{version_str} {msg}{reset}')
        except Exception:
            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 open(LOG_FILE, encoding='utf-8') as f:
        for line in f:
            try:
                entry = json.loads(line)
            except Exception:
                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 open(LOG_FILE, encoding='utf-8') as f:
        lines = f.readlines()
    with open(LOG_FILE, '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 Exception:
                f.write(line)
                kept += 1

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


def main():
    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()

    if args.command == 'add':
        details = 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)
    elif args.command == 'tail':
        tail_log(args.lines)
    elif args.command == 'show':
        show_log(args.step, args.type, args.grep)
    elif args.command == 'clear':
        clear_log(args.before)


if __name__ == '__main__':
    main()