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

scripts/verify_step_completion.py
546 строк26 KB
Dmitry Kochenov
v0.0.7: фаза 11+ — расширение правил, покрытие 24/24, улучшенная валидация
20 июл 2026, 08:19
20 июл 2026, 08:19d7decb3
#!/usr/bin/env python3
"""
Многосторонний верификатор завершения шага.

Запускается в фазе TESTING как дополнительная проверка после основного verify.
Проверяет:
1. Доступность документации для пользователя (если шаг создал docs_target)
2. Целостность ссылок в документации
3. Существование обязательных файлов шага
4. Логирование всех проверок через agent_log.py

Запуск:
    uv run python scripts/verify_step_completion.py <step_id>
    uv run python scripts/verify_step_completion.py 1.7

Exit codes:
    0 — все проверки прошли
    1 — есть критичные ошибки
    2 — шаг не найден
"""
from __future__ import annotations

import argparse
import re
import shlex
import subprocess
import sys
from pathlib import Path

import yaml

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

BASE = detect_base(__file__)
INFRA_DIR = resolve_infra_dir(BASE)
PLANS_DIR = BASE / 'plans'

# Импорт функций из agent_log (как модуль). В submodule-layout agent_log.py в INFRA_DIR/scripts/.
_scripts_dir = INFRA_DIR / 'scripts' if (INFRA_DIR / 'scripts').exists() else BASE / 'scripts'
sys.path.insert(0, str(_scripts_dir))
from agent_log import add_entry  # noqa: E402


def find_step(step_id: str) -> tuple[dict, str] | None:
    """Найти шаг в YAML. Вернуть (step_dict, yaml_filename)."""
    for yfile in sorted(PLANS_DIR.glob('*.yaml')):
        if yfile.name.startswith('99'):
            continue
        try:
            data = yaml.safe_load(yfile.read_text(encoding='utf-8'))
        except Exception:
            continue
        for step in data.get('steps', []):
            if step.get('id') == step_id:
                return step, yfile.name
    return None

def check_docs_target_exists(step: dict) -> tuple[bool, str]:
    """Проверить, что docs_target существует."""
    docs_target = step.get('docs_target', '')
    if not docs_target:
        return False, 'docs_target не указан в YAML'
    full = BASE / docs_target
    if not full.exists():
        return False, f'файл {docs_target} не существует'
    if full.stat().st_size == 0:
        return False, f'файл {docs_target} пустой'
    return True, f'файл {docs_target} существует ({full.stat().st_size} байт)'

def check_docs_target_phase2(step: dict) -> tuple[bool, str]:
    """Проверить, что docs_target в Phase 2 (для завершённого шага)."""
    docs_target = step.get('docs_target', '')
    if not docs_target:
        return False, 'docs_target не указан'
    full = BASE / docs_target
    if not full.exists():
        return False, f'файл {docs_target} не существует'
    content = full.read_text(encoding='utf-8')
    if '> Phase 1 — инструкция для AI-агента' in content and '> Обучающая документация' not in content:
        return False, 'документация в Phase 1 (Phase 2 не применена)'
    if '> Обучающая документация' not in content:
        return False, f'нет маркера Phase 2 в {docs_target}'
    return True, 'Phase 2 применена'

def check_final_testing_section(step: dict) -> tuple[bool, str]:
    """Проверить, что в docs_target есть секция '## Final Testing'."""
    docs_target = step.get('docs_target', '')
    if not docs_target:
        return False, 'docs_target не указан'
    full = BASE / docs_target
    if not full.exists():
        return False, f'файл {docs_target} не существует'
    content = full.read_text(encoding='utf-8')
    if '## Final Testing' not in content:
        return False, f'нет секции "## Final Testing" в {docs_target} — FINAL_TESTING не пройден'
    # Проверить, что в секции есть PASS
    ft_section = content.split('## Final Testing')[-1].split('## ')[0]
    if 'PASS' not in ft_section:
        return False, 'в секции "## Final Testing" нет PASS — verify провален'
    return True, 'секция Final Testing присутствует, PASS зафиксирован'

