scripts/agent_log.py - main - kochenov/universal-modular

scripts/agent_log.py
197 строк7 KB
Kochenov Dmitry
feat(foundation): Шаг 1.5 — Создать pyproject.toml + ruff.toml + pre-commit конфиг
10 июл 2026, 16:06
10 июл 2026, 16:06e52daa8
#!/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
from datetime import datetime
from pathlib import Path

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

# Типы записей
VALID_TYPES = [
    'action',         # действие агента (создал файл, запустил команду)
    'error',          # ошибка
    'warning',        # предупреждение
    'manual_change',  # ручная правка пользователя
    'phase',          # смена фазы
    'gate',           # гейт (settings, deps, adr, и т.д.)
    'test',           # результат теста
    'commit',         # git commit
    'subagent',       # вызов субагента
    'loop_detected',  # обнаружена петля
    'user_input',     # команда пользователя
    'context',        # загрузка контекста
]


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

    entry = {
        'timestamp': datetime.now().isoformat(),
        'type': entry_type,
        'message': message,
    }
    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')

    print(f'Logged: [{entry_type}] {message}')


def tail_log(lines: int = 50) -> None:
    """Показать последние N записей."""
    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', '')
            step_str = f' [step {step}]' if step else ''
            print(f'{ts} [{etype:14}]{step_str} {msg}')
        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()