scripts/finalize_step.py - main - kochenov/universal-m

scripts/finalize_step.py
417 строк20 KB
Kochenov Dmitry
feat(step-1.0): Verify clone + STEP_STATE + .agent/ присутствуют
11 июл 2026, 10:26
11 июл 2026, 10:263fed4c2
#!/usr/bin/env python3
"""
Автоматический финализатор шага (аудит v11).

Заменяет чтение step-finalizer.md (46KB) одной командой.

Выполняет:
1. Проверка docs_target (Phase 2, VERIFIED_CHECKLIST, история ≥ 5)
2. Многосторонняя проверка через verify_step_completion.py
3. Git: add, commit (squash), checkout main, merge --squash, push
4. Обновление STEP_STATE.md (CURRENT_PHASE=awaiting-continue, COMPLETED_STEPS)
5. Удаление feature-ветки
6. Обновление снимка файлов (detect_manual_changes.py --update-snapshot)
7. Логирование всех действий

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

Exit codes:
    0 — шаг финализирован успешно
    1 — ошибка (проверки не пройдены, git конфликт, и т.д.)
    2 — шаг не найден
"""
import argparse
import os
import re
import subprocess
import sys
import yaml
from datetime import datetime
from pathlib import Path

BASE = Path(__file__).parent.parent
PLANS_DIR = BASE / 'plans'

# Импорт функций из agent_log
sys.path.insert(0, str(BASE / 'scripts'))
from agent_log import add_entry


def find_step(step_id: str) -> tuple[dict, str, str] | None:
    """Найти шаг в YAML. Вернуть (step_dict, yaml_filename, next_step_dict)."""
    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
        steps = data.get('steps', [])
        for i, step in enumerate(steps):
            if step.get('id') == step_id:
                next_step = steps[i + 1] if i + 1 < len(steps) else None
                return step, yfile.name, next_step
    return None


def run(cmd: str, check: bool = True, cwd: Path | None = None) -> subprocess.CompletedProcess:
    """Запустить shell-команду."""
    if cwd is None:
        cwd = BASE
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True, cwd=cwd, timeout=120)
    if check and result.returncode != 0:
        print(f'ERROR: command failed: {cmd}', file=sys.stderr)
        print(f'stdout: {result.stdout[:500]}', file=sys.stderr)
        print(f'stderr: {result.stderr[:500]}', file=sys.stderr)
    return result


def log(msg: str, step: str, entry_type: str = 'action', details: dict | None = None):
    """Логировать действие."""
    print(f'[{entry_type}] {msg}')
    add_entry(msg, entry_type, step=step, details=details)


def check_docs_target(step: dict, step_id: str) -> tuple[bool, str]:
    """Проверить docs_target."""
    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')

    # Phase 2 маркер
    if '> Phase 1 — инструкция для AI-агента' in content and '> Обучающая документация' not in content:
        return False, f'Phase 2 не применена (остался Phase 1 маркер)'

    # VERIFIED_CHECKLIST — все [x]
    match = re.search(r'## VERIFIED_CHECKLIST\s*\n(.*?)(?=\n## |\Z)', content, re.DOTALL)
    if not match:
        return False, 'секция VERIFIED_CHECKLIST не найдена'
    open_items = re.findall(r'^- \[ \]\s+(.+)$', match.group(1), re.MULTILINE)
    if open_items:
        return False, f'незакрытые пункты: {open_items[:3]}'

    # История ≥ 5 (или ≥ 3 для verify-only шагов: document, user_action)
    min_history = 3 if step.get('step_type', '') in ('document', 'user_action') else 5
    match = re.search(r'## История действий агента\s*\n(.*?)(?=\n## |\Z)', content, re.DOTALL)
    if match:
        rows = re.findall(r'^\|.*\|$', match.group(1), re.MULTILINE)
        data_rows = [r for r in rows if not re.match(r'^\|[\s\-:|]+\|$', r) and 'Время' not in r]
        if len(data_rows) < min_history:
            return False, f'история действий: {len(data_rows)} записей (нужно ≥ {min_history})'

    return True, 'OK'