def check_final_review_section(step: dict) -> tuple[bool, str]:
    """Проверить, что в docs_target есть секция '## Final Review' и P0=0."""
    docs_target = step.get('docs_target', '')
    if not docs_target:
        return False, 'docs_target не указан'
    full = BASE / docs_target
    if not full.exists():
        return False, f'файл {docs_target} не существует'
    content = full.read_text(encoding='utf-8')
    if '## Final Review' not in content:
        return False, f'нет секции "## Final Review" в {docs_target} — FINAL_REVIEW не пройден'
    # Проверить, что P0 = 0 (блокирующих ошибок нет)
    fr_section = content.split('## Final Review')[-1].split('## ')[0]
    import re
    # Ищем "P0 (критические): N" или "P0.*: N"
    p0_match = re.search(r'P0[^:]*?:\s*(\d+)', fr_section)
    if p0_match and int(p0_match.group(1)) > 0:
        return False, f'в Final Review P0 = {p0_match.group(1)} > 0 — есть блокирующие проблемы'
    return True, 'секция Final Review присутствует, P0=0 (release readiness OK)'

def check_required_sections(step: dict) -> tuple[bool, list[str]]:
    """Проверить обязательные секции Phase 2."""
    docs_target = step.get('docs_target', '')
    if not docs_target:
        return False, ['docs_target не указан']
    full = BASE / docs_target
    if not full.exists():
        return False, [f'{docs_target} не существует']
    content = full.read_text(encoding='utf-8')

    required = [
        '## Что вы получите в конце шага',
        '## Предварительные требования',
        '## Почему именно так',
        '## Пошаговая инструкция',
        '## Как проверить',
        '## Типичные ошибки и их решения',
        '## Что дальше',
        '## История действий агента',
        '## Реестр файлов шага',
        '## Реестр тестов шага',
        '## VERIFIED_CHECKLIST',
    ]
    missing = [s for s in required if s not in content]
    if missing:
        return False, missing
    return True, []

def check_verified_checklist_closed(step: dict) -> tuple[bool, list[str]]:
    """Проверить, что все пункты VERIFIED_CHECKLIST закрыты [x]."""
    docs_target = step.get('docs_target', '')
    if not docs_target:
        return False, ['docs_target не указан']
    full = BASE / docs_target
    if not full.exists():
        return False, [f'{docs_target} не существует']
    content = full.read_text(encoding='utf-8')

    # Найти секцию VERIFIED_CHECKLIST
    match = re.search(r'## VERIFIED_CHECKLIST\s*\n(.*?)(?=\n## |\Z)', content, re.DOTALL)
    if not match:
        return False, ['секция VERIFIED_CHECKLIST не найдена']

    section = match.group(1)
    # Найти незакрытые пункты
    open_items = re.findall(r'^- \[ \]\s+(.+)$', section, re.MULTILINE)
    if open_items:
        return False, open_items
    return True, []

def check_history_min_records(step: dict) -> tuple[bool, str]:
    """Проверить, что в Истории действий минимум 5 записей."""
    docs_target = step.get('docs_target', '')
    if not docs_target:
        return False, 'docs_target не указан'
    full = BASE / docs_target
    if not full.exists():
        return False, f'{docs_target} не существует'
    content = full.read_text(encoding='utf-8')

    # Найти секцию История действий
    match = re.search(r'## История действий агента\s*\n(.*?)(?=\n## |\Z)', content, re.DOTALL)
    if not match:
        return False, 'секция История действий не найдена'

    section = match.group(1)
    # Подсчитать строки таблицы (исключая заголовок и разделитель)
    rows = re.findall(r'^\|.*\|$', section, re.MULTILINE)
    # Убрать заголовок и разделитель
    data_rows = [r for r in rows if not re.match(r'^\|[\s\-:|]+\|$', r) and 'Время' not in r]
    # Для verify-only шагов (document, user_action, bootstrap) минимум 3, иначе 5
    step_type = step.get('step_type', '')
    min_history = 3 if step_type in ('document', 'user_action', 'bootstrap') else 5
    if len(data_rows) < min_history:
        return False, f'записей: {len(data_rows)} (нужно минимум {min_history})'
    return True, f'записей: {len(data_rows)}'

def check_test_registry_has_pass(step: dict) -> tuple[bool, str]:
    """Проверить, что Реестр тестов содержит хотя бы одну запись PASS."""
    docs_target = step.get('docs_target', '')
    if not docs_target:
        return False, 'docs_target не указан'
    full = BASE / docs_target
    if not full.exists():
        return False, f'{docs_target} не существует'
    content = full.read_text(encoding='utf-8')

    match = re.search(r'## Реестр тестов шага\s*\n(.*?)(?=\n## |\Z)', content, re.DOTALL)
    if not match:
        return False, 'секция Реестр тестов не найдена'

    section = match.group(1)
    if 'PASS' not in section and 'PASSED' not in section:
        return False, 'нет записей PASS/PASSED в реестре тестов'
    return True, 'есть PASS/PASSED записи'

