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

scripts/run_phase.py
500 строк20 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
"""
Универсальный исполнитель фаз шага (аудит v12).

ЗАМЕЩАЕТ субагентов для механических фаз. Агент вызывает этот скрипт,
а не пытается прочитать 43KB промпт в контекст субагента.

Фазы, которые скрипт выполняет ПОЛНОСТЬЮ (без участия агента):
  PREFLIGHT  — создать ветку, проверить контекст, обновить STEP_STATE
  TESTING    — запустить verify, записать результаты в docs_target
  PHASE2B    — проверить сохраняемые секции
  FINALIZATION — вызвать finalize_step.py

Фазы, где скрипт делает ПОДГОТОВКУ, а агент — ТВОРЧЕСКУЮ работу:
  PHASE1       — скрипт создаёт шаблон docs_target, агент заполняет секции
  IMPLEMENTATION — скрипт подготавливает окружение, агент пишет код
  PHASE2A      — скрипт отмечает секции для переписывания, агент переписывает

АВТО-ЛОГИРОВАНИЕ: все действия автоматически пишутся в _meta/agent.log

Запуск:
    uv run python scripts/run_phase.py <step_id> <phase>
    uv run python scripts/run_phase.py 1.0 PREFLIGHT
    uv run python scripts/run_phase.py 1.0 PHASE1
    uv run python scripts/run_phase.py 1.0 TESTING

Exit codes:
    0 — фаза выполнена
    1 — ошибка
    2 — шаг не найден
    3 — precondition не пройден
"""
import argparse
import re
import subprocess
import sys
from datetime import datetime
from pathlib import Path

import yaml

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

sys.path.insert(0, str(BASE / 'scripts'))
from agent_log import add_entry


def find_step(step_id: str) -> dict | None:
    """Найти шаг в YAML."""
    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
    return None


def run(cmd: str, check: bool = True) -> subprocess.CompletedProcess:
    """Запустить команду."""
    return subprocess.run(cmd, shell=True, capture_output=True, text=True,
                          cwd=BASE, timeout=120)


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 set_phase(step_id: str, phase: str, status: str):
    """Обновить прогресс-файл через orchestrate_step.py."""
    run(f'python3 scripts/orchestrate_step.py {step_id} --set-phase {phase} {status}', check=False)


def sed_step_state(old: str, new: str):
    """Атомарно обновить CURRENT_PHASE в STEP_STATE.md."""
    step_state = BASE / 'STEP_STATE.md'
    if step_state.exists():
        run(f"sed -i 's/^CURRENT_PHASE: {old}$/CURRENT_PHASE: {new}/' STEP_STATE.md", check=False)


# ============================================================================
# ФАЗА: PREFLIGHT
# ============================================================================