def git_operations(step: dict, step_id: str, step_branch: str) -> tuple[bool, str]:
    """Выполнить git-операции: commit, push, PR, merge (аудит v18).

    Стратегия (аудит v18):
    - Один шаг = одна ветка = один PR = merge в main
    - Если агент не на feature-ветке — переключиться на неё
    - После тестов → push feature-ветки → создать PR
    - Если тесты PASS → merge PR в main
    - Если merge успешен → удалить feature-ветку
    """
    step_title = step.get('title', '')

    # 1. Убедиться, что мы на feature-ветке (аудит v18 — автопереключение)
    branch = run('git branch --show-current').stdout.strip()
    if branch != step_branch:
        log(f'Текущая ветка: {branch}, нужно: {step_branch}', step_id, 'warning')
        # Проверить, существует ли feature-ветка
        r = run(f'git rev-parse --verify {step_branch} 2>/dev/null')
        if r.returncode == 0:
            # Ветка существует — переключаемся
            log(f'Переключение на {step_branch}', step_id, 'action')
            run('git stash 2>/dev/null || true', check=False)
            r = run(f'git checkout {step_branch}')
            run('git stash pop 2>/dev/null || true', check=False)
            if r.returncode != 0:
                return False, f'Не удалось переключиться на {step_branch}: {r.stderr}'
        else:
            # Ветка не существует — создаём от main
            log(f'Создание {step_branch} от main', step_id, 'action')
            run('git stash 2>/dev/null || true', check=False)
            r = run('git rev-parse --verify main 2>/dev/null')
            if r.returncode == 0:
                run('git checkout main')
            run(f'git checkout -b {step_branch}')
            run('git stash pop 2>/dev/null || true', check=False)
        # Поверить повторно
        branch = run('git branch --show-current').stdout.strip()
        if branch != step_branch:
            return False, f'Не на feature-ветке (current={branch}, expected={step_branch})'

    # 2. Удалить runtime-файлы из git index
    log(f'Очистка runtime-файлов из git index', step_id, 'action')
    run('git rm --cached -r _meta/ 2>/dev/null || true', check=False)
    run('git rm --cached _meta/agent.log _meta/agent-snapshot.json _meta/manual-changes.log _meta/step-progress.yaml 2>/dev/null || true', check=False)

    # 3. git add .
    log(f'git add .', step_id, 'action')
    r = run('git add .')
    if r.returncode != 0:
        return False, f'git add failed: {r.stderr}'

    # 4. git commit в feature-ветке
    commit_msg = f'feat(step-{step_id}): {step_title}\n\nДокументация: {step.get("docs_target", "")}'
    msg_file = BASE / '_meta' / f'commit-msg-{step_id}.txt'
    msg_file.parent.mkdir(parents=True, exist_ok=True)
    msg_file.write_text(commit_msg, encoding='utf-8')
    log(f'git commit в feature-ветке', step_id, 'commit')
    r = run(f'git commit -F {msg_file}')
    if r.returncode != 0 and 'nothing to commit' not in r.stdout and 'nothing to commit' not in r.stderr:
        log(f'git commit -F failed, trying -m', step_id, 'warning')
        short_msg = f'feat(step-{step_id}): {step_title}'
        r = run(f"git commit -m '{short_msg}'")
        if r.returncode != 0 and 'nothing to commit' not in r.stdout and 'nothing to commit' not in r.stderr:
            log(f'git commit returned {r.returncode}: {r.stdout[:200]}', step_id, 'warning')

    # 5. Push feature-ветки
    log(f'git push origin {step_branch}', step_id, 'action')
    r = run(f'git push origin {step_branch} 2>&1', check=False)
    if r.returncode != 0:
        # Возможно --set-upstream
        r = run(f'git push --set-upstream origin {step_branch} 2>&1', check=False)
        if r.returncode != 0:
            log(f'git push failed: {r.stderr[:200]}', step_id, 'error')
            # Не return — продолжить, push может быть невозможен без remote
            log(f'Продолжаем без push (возможно нет remote)', step_id, 'warning')
        else:
            log(f'git push OK (set-upstream)', step_id, 'action')
    else:
        log(f'git push OK', step_id, 'action')

    # 6. Создать PR (через gh CLI для GitVerse/GitHub)
    log(f'Создание PR для шага {step_id}', step_id, 'action')
    pr_created = False
    pr_url = ''

    # Попытка через gh CLI
    r = run(f'gh pr create --title "Шаг {step_id}: {step_title}" --body "Автоматический PR для шага {step_id}" --base main --head {step_branch} 2>&1', check=False)
    if r.returncode == 0:
        pr_url = r.stdout.strip()
        pr_created = True
        log(f'PR создан через gh CLI: {pr_url}', step_id, 'action', {'pr_url': pr_url})
    else:
        log(f'gh CLI не установлен или не сработал: {r.stderr[:200]}', step_id, 'warning')
        # Альтернатива: GitVerse API
        # r = run(f'curl -s -X POST ... ')
        # Пока — создать PR вручную
        log(f'Создайте PR вручную на GitVerse: {step_branch} → main', step_id, 'warning')

    # 7. Merge PR в main (если PR создан и тесты прошли)
    if pr_created:
        log(f'Merge PR в main', step_id, 'action')
        r = run(f'gh pr merge {step_branch} --squash --delete-branch 2>&1', check=False)
        if r.returncode == 0:
            log(f'PR слит в main, feature-ветка удалена', step_id, 'commit')
        else:
            log(f'gh pr merge failed: {r.stderr[:200]}', step_id, 'warning')
            # Fallback: локальный squash-merge
            log(f'Fallback: локальный squash-merge', step_id, 'action')
            run('git stash 2>/dev/null || true', check=False)
            r = run('git rev-parse --verify main 2>/dev/null')
            if r.returncode == 0:
                run('git checkout main')
            else:
                run('git checkout -b main')
            run('git stash pop 2>/dev/null || true', check=False)
            run(f'git merge --squash {step_branch}', check=False)
            os.environ['STEP_FINALIZER_ACTIVE'] = '1'
            if msg_file.exists():
                run(f'git commit -F {msg_file}')
            else:
                run(f"git commit -m 'feat(step-{step_id}): {step_title}'")
    else:
        # Нет PR — локальный squash-merge (аудит v16: для случая без remote)
        log(f'Локальный squash-merge (нет remote/gh)', step_id, 'action')
        run('git stash 2>/dev/null || true', check=False)
        r = run('git rev-parse --verify main 2>/dev/null')
        if r.returncode == 0:
            r = run('git checkout main')
            if r.returncode != 0:
                current = run('git branch --show-current').stdout.strip()
                if current != 'main':
                    return False, f'git checkout main failed: {r.stderr}'
        else:
            r = run('git checkout -b main')
            if r.returncode != 0:
                return False, f'git checkout -b main failed: {r.stderr}'
        run('git stash pop 2>/dev/null || true', check=False)

        log(f'git merge --squash {step_branch}', step_id, 'action')
        r = run(f'git merge --squash {step_branch}')
        if r.returncode != 0:
            r2 = run('git status --porcelain')
            if 'UU' in r2.stdout or 'AA' in r2.stdout:
                return False, f'MERGE CONFLICT: {r2.stdout[:200]}. Требуется ручное разрешение.'
            return False, f'git merge --squash failed: {r.stderr}'

        os.environ['STEP_FINALIZER_ACTIVE'] = '1'
        log(f'git commit в main', step_id, 'commit')
        if msg_file.exists():
            r = run(f'git commit -F {msg_file}')
        else:
            r = run(f"git commit -m 'feat(step-{step_id}): {step_title}'")
        if r.returncode != 0 and 'nothing to commit' not in r.stdout and 'nothing to commit' not in r.stderr:
            r = run(f"git commit -m 'feat(step-{step_id})'")
            if r.returncode != 0 and 'nothing to commit' not in r.stdout and 'nothing to commit' not in r.stderr:
                return False, f'git commit in main failed: exit={r.returncode} stderr={r.stderr[:200]}'

    # 8. Push main (если есть remote)
    r = run('git remote -v')
    if r.stdout.strip():
        log(f'git push origin main', step_id, 'action')
        r = run('git push origin main 2>&1', check=False)
        if r.returncode == 0:
            log(f'git push main OK', step_id, 'action')
        else:
            log(f'git push main failed (продолжаем): {r.stderr[:200]}', step_id, 'warning')

    # 9. Удалить feature-ветку (локально)
    log(f'Удаление feature-ветки {step_branch}', step_id, 'action')
    run(f'git branch -D {step_branch} 2>/dev/null || true', check=False)
    run(f'git push origin --delete {step_branch} 2>/dev/null || true', check=False)

    # 10. Удалить временный файл
    msg_file.unlink(missing_ok=True)

    return True, f'PR/merge выполнен, feature-ветка удалена'


def update_step_state(step: dict, step_id: str, next_step: dict | None):
    """Обновить STEP_STATE.md."""
    step_state = BASE / 'STEP_STATE.md'
    if not step_state.exists():
        log('STEP_STATE.md не существует — пропускаю обновление', step_id, 'warning')
        return

    now = datetime.now().strftime('%Y-%m-%d')
    content = step_state.read_text(encoding='utf-8')

    # CURRENT_STEP → next_step_id (или остаётся, если next нет)
    next_step_id = next_step.get('id', step_id) if next_step else step_id
    next_step_docs = next_step.get('docs_target', '') if next_step else ''

    # LAST_COMPLETED_STEP → step_id
    content = re.sub(r'LAST_COMPLETED_STEP:.*', f'LAST_COMPLETED_STEP: {step_id}', content)

    # CURRENT_PHASE → awaiting-continue
    content = re.sub(r'CURRENT_PHASE:.*', 'CURRENT_PHASE: awaiting-continue', content)

    # CURRENT_STEP → next
    if next_step:
        content = re.sub(r'CURRENT_STEP:.*', f'CURRENT_STEP: {next_step_id}', content)
        content = re.sub(r'CURRENT_STEP_DOCS:.*', f'CURRENT_STEP_DOCS: {next_step_docs}', content)

    # Добавить в COMPLETED_STEPS
    step_title = step.get('title', '')
    docs_target = step.get('docs_target', '')
    new_row = f'| {step_id} | {step_title} | {now} | [{docs_target}]({docs_target}) |'

    # Найти секцию COMPLETED_STEPS и добавить строку после заголовка таблицы
    content = re.sub(
        r'(\| Шаг \| Название \| Дата \| Документация \|\n\|-----\|----------\|------\|--------------\|\n)',
        rf'\1{new_row}\n',
        content,
        count=1
    )

    step_state.write_text(content, encoding='utf-8')
    log(f'STEP_STATE.md обновлён: CURRENT_STEP={next_step_id}, LAST_COMPLETED={step_id}', step_id, 'action')


def main():
    parser = argparse.ArgumentParser(description='Автоматический финализатор шага')
    parser.add_argument('step_id', help='ID шага, например 1.0')
    parser.add_argument('--skip-verify', action='store_true', help='Пропустить verify_step_completion.py')
    args = parser.parse_args()

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

    step, yfile, next_step = step_info
    step_branch = step.get('branch', f'feature/step-{args.step_id}')
    step_type = step.get('step_type', '')

    print(f'\n{"=" * 70}')
    print(f'ФИНАЛИЗАЦИЯ ШАГА {args.step_id} — {step.get("title", "")}')
    print(f'Ветка: {step_branch}')
    print(f'Тип: {step_type}')
    print(f'{"=" * 70}\n')

    # 1. Проверка docs_target
    print('--- Проверка docs_target ---')
    ok, msg = check_docs_target(step, args.step_id)
    print(f'  [{"✓" if ok else "✗"}] {msg}')
    if not ok:
        log(f'Финализация FAIL: docs_target — {msg}', args.step_id, 'error')
        sys.exit(1)
    log(f'docs_target проверен: {msg}', args.step_id, 'test')

    # 2. Многосторонняя проверка (если не --skip-verify)
    if not args.skip_verify:
        print('\n--- Многосторонняя проверка (verify_step_completion.py) ---')
        r = run(f'uv run python3 scripts/verify_step_completion.py {args.step_id}', check=False)
        print(r.stdout)
        if r.returncode != 0:
            print(f'\n❌ Многосторонняя проверка НЕ пройдена (exit {r.returncode})')
            print('Используйте --skip-verify для пропуска (не рекомендуется).')
            log(f'Финализация FAIL: verify_step_completion exit {r.returncode}', args.step_id, 'error')
            sys.exit(1)
        log('verify_step_completion PASSED', args.step_id, 'test')

    # 3. Git-операции
    print('\n--- Git-операции ---')
    ok, msg = git_operations(step, args.step_id, step_branch)
    print(f'  [{"✓" if ok else "✗"}] {msg}')
    if not ok:
        log(f'Финализация FAIL: git — {msg}', args.step_id, 'error')
        sys.exit(1)

    # 4. Обновление STEP_STATE.md
    print('\n--- Обновление STEP_STATE.md ---')
    update_step_state(step, args.step_id, next_step)
    print(f'  ✓ STEP_STATE.md обновлён')

    # 5. Обновить снимок файлов
    print('\n--- Обновление снимка файлов ---')
    run('uv run python3 scripts/detect_manual_changes.py --update-snapshot --quiet', check=False)
    print(f'  ✓ Снимок обновлён')

    # 6. Финальный лог
    next_step_id = next_step.get('id', '?') if next_step else '?'
    next_step_title = next_step.get('title', '') if next_step else ''
    log(f'Шаг {args.step_id} финализирован. Следующий: {next_step_id}', args.step_id, 'commit',
        details={'next_step': next_step_id, 'next_title': next_step_title})

    print(f'\n{"=" * 70}')
    print(f'✓ ШАГ {args.step_id} ФИНАЛИЗИРОВАН')
    print(f'{"=" * 70}')
    print(f'\nGit: squash-merge в main выполнен, feature-ветка удалена.')
    print(f'Документация: {step.get("docs_target", "")}')
    print(f'\nСледующий шаг: {next_step_id} — {next_step_title}')
    print(f'\nСкажите "продолжить" для запуска следующего шага.')
    print()

    # Переименовать progress-файл
    progress_file = BASE / '_meta' / 'step-progress.yaml'
    done_file = BASE / '_meta' / f'step-progress-{args.step_id}.done.yaml'
    if progress_file.exists():
        progress_file.rename(done_file)
        print(f'Progress-файл: {done_file.name}')

    sys.exit(0)


if __name__ == '__main__':
    main()