def check_verify_command(step: dict) -> tuple[bool, str]:
    """Запустить verify-команду из YAML.

    P0-7 fix: ранее использовался subprocess.run(cmd, shell=True, ...), что
    позволяло инъекции через cmd (взято из YAML-плана). Теперь — list-form
    через shlex.split(). Если в cmd есть shell-конструкции (|, &&, $(), ...),
    выводится FAIL с понятным сообщением — автор плана должен явно переписать
    на list-form в YAML (verify: ['cmd', 'arg1', 'arg2']) или использовать
    отдельный shell-скрипт.
    """
    verify = step.get('verify', [])
    if not verify:
        return True, 'verify не указан (пропуск)'  # для user_action
    cmd = verify[0]
    # Поддержка обоих форматов: list-form ['cmd', 'arg'] и str 'cmd arg'.
    if isinstance(cmd, list):
        args = cmd
    else:
        # Проверяем на shell-метасимволы — если есть, отказываемся выполнять.
        shell_metachars = ('&&', '||', ';', '|', '>', '<', '$(', '`', '*')
        if any(mc in cmd for mc in shell_metachars):
            return False, (
                f'verify FAIL: команда содержит shell-метасимволы: {cmd!r}\n'
                f'  Перепишите verify в YAML как list-form: verify: [["cmd", "arg1", "arg2"]]\n'
                f'  или вынесите логику в отдельный .sh-скрипт и вызывайте его.'
            )
        args = shlex.split(cmd)
    try:
        result = subprocess.run(
            args, capture_output=True, text=True,
            cwd=BASE, timeout=120,
        )
        if result.returncode == 0:
            return True, f'verity PASS: {cmd}'
        return False, f'verify FAIL (exit {result.returncode}):\n  cmd: {cmd}\n  stderr: {result.stderr[:500]}'
    except subprocess.TimeoutExpired:
        return False, f'verify TIMEOUT (120s): {cmd}'
    except Exception as e:
        return False, f'verify ERROR: {e}'

def check_docs_accessible_to_user(step: dict) -> tuple[bool, str]:
    """Проверить, что документация доступна пользователю.

    Для шагов, создающих docs_target — это означает:
    1. Файл существует и читается
    2. Файл валидный markdown (баланс бэктиков, таблицы)
    3. Файл доступен через VitePress (если docs/.vitepress/ есть) — это проверяется
       через npm run docs:build
    """
    docs_target = step.get('docs_target', '')
    if not docs_target or not (BASE / docs_target).exists():
        return False, f'docs_target {docs_target} не существует'

    full = BASE / docs_target
    content = full.read_text(encoding='utf-8')

    # Проверка балансировки бэктиков
    backticks = content.count('```')
    if backticks % 2 != 0:
        return False, f'незакрытый код-блок: {backticks} тройных бэктиков (нечётное)'

    # Проверка H1
    has_h1 = bool(re.search(r'^# .+', content, re.MULTILINE))
    if not has_h1:
        return False, 'нет H1 заголовка'

    # Проверка через validate_md.py (если существует)
    validate_md = _scripts_dir / 'validate_md.py'
    if validate_md.exists():
        try:
            result = subprocess.run(
                # : sys.executable — гарантированно тот же интерпретатор.
                # uv run здесь не нужен: verify_step_completion уже запущен внутри uv run
                # (через make step-verify или run_phase.py), поэтому venv активен.
                [sys.executable, str(validate_md), str(full)],
                capture_output=True, text=True, cwd=BASE, timeout=30,
            )
            if result.returncode != 0:
                return False, f'validate_md.py FAIL:\n{result.stdout[:500]}'
        except Exception as e:
            return False, f'validate_md.py ERROR: {e}'

    return True, 'документация валидна и доступна'

def check_docs_basic(step: dict) -> tuple[bool, str]:
    """Базовая проверка для bootstrap/user_action шагов.

    Только: H1 заголовок + баланс бэктиков. Не требует Phase 2 маркер.
    """
    docs_target = step.get('docs_target', '')
    if not docs_target or not (BASE / docs_target).exists():
        return False, f'docs_target {docs_target} не существует'

    full = BASE / docs_target
    content = full.read_text(encoding='utf-8')

    # Проверка балансировки бэктиков
    backticks = content.count('```')
    if backticks % 2 != 0:
        return False, f'незакрытый код-блок: {backticks} тройных бэктиков (нечётное)'

    # Проверка H1
    has_h1 = bool(re.search(r'^# .+', content, re.MULTILINE))
    if not has_h1:
        return False, 'нет H1 заголовка'

    return True, 'базовая структура OK'

def check_no_placeholders(step: dict) -> tuple[bool, list[str]]:
    """Проверить, что в docs_target нет незаменённых плейсхолдеров."""
    docs_target = step.get('docs_target', '')
    if not docs_target:
        return True, []
    full = BASE / docs_target
    if not full.exists():
        return True, []

    content = full.read_text(encoding='utf-8')
    # Найти <...> вне код-блоков
    in_code = False
    placeholders = []
    # Разрешённые placeholder'ы (инструкции для пользователя — не баг)
    ALLOWED = {'username', 'owner', 'your_email@example.com', 'you',
               'personal-owner', 'work', 'path', 'version', 'X.Y', 'NN', 'N', 'M',
               'step_id', 'tool', 'имя_инструмента', 'действие', 'URL', 'URL_репозитория'}
    for i, line in enumerate(content.split('\n'), 1):
        # Пропустить HTML-комментарии <!-- ... -->
        if line.strip().startswith('<!--') or line.strip().endswith('-->'):
            continue
        if line.count('```') % 2 == 1:
            in_code = not in_code
            continue
        if in_code:
            continue
        # Найти <...> кроме HTML-тегов и разрешённых placeholder'ов
        for m in re.finditer(r'<([^>]+)>', line):
            inner = m.group(1).strip().lower()
            # P11-B-2 fix (фаза 11, Неделя 2): закрытые HTML-теги (/sup, /sub,
            # /code, ...) ранее отсутствовали в skip-списке — это приводило к
            # ложному срабатыванию на </sup> в markdown-документации. Список
            # теперь симметричен: для каждого открывающего тега есть закрывающий.
            if inner in ['br', 'hr',
                         'sup', 'sub', 'b', 'i', 'strong', 'em', 'code',
                         '/sup', '/sub', '/b', '/i', '/strong', '/em', '/code',
                         '/hr', '/br',
                         'p', '/p', 'span', '/span', 'div', '/div',
                         'ul', '/ul', 'ol', '/ol', 'li', '/li',
                         'table', '/table', 'tr', '/tr', 'td', '/td', 'th', '/th',
                         'h1', '/h1', 'h2', '/h2', 'h3', '/h3', 'h4', '/h4',
                         'h5', '/h5', 'h6', '/h6',
                         'pre', '/pre', 'blockquote', '/blockquote',
                         'a', '/a', 'img', 'details', '/details', 'summary', '/summary']:
                continue
            if inner in ALLOWED:
                continue
            placeholders.append(f'строка {i}: <{m.group(1)}>')

    if placeholders:
        return False, placeholders
    return True, []

def main():
    parser = argparse.ArgumentParser(description='Многосторонний верификатор шага')
    parser.add_argument('step_id', help='ID шага, например 1.7')
    parser.add_argument('--strict', action='store_true',
                        help='Строгий режим: warning = error')
    args = parser.parse_args()

    step_info = find_step(args.step_id)
    if step_info is None:
        print(f'ERROR: шаг {args.step_id} не найден')
        sys.exit(2)

    step, yfile = step_info
    step_type = step.get('step_type', '')

    print(f'\n{"=" * 70}')
    print(f'Многосторонняя проверка шага {args.step_id} ({step_type})')
    print(f'YAML: {yfile}')
    print(f'{"=" * 70}\n')

    # Для user_action, bootstrap и document — упрощённая проверка
    # Bootstrap/Document-шаги (verify-only) проверяют, что файлы/инструменты на месте,
    # но не создают новый код. Для них не требуется полная Phase 2 документация.
    # : добавлен document — document-шаги используют упрощённый шаблон
    # без Phase 1 маркера, поэтому PHASE2A не может установить Phase 2 маркер.
    if step_type in ('user_action', 'bootstrap', 'document'):
        print(f'Шаг типа {step_type} — упрощённая проверка')
        print('  Требуется: docs_target существует, verify passed,')
        print('             VERIFIED_CHECKLIST закрыт, история ≥ 3 записей')
        print()

        checks_bootstrap = [
            ('docs_target exists', lambda: check_docs_target_exists(step)),
            ('VERIFIED_CHECKLIST closed', lambda: check_verified_checklist_closed(step)),
            ('history ≥ 3 records', lambda: check_history_min_records(step)),
            ('verify command passes', lambda: check_verify_command(step)),
            ('docs accessible (basic)', lambda: check_docs_basic(step)),
        ]

        passed = 0
        failed = 0
        failures = []

        for name, check_fn in checks_bootstrap:
            try:
                result = check_fn()
                if isinstance(result, tuple) and len(result) == 2:
                    ok, details = result
                else:
                    ok, details = result, ''
                if ok:
                    print(f'  [✓] {name}: {details if isinstance(details, str) else "OK"}')
                    passed += 1
                    add_entry(f'Check "{name}" PASSED: {details if isinstance(details, str) else ""}',
                              'test', step=args.step_id,
                              details={'check': name, 'result': 'pass'})
                else:
                    details_str = '; '.join(str(d) for d in details[:3]) if isinstance(details, list) else str(details)
                    print(f'  [✗] {name}: {details_str}')
                    failed += 1
                    failures.append((name, details_str))
                    add_entry(f'Check "{name}" FAILED: {details_str}',
                              'error', step=args.step_id,
                              details={'check': name, 'result': 'fail', 'details': details_str})
            except Exception as e:
                print(f'  [!] {name}: EXCEPTION {e}')
                failed += 1
                failures.append((name, f'EXCEPTION: {e}'))

        print(f'\n{"=" * 70}')
        print(f'Итог ({step_type}): {passed} passed, {failed} failed')
        if failures:
            print('Неудачные проверки:')
            for name, msg in failures:
                print(f'  - {name}: {msg}')
        print(f'{"=" * 70}')

        if failed > 0:
            add_entry(f'Step {args.step_id} verification FAILED ({failed} checks, {step_type})',
                      'error', step=args.step_id,
                      details={'passed': passed, 'failed': failed, 'failures': [f[0] for f in failures]})
            sys.exit(1)
        else:
            add_entry(f'Step {args.step_id} verification PASSED ({passed} checks, {step_type})',
                      'test', step=args.step_id,
                      details={'passed': passed, 'failed': 0})
            sys.exit(0)

    # Полный набор проверок для обычных шагов
    checks = [
        ('docs_target exists', lambda: check_docs_target_exists(step)),
        ('Phase 2 applied', lambda: check_docs_target_phase2(step)),
        ('required sections', lambda: check_required_sections(step)),
        ('VERIFIED_CHECKLIST closed', lambda: check_verified_checklist_closed(step)),
        ('history ≥ 5 records', lambda: check_history_min_records(step)),
        ('test registry has PASS', lambda: check_test_registry_has_pass(step)),
        ('verify command passes', lambda: check_verify_command(step)),
        ('docs accessible to user', lambda: check_docs_accessible_to_user(step)),
        ('no placeholders', lambda: check_no_placeholders(step)),
        ('Final Testing section (v36)', lambda: check_final_testing_section(step)),
        ('Final Review section (v36)', lambda: check_final_review_section(step)),
    ]

    passed = 0
    failed = 0
    failures = []

    for name, check_fn in checks:
        try:
            result = check_fn()
            if isinstance(result, tuple) and len(result) == 2:
                ok, details = result
            else:
                ok, details = result, ''
            if ok:
                print(f'  [✓] {name}: {details if isinstance(details, str) else "OK"}')
                passed += 1
                add_entry(f'Check "{name}" PASSED: {details if isinstance(details, str) else ""}',
                          'test', step=args.step_id,
                          details={'check': name, 'result': 'pass'})
            else:
                if isinstance(details, list):
                    details_str = '; '.join(str(d) for d in details[:3])
                    if len(details) > 3:
                        details_str += f' (и ещё {len(details) - 3})'
                else:
                    details_str = str(details)
                print(f'  [✗] {name}: {details_str}')
                failed += 1
                failures.append((name, details_str))
                add_entry(f'Check "{name}" FAILED: {details_str}',
                          'error', step=args.step_id,
                          details={'check': name, 'result': 'fail', 'details': details_str})
        except Exception as e:
            print(f'  [!] {name}: EXCEPTION {e}')
            failed += 1
            failures.append((name, f'EXCEPTION: {e}'))
            add_entry(f'Check "{name}" EXCEPTION: {e}',
                      'error', step=args.step_id,
                      details={'check': name, 'result': 'exception'})

    print(f'\n{"=" * 70}')
    print(f'Итог: {passed} passed, {failed} failed')
    if failures:
        print('Неудачные проверки:')
        for name, msg in failures:
            print(f'  - {name}: {msg}')
    print(f'{"=" * 70}')

    if failed > 0:
        add_entry(f'Step {args.step_id} verification FAILED ({failed} checks)',
                  'error', step=args.step_id,
                  details={'passed': passed, 'failed': failed, 'failures': [f[0] for f in failures]})
        sys.exit(1)
    else:
        add_entry(f'Step {args.step_id} verification PASSED ({passed} checks)',
                  'test', step=args.step_id,
                  details={'passed': passed, 'failed': 0})
        sys.exit(0)

if __name__ == '__main__':
    main()