def phase_preflight(step: dict, step_id: str) -> int:
    """Pre-flight: создать ветку, проверить контекст."""
    step_branch = step.get('branch', f'feature/step-{step_id}')
    step_type = step.get('step_type', '')
    requires_approval = step.get('requires_approval', False)

    print(f'\n=== PREFLIGHT для шага {step_id} ===')

    # 1. Проверить cwd
    log('Проверка cwd', step_id, 'action')
    r = run('test -f AGENTS.md && echo OK || echo FAIL')
    if 'FAIL' in r.stdout:
        log('ERROR: не в корне проекта (нет AGENTS.md)', step_id, 'error')
        return 1
    log('cwd OK', step_id, 'test')

    # 2. Проверить git
    r = run('git rev-parse --is-inside-work-tree 2>/dev/null')
    if r.stdout.strip() != 'true':
        log('Git не инициализирован', step_id, 'error')
        return 1

    # 3. Проверить STEP_STATE.md для resume
    step_state = BASE / 'STEP_STATE.md'
    if step_state.exists():
        content = step_state.read_text(encoding='utf-8')
        # Извлечь CURRENT_PHASE
        m = re.search(r'CURRENT_PHASE:\s*(\S+)', content)
        if m:
            current_phase = m.group(1)
            if current_phase == 'awaiting-continue':
                log('CONTINUE-ГЕЙТ: awaiting-continue. Скажите "продолжить"', step_id, 'gate')
                print('\n⚠️ CONTINUE-ГЕЙТ: предыдущий шаг финализирован.')
                print('   Скажите "продолжить" для запуска этого шага.')
                return 1
            log(f'Resume: CURRENT_PHASE={current_phase}', step_id, 'action')

    # 4. Approval gate
    if requires_approval:
        approval_reason = step.get('approval_reason', '')
        log(f'Approval gate: {approval_reason}', step_id, 'gate')
        print(f'\n⏸ ТРЕБУЕТСЯ ПОДТВЕРЖДЕНИЕ: {approval_reason}')
        print('   Скажите "подтверждаю" для продолжения.')
        return 1

    # 5. Создать feature-ветку
    r = run('git branch --show-current')
    current_branch = r.stdout.strip()
    if current_branch == step_branch:
        log(f'Уже на ветке {step_branch}', step_id, 'action')
    else:
        if current_branch != 'main':
            run('git checkout main', check=False)
        r = run(f'git checkout -b {step_branch}', check=False)
        if r.returncode != 0:
            run(f'git checkout {step_branch}', check=False)
        log(f'Ветка {step_branch} активна', step_id, 'action')

    # 6. Обновить STEP_STATE.md
    if not step_state.exists() and step_id == '1.0':
        # Шаг 1.0 — копировать из template
        run('cp STEP_STATE.template.md STEP_STATE.md')
        log('STEP_STATE.md создан из template', step_id, 'action')

    sed_step_state('awaiting-continue', 'pre-flight')
    sed_step_state('implementation', 'pre-flight')

    # 7. Обновить прогресс
    set_phase(step_id, 'PREFLIGHT', 'completed')
    log('PREFLIGHT завершён', step_id, 'phase')

    print(f'\n✓ PREFLIGHT завершён. Ветка: {step_branch}')
    return 0


# ============================================================================
# ФАЗА: PHASE1 — подготовка шаблона, агент заполняет
# ============================================================================

def phase_phase1(step: dict, step_id: str) -> int:
    """Phase 1: создать шаблон docs_target для заполнения агентом."""
    docs_target = step.get('docs_target', '')
    step_title = step.get('title', '')
    verify_cmd = step.get('verify', [''])[0]
    expected_result = step.get('expected_result', '')

    print(f'\n=== PHASE1 для шага {step_id} ===')

    if not docs_target:
        log('docs_target не указан', step_id, 'error')
        return 1

    full = BASE / docs_target
    full.parent.mkdir(parents=True, exist_ok=True)

    if full.exists():
        content = full.read_text(encoding='utf-8')
        if '> Phase 1 — инструкция для AI-агента' in content:
            log(f'{docs_target} уже в Phase 1 — обновляю', step_id, 'action')
        elif '> Обучающая документация' in content:
            log(f'{docs_target} уже в Phase 2 — пропускаю', step_id, 'warning')
            set_phase(step_id, 'PHASE1', 'completed')
            return 0

    # Создать шаблон Phase 1
    now = datetime.now().strftime('%Y-%m-%d %H:%M')
    template = f"""# Шаг {step_id} — {step_title}

> Phase 1 — инструкция для AI-агента. После завершения шага этот файл
> будет переписан в Phase 2 (обучающий формат для человека) промптом
> .agent/prompts/docs-rewriter-human.md

## Цель

{expected_result}

## Контекст для обязательного чтения

Агент обязан перейти по каждой ссылке и прочитать файл полностью:

1. [00-manifest.md](../../00-manifest.md) — архитектура проекта
2. [.agent/rules/01-step-lifecycle.md](../../.agent/rules/01-step-lifecycle.md) — цикл жизни шага

## План

1. Проверить предусловия шага
2. Выполнить основную задачу шага
3. Запустить верификацию

## Ожидаемый результат

{expected_result}

## VERIFIED_CHECKLIST

### Файлы
- [ ] Файлы шага созданы

### Тесты
- [ ] `{verify_cmd}` — PASSED

### Документация
- [ ] Phase 1 написана (этот файл)
- [ ] Sidebar в docs/.vitepress/config.mts обновлён

## Верификация

```bash
{verify_cmd}
```

## История действий агента

| Время | Действие | Результат |
|------|---------|-----------|
| {now} | Старт Phase 1 | начало |

## Реестр файлов шага

### Созданные файлы
| Файл | Описание |
|------|----------|

### Изменённые файлы
| Файл | Изменение |
|------|-----------|

### Удалённые файлы
| Файл | Причина |
|------|---------|

## Реестр тестов шага

| Время | Команда | Результат |
|------|---------|-----------|
"""

    full.write_text(template, encoding='utf-8')
    log(f'Шаблон Phase 1 создан: {docs_target}', step_id, 'action',
        {'lines': len(template.split('\n'))})

    # Обновить STEP_STATE
    sed_step_state('pre-flight', 'phase1')
    set_phase(step_id, 'PHASE1', 'completed')

    print(f'\n✓ PHASE1: шаблон создан в {docs_target}')
    print('  Агент должен заполнить секции: Цель, Контекст, План, VERIFIED_CHECKLIST')
    return 0


# ============================================================================
# ФАЗА: TESTING — запустить verify, записать результаты
# ============================================================================

def phase_testing(step: dict, step_id: str) -> int:
    """Testing: запустить verify-команду, записать в docs_target."""
    verify_cmd = step.get('verify', [''])[0]
    docs_target = step.get('docs_target', '')

    print(f'\n=== TESTING для шага {step_id} ===')

    if not verify_cmd:
        log('verify не указан — пропускаю', step_id, 'warning')
        set_phase(step_id, 'TESTING', 'completed')
        return 0

    # Запустить verify
    log(f'Запуск verify: {verify_cmd}', step_id, 'test')
    r = run(verify_cmd, check=False)
    now = datetime.now().strftime('%Y-%m-%d %H:%M')

    if r.returncode == 0:
        result_str = 'PASS'
        log('verify PASS', step_id, 'test')
    else:
        result_str = f'FAIL (exit {r.returncode})'
        log(f'verify FAIL: {r.stderr[:200]}', step_id, 'error')

    # Записать в docs_target → Реестр тестов
    if docs_target and (BASE / docs_target).exists():
        content = (BASE / docs_target).read_text(encoding='utf-8')
        test_row = f'| {now} | `{verify_cmd}` | {result_str} |\n'

        # Найти секцию Реестр тестов и добавить строку
        content = re.sub(
            r'(\| Время \| Команда \| Результат \|\n\|------\|---------\|-----------\|\n)',
            rf'\1{test_row}',
            content,
            count=1
        )

        # Закрыть VERIFIED_CHECKLIST → Тесты
        if r.returncode == 0:
            content = content.replace('- [ ] `' + verify_cmd + '` — PASSED', f'- [x] `{verify_cmd}` — PASSED')
            content = re.sub(r'- \[ \] (.+PASSED.*)', r'- [x] \1', content)

        (BASE / docs_target).write_text(content, encoding='utf-8')
        log(f'Результат записан в {docs_target}', step_id, 'action')

    # Обновить STEP_STATE
    sed_step_state('testing', 'phase2')
    set_phase(step_id, 'TESTING', 'completed' if r.returncode == 0 else 'failed')

    if r.returncode != 0:
        print(f'\n✗ TESTING FAIL: verify вернул exit {r.returncode}')
        print(f'  Команда: {verify_cmd}')
        print(f'  stderr: {r.stderr[:300]}')
        return 1

    print('\n✓ TESTING: verify PASS')
    return 0


# ============================================================================
# ФАЗА: PHASE2B — проверить сохраняемые секции
# ============================================================================

def phase_phase2b(step: dict, step_id: str) -> int:
    """Phase 2B: проверить, что сохраняемые секции не потеряны."""
    docs_target = step.get('docs_target', '')

    print(f'\n=== PHASE2B для шага {step_id} ===')

    if not docs_target or not (BASE / docs_target).exists():
        log(f'docs_target не существует: {docs_target}', step_id, 'error')
        return 1

    content = (BASE / docs_target).read_text(encoding='utf-8')
    checks = []

    # 1. История действий ≥ 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]
        ok = len(data_rows) >= 5
        checks.append(('История ≥ 5', ok, f'{len(data_rows)} записей'))
    else:
        checks.append(('История действий', False, 'секция не найдена'))

    # 2. Реестр файлов — 3 подсекции
    for sub in ['### Созданные файлы', '### Изменённые файлы', '### Удалённые файлы']:
        ok = sub in content
        checks.append((sub, ok, 'OK' if ok else 'отсутствует'))

    # 3. Реестр тестов — есть PASS
    match = re.search(r'## Реестр тестов шага\s*\n(.*?)(?=\n## |\Z)', content, re.DOTALL)
    if match:
        has_pass = 'PASS' in match.group(1)
        checks.append(('Реестр тестов has PASS', has_pass, 'OK' if has_pass else 'нет PASS'))
    else:
        checks.append(('Реестр тестов', False, 'секция не найдена'))

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

    # Вывод
    all_ok = True
    for name, ok, detail in checks:
        icon = '✓' if ok else '✗'
        print(f'  [{icon}] {name}: {detail}')
        log(f'Check "{name}": {"PASS" if ok else "FAIL"} — {detail}', step_id,
            'test' if ok else 'error')

    if all_ok:
        set_phase(step_id, 'PHASE2B', 'completed')
        print('\n✓ PHASE2B: все сохраняемые секции intact')
        return 0
    else:
        set_phase(step_id, 'PHASE2B', 'failed')
        print('\n✗ PHASE2B: есть проблемы — исправьте и повторите')
        return 1


# ============================================================================
# ФАЗА: FINALIZATION — вызвать finalize_step.py
# ============================================================================

def phase_finalization(step: dict, step_id: str) -> int:
    """Finalization: вызвать finalize_step.py."""
    print(f'\n=== FINALIZATION для шага {step_id} ===')
    log('Запуск finalize_step.py', step_id, 'action')

    r = run(f'python3 scripts/finalize_step.py {step_id}', check=False)

    if r.returncode == 0:
        set_phase(step_id, 'FINALIZATION', 'completed')
        log('FINALIZATION завершена', step_id, 'phase')
        print(r.stdout)
        return 0
    else:
        log(f'FINALIZATION FAIL (exit {r.returncode})', step_id, 'error')
        print(r.stdout)
        print(r.stderr, file=sys.stderr)
        return 1


# ============================================================================
# MAIN
# ============================================================================

def main():
    parser = argparse.ArgumentParser(description='Универсальный исполнитель фаз шага')
    parser.add_argument('step_id', help='ID шага, например 1.0')
    parser.add_argument('phase', help='Фаза: PREFLIGHT|PHASE1|IMPLEMENTATION|TESTING|PHASE2A|PHASE2B|FINALIZATION')

    args = parser.parse_args()

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

    phase = args.phase.upper()

    # Авто-лог: запуск фазы
    log(f'Фаза {phase} начата для шага {args.step_id}', args.step_id, 'phase')

    # Обновить прогресс: in_progress
    set_phase(args.step_id, phase, 'in_progress')

    # Маршрутизация
    handlers = {
        'PREFLIGHT': phase_preflight,
        'PHASE1': phase_phase1,
        'TESTING': phase_testing,
        'PHASE2B': phase_phase2b,
        'FINALIZATION': phase_finalization,
    }

    handler = handlers.get(phase)
    if handler:
        exit_code = handler(step, args.step_id)
    else:
        # IMPLEMENTATION и PHASE2A — агент делает сам
        print(f'\n=== {phase} для шага {args.step_id} ===')
        print('Эта фаза требует творческой работы агента.')
        print('Скрипт подготовил окружение. Агент, выполни задачу:')
        if phase == 'IMPLEMENTATION':
            print(f'  1. Прочитай docs_target: {step.get("docs_target", "")}')
            print('  2. Создай файлы согласно плану')
            print('  3. Обнови Историю действий и Реестр файлов в docs_target')
            print(f'  4. После завершения — запусти: run_phase.py {args.step_id} TESTING')
        elif phase == 'PHASE2A':
            print(f'  1. Прочитай docs_target: {step.get("docs_target", "")}')
            print('  2. Замени маркер Phase 1 на Phase 2')
            print('  3. Добавь секции: "Что получите", "Почему", "Пошаговая", "Ошибки", "Что дальше"')
            print('  4. НЕ трогай: История, Реестры, VERIFIED_CHECKLIST')
            print(f'  5. После завершения — запусти: run_phase.py {args.step_id} PHASE2B')
        log(f'Фаза {phase} — ожидает творческой работы агента', args.step_id, 'action')
        exit_code = 0

    if exit_code == 0:
        log(f'Фаза {phase} завершена успешно', args.step_id, 'phase')
    else:
        log(f'Фаза {phase} завершилась с ошибкой (exit {exit_code})', args.step_id, 'error')

    sys.exit(exit_code)


if __name__ == '__main__':
    main()