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

scripts/orchestrate_step.py
584 строки25 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
"""
Helper script for step-orchestrator.md.

Helps the orchestrator parse YAML, generate plan structure, manage progress file,
and check preconditions for phase transitions. This is deterministic — the agent
doesn't need to parse YAML itself (which is error-prone and token-expensive).

Запуск:
    uv run python scripts/orchestrate_step.py <step_id> --print-step
    uv run python scripts/orchestrate_step.py <step_id> --init-progress
    uv run python scripts/orchestrate_step.py <step_id> --set-phase <PHASE> <STATUS>
    uv run python scripts/orchestrate_step.py <step_id> --show-progress
    uv run python scripts/orchestrate_step.py <step_id> --can-start <PHASE>
    uv run python scripts/orchestrate_step.py <step_id> --gen-plan > docs/NN-stage/step-X-Y-PLAN.md

Exit codes:
    0 = OK
    1 = step not found in YAML
    2 = progress file missing (for --set-phase, --show-progress, --can-start)
    3 = invalid phase transition
"""
import argparse
import sys
from datetime import datetime
from pathlib import Path

import yaml

BASE = Path(__file__).parent.parent
PLANS_DIR = BASE / 'plans'
PROGRESS_FILE = BASE / '_meta' / 'step-progress.yaml'

# 7 фаз в строгом порядке (см. 07-plan-first-mode.md §2)
PHASES = [
    'PREFLIGHT',
    'PHASE1',
    'IMPLEMENTATION',
    'TESTING',
    'PHASE2A',
    'PHASE2B',
    'FINALIZATION',
]

# Valid status values
VALID_STATUSES = ['pending', 'in_progress', 'completed', 'failed', 'skipped']


def find_step_in_yaml(step_id: str) -> dict | None:
    """Найти запись шага во всех YAML-файлах plans/."""
    if not PLANS_DIR.exists():
        return None
    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:
                step['_yaml_file'] = yfile.name
                step['_stage'] = data.get('stage')
                step['_stage_name'] = data.get('name')
                return step
    return None


def print_step(step: dict) -> None:
    """Напечатать краткую инфу о шаге (для генерации плана)."""
    print(f"step_id: {step.get('id')}")
    print(f"title: {step.get('title')}")
    print(f"step_type: {step.get('step_type')}")
    print(f"branch: {step.get('branch')}")
    print(f"docs_target: {step.get('docs_target')}")
    print(f"requires_approval: {step.get('requires_approval', False)}")
    print(f"parallel_group: {step.get('parallel_group', '')}")
    print(f"verify: {step.get('verify', [''])[0]}")
    print(f"expected_result: {step.get('expected_result', '')}")
    print("context_to_read:")
    for ctx in step.get('context_to_read', []):
        print(f"  - {ctx}")
    print("prompts_to_use:")
    for p in step.get('prompts_to_use', []):
        print(f"  - {p}")


def init_progress_file(step: dict, force: bool = False, require_ack: bool = True) -> None:
    """Создать _meta/step-progress.yaml с начальным состоянием.

    Защита от случайного перезаписи (Проблема #15) и требование ack-файла (Проблема #1).
    """
    PROGRESS_FILE.parent.mkdir(parents=True, exist_ok=True)

    # Проблема #1: требовать ack-файл от пользователя (подтверждение плана)
    ack_file = PROGRESS_FILE.parent / f"step-{step.get('id')}-PLAN.ack"
    if require_ack and not ack_file.exists():
        print(f"ERROR: plan not acknowledged. "
              f"Run --gen-plan, show to user, then create {ack_file} "
              f"after user says 'выполняй'.", file=sys.stderr)
        sys.exit(4)

    # Проблема #15: защита от случайного перезаписи существующего файла
    if PROGRESS_FILE.exists() and not force:
        try:
            existing = yaml.safe_load(PROGRESS_FILE.read_text(encoding='utf-8'))
        except Exception:
            existing = None
        if existing and existing.get('step_id') == step.get('id'):
            print(f"ERROR: progress file for step '{step.get('id')}' already exists "
                  f"(current_phase={existing.get('current_phase')}). "
                  f"This is a RESUME scenario — use --show-progress to inspect, "
                  f"do NOT re-init. To force re-init, use --force.",
                  file=sys.stderr)
            sys.exit(6)
        if existing and existing.get('step_id') != step.get('id'):
            print(f"ERROR: progress file exists for step '{existing.get('step_id')}', "
                  f"cannot init for '{step.get('id')}'. "
                  f"Move or rename _meta/step-progress.yaml first.",
                  file=sys.stderr)
            sys.exit(6)

    now = datetime.now().strftime('%Y-%m-%d %H:%M')
    progress = {
        'step_id': step.get('id'),
        'step_title': step.get('title'),
        'branch': step.get('branch'),
        'docs_target': step.get('docs_target'),
        'step_type': step.get('step_type'),
        'started_at': now,
        'current_phase': 'PREFLIGHT',
        'finished_at': None,
        'phases': {},
        'loop_detector': {
            'last_action': None,
            'last_action_count': 0,
            'action_history': [],
        },
        'notes': [],
    }

    for phase in PHASES:
        # Для step_type=document фаза IMPLEMENTATION пропускается
        if phase == 'IMPLEMENTATION' and step.get('step_type') == 'document':
            progress['phases'][phase] = {
                'status': 'skipped',
                'started_at': None,
                'finished_at': None,
                'attempts': 0,
                'last_error': 'skipped: step_type=document',
            }
        else:
            progress['phases'][phase] = {
                'status': 'pending',
                'started_at': None,
                'finished_at': None,
                'attempts': 0,
                'last_error': None,
            }

    with open(PROGRESS_FILE, 'w', encoding='utf-8') as f:
        yaml.safe_dump(progress, f, allow_unicode=True, sort_keys=False, default_flow_style=False)

    print(f"Created: {PROGRESS_FILE}")
    print(f"Current phase: {progress['current_phase']}")
    if ack_file.exists():
        print(f"Acknowledged via: {ack_file}")


def _validate_phase_completion(step_id: str, phase: str, progress: dict) -> str | None:
    """Проблема #3: валидация фактического выполнения фазы перед отметкой completed.

    Возвращает строку с ошибкой, если фаза не может быть completed.
    """
    import subprocess

    def _git(args: list[str]) -> str:
        try:
            return subprocess.run(['git'] + args, capture_output=True, text=True,
                                  cwd=BASE, timeout=10).stdout.strip()
        except Exception:
            return ''

    if phase == 'PREFLIGHT':
        # Должна быть активна feature-ветка
        branch = _git(['branch', '--show-current'])
        expected = progress.get('branch', '')
        if not branch.startswith('feature/step-'):
            return (f"PREFLIGHT cannot be completed: current branch is '{branch}', "
                    f"expected feature-branch '{expected}'. "
                    f"Run step-pre-flight.md first to create the branch.")
        if expected and branch != expected:
            return (f"PREFLIGHT cannot be completed: branch mismatch "
                    f"(actual='{branch}', expected='{expected}')")

    elif phase == 'PHASE1':
        docs_target = progress.get('docs_target', '')
        if not docs_target or not (BASE / docs_target).exists():
            return (f"PHASE1 cannot be completed: docs_target '{docs_target}' does not exist. "
                    f"Run docs-writer.md first.")
        content = (BASE / docs_target).read_text(encoding='utf-8')
        if '> Phase 1 — инструкция для AI-агента' not in content:
            return (f"PHASE1 cannot be completed: Phase 1 marker missing in {docs_target}. "
                    f"docs-writer.md did not write Phase 1 documentation.")

    elif phase == 'TESTING':
        # verify должен быть запущен и зафиксирован в реестре тестов
        docs_target = progress.get('docs_target', '')
        if docs_target and (BASE / docs_target).exists():
            content = (BASE / docs_target).read_text(encoding='utf-8')
            if 'PASS' not in content and 'PASSED' not in content:
                return (f"TESTING cannot be completed: no PASS/PASSED record in {docs_target}. "
                        f"Run test-runner.md first.")

    elif phase == 'FINALIZATION':
        # Должны быть на main, feature-ветка удалена
        branch = _git(['branch', '--show-current'])
        if branch != 'main':
            return (f"FINALIZATION cannot be completed: not on main (current='{branch}'). "
                    f"Run step-finalizer.md to switch to main after squash-merge.")
        expected = progress.get('branch', '')
        branches = _git(['branch', '--list', expected])
        if branches:
            return (f"FINALIZATION cannot be completed: feature-branch '{expected}' still exists. "
                    f"step-finalizer.md should delete it after squash-merge.")

    return None


def load_progress() -> dict | None:
    """Загрузить progress-файл. Если нет — вернуть None."""
    if not PROGRESS_FILE.exists():
        return None
    return yaml.safe_load(PROGRESS_FILE.read_text(encoding='utf-8'))


def save_progress(progress: dict) -> None:
    """Сохранить progress-файл."""
    with open(PROGRESS_FILE, 'w', encoding='utf-8') as f:
        yaml.safe_dump(progress, f, allow_unicode=True, sort_keys=False, default_flow_style=False)


def set_phase(step_id: str, phase: str, status: str) -> None:
    """Обновить статус фазы в progress-файле."""
    if phase not in PHASES:
        print(f"ERROR: invalid phase '{phase}'. Valid: {', '.join(PHASES)}", file=sys.stderr)
        sys.exit(3)
    if status not in VALID_STATUSES:
        print(f"ERROR: invalid status '{status}'. Valid: {', '.join(VALID_STATUSES)}", file=sys.stderr)
        sys.exit(3)

    progress = load_progress()
    if progress is None:
        print("ERROR: progress file does not exist. Run --init-progress first.", file=sys.stderr)
        sys.exit(2)

    if progress.get('step_id') != step_id:
        print(f"ERROR: progress file is for step '{progress.get('step_id')}', not '{step_id}'",
              file=sys.stderr)
        sys.exit(2)

    # Проблема #3: валидация фактического выполнения фазы
    if status == 'completed':
        err = _validate_phase_completion(step_id, phase, progress)
        if err:
            print(f"ERROR: {err}", file=sys.stderr)
            sys.exit(5)

    now = datetime.now().strftime('%Y-%m-%d %H:%M')
    progress['phases'][phase]['status'] = status
    if status == 'in_progress':
        progress['phases'][phase]['started_at'] = now
        progress['phases'][phase]['attempts'] = progress['phases'][phase].get('attempts', 0) + 1
    elif status in ('completed', 'failed', 'skipped'):
        progress['phases'][phase]['finished_at'] = now
        if status == 'completed' and phase != 'FINALIZATION':
            # Перейти к следующей фазе
            idx = PHASES.index(phase)
            if idx + 1 < len(PHASES):
                next_phase = PHASES[idx + 1]
                # Пропустить skipped фазы
                while next_phase in progress['phases'] and progress['phases'][next_phase]['status'] == 'skipped':
                    idx2 = PHASES.index(next_phase)
                    if idx2 + 1 < len(PHASES):
                        next_phase = PHASES[idx2 + 1]
                    else:
                        next_phase = None
                        break
                if next_phase:
                    progress['current_phase'] = next_phase
        elif phase == 'FINALIZATION' and status == 'completed':
            progress['current_phase'] = 'DONE'
            progress['finished_at'] = now
    elif status == 'failed':
        progress['current_phase'] = 'FAILED'

    save_progress(progress)
    print(f"Updated: phase={phase}, status={status}")
    print(f"Current phase: {progress['current_phase']}")


def record_action(step_id: str, action: str) -> None:
    """Проблема #8: loop detector — записать действие и проверить петлю."""
    progress = load_progress()
    if progress is None:
        print("ERROR: no progress file. Run --init-progress first.", file=sys.stderr)
        sys.exit(2)

    ld = progress.setdefault('loop_detector', {})
    history = ld.setdefault('action_history', [])
    history.append(action)
    # Хранить только последние 10
    if len(history) > 10:
        history = history[-10:]
        ld['action_history'] = history

    # Проверка петли: 3+ одинаковых подряд
    loop_detected = False
    loop_msg = ''
    if len(history) >= 3 and len(set(history[-3:])) == 1:
        loop_detected = True
        loop_msg = f"action '{history[-1]}' repeated 3+ times in a row"
    else:
        # Также: тот же readFile 3+ раза за последние 10
        from collections import Counter
        reads = [a for a in history if a.startswith('readFile:')]
        if reads:
            counter = Counter(reads)
            most_common, count = counter.most_common(1)[0]
            if count >= 3:
                loop_detected = True
                loop_msg = f"'{most_common}' recorded {count} times in last {len(history)} actions"

    if loop_detected:
        ld['last_action'] = history[-1] if history else None
        ld['last_action_count'] = 3
        save_progress(progress)
        print(f"LOOP DETECTED: {loop_msg}", file=sys.stderr)
        print(f"Action history (last {len(history)}):", file=sys.stderr)
        for i, a in enumerate(history, 1):
            print(f"  {i}. {a}", file=sys.stderr)
        print("STOP. See 08-antipatterns.md for recovery.", file=sys.stderr)
        sys.exit(7)

    ld['last_action'] = history[-1] if history else None
    ld['last_action_count'] = 1
    save_progress(progress)
    print(f"Recorded: {action}")


def check_loop(step_id: str) -> None:
    """Проблема #8: проверить loop_detector без записи действия."""
    progress = load_progress()
    if progress is None:
        print("OK: no progress file")
        sys.exit(0)
    ld = progress.get('loop_detector', {})
    count = ld.get('last_action_count', 0)
    if count >= 3:
        print(f"LOOP DETECTED: last_action_count={count}", file=sys.stderr)
        sys.exit(7)
    print(f"OK: no loop detected (last_action_count={count})")


def show_progress(step_id: str) -> None:
    """Показать текущий прогресс."""
    progress = load_progress()
    if progress is None:
        print("ERROR: progress file does not exist.", file=sys.stderr)
        sys.exit(2)

    print(f"Step: {progress.get('step_id')} — {progress.get('step_title')}")
    print(f"Branch: {progress.get('branch')}")
    print(f"Started: {progress.get('started_at')}")
    print(f"Current phase: {progress.get('current_phase')}")
    print(f"Finished: {progress.get('finished_at', '—')}")
    print()
    print("Phases:")
    for phase in PHASES:
        ph = progress.get('phases', {}).get(phase, {})
        status = ph.get('status', 'pending')
        icon = {'completed': '✓', 'in_progress': '→', 'failed': '✗',
                'skipped': '—', 'pending': ' '}[status]
        attempts = ph.get('attempts', 0)
        attempts_str = f" (попытка {attempts})" if attempts > 0 else ""
        print(f"  [{icon}] {phase:15} {status:11}{attempts_str}")
        if ph.get('last_error'):
            print(f"      last_error: {ph['last_error']}")

    # Проблема #17: показывать action_history
    print()
    print("Loop detector:")
    ld = progress.get('loop_detector', {})
    print(f"  last_action: {ld.get('last_action', '—')}")
    print(f"  last_action_count: {ld.get('last_action_count', 0)}")
    history = ld.get('action_history', [])
    if history:
        print(f"  action_history (last {len(history)}):")
        for i, action in enumerate(history, 1):
            print(f"    {i}. {action}")
    else:
        print("  action_history: (empty)")
    notes = progress.get('notes', [])
    if notes:
        print()
        print(f"Notes ({len(notes)}):")
        for n in notes:
            print(f"  - {n}")


def can_start_phase(step_id: str, phase: str) -> None:
    """Проверить, можно ли запустить фазу (все предыдущие completed/skipped)."""
    if phase not in PHASES:
        print(f"ERROR: invalid phase '{phase}'", file=sys.stderr)
        sys.exit(3)

    progress = load_progress()
    if progress is None:
        print("ERROR: progress file does not exist.", file=sys.stderr)
        sys.exit(2)

    idx = PHASES.index(phase)
    if idx == 0:
        print(f"OK: {phase} — первая фаза, можно запускать.")
        sys.exit(0)

    # Проверить все предыдущие фазы
    for prev_phase in PHASES[:idx]:
        prev_status = progress.get('phases', {}).get(prev_phase, {}).get('status', 'pending')
        if prev_status not in ('completed', 'skipped'):
            print(f"BLOCKED: предыдущая фаза {prev_phase} не завершена (status={prev_status}).",
                  file=sys.stderr)
            sys.exit(3)

    print(f"OK: {phase} — все предыдущие фазы завершены, можно запускать.")
    sys.exit(0)


def generate_plan(step: dict) -> str:
    """Сгенерировать шаблон плана docs/NN-stage/step-X-Y-PLAN.md."""
    step_id = step.get('id')
    title = step.get('title', '')
    stage = step.get('_stage', '')
    stage_name = step.get('_stage_name', '')
    docs_target = step.get('docs_target', '')
    verify = step.get('verify', [''])[0]
    expected_result = step.get('expected_result', '')
    step_type = step.get('step_type', 'implementation')
    branch = step.get('branch', '')
    context_to_read = step.get('context_to_read', [])
    requires_approval = step.get('requires_approval', False)
    parallel_group = step.get('parallel_group', '')

    now = datetime.now().strftime('%Y-%m-%d %H:%M')

    # Адаптировать фазы под step_type
    if step_type == 'document':
        impl_phase = '- [~] **IMPLEMENTATION** — пропускается (step_type=document)'
    else:
        impl_phase = '- [ ] **IMPLEMENTATION** — реализовать код/файлы (code-writer.md)'

    risks = []
    if requires_approval:
        risks.append('- ⚠️ Требует approval gate (см. step-pre-flight.md §0.2)')
    if parallel_group:
        risks.append(f'- ⚠️ Параллельная группа `{parallel_group}` — может потребоваться rebase')
    if 'npm' in verify or 'uv run' in verify:
        risks.append('- Verify запускает внешние инструменты — может упасть на CI')
    if not risks:
        risks.append('- Стандартный шаг без особых рисков')

    context_lines = '\n'.join(f"  {i+1}. {ctx}" for i, ctx in enumerate(context_to_read))

    # Проблема #16: добавить секцию «ОЖИДАНИЕ ПОДТВЕРЖДЕНИЯ»
    plan = f"""# План шага {step_id} — {title}

> Сгенерировано: {now}
> Команда пользователя: выполни шаг {step_id}
> Stage: {stage} — {stage_name}

## ⏸ ОЖИДАНИЕ ПОДТВЕРЖДЕНИЯ (gate: plan-approval)

**Этот план НЕ будет выполнен без явной команды пользователя.**

После прочтения плана скажите:
  "выполняй"  — запустить все 7 фаз по порядку
  "правки"    — я отредактирую план, затем скажу "выполняй"
  "стоп"      — отменить выполнение шага

Пока вы не ответили, агент НЕ создаёт `_meta/step-progress.yaml`,
НЕ создаёт feature-ветку, НЕ запускает субагентов.

## Контекст шага

- **Тип:** {step_type}
- **Ветка:** {branch}
- **docs_target:** {docs_target}
- **Verify:** `{verify}`
- **Ожидаемый результат:** {expected_result}

## Фазы выполнения (state machine)

- [ ] **PREFLIGHT** — создать feature-ветку, проверить контекст (step-pre-flight.md)
- [ ] **PHASE1** — написать документацию для агента (docs-writer.md)
{impl_phase}
- [ ] **TESTING** — запустить verify и тесты (test-runner.md)
- [ ] **PHASE2A** — переписать новые секции для человека (docs-rewriter-human.md §1-2)
- [ ] **PHASE2B** — проверить сохраняемые секции (docs-rewriter-human.md §3)
- [ ] **FINALIZATION** — git commit, squash-merge, push (step-finalizer.md)

## Контекст для чтения (субагенты прочитают сами)

{context_lines}

## Риски шага

{chr(10).join(risks)}

## Прогресс

После запуска выполнения прогресс отслеживается в `_meta/step-progress.yaml`.
Команда просмотра: `uv run python scripts/orchestrate_step.py {step_id} --show-progress`
"""

    return plan


def main():
    parser = argparse.ArgumentParser(description='Step orchestrator helper')
    parser.add_argument('step_id', help='Идентификатор шага, например 1.6')
    parser.add_argument('--print-step', action='store_true',
                        help='Напечатать краткую инфу о шаге из YAML')
    parser.add_argument('--init-progress', action='store_true',
                        help='Создать _meta/step-progress.yaml (требует ack-файл)')
    parser.add_argument('--force', action='store_true',
                        help='Принудительно перезаписать progress-файл (только для recovery)')
    parser.add_argument('--no-ack', action='store_true',
                        help='Не требовать ack-файл (только для тестов/CI)')
    parser.add_argument('--set-phase', nargs=2, metavar=('PHASE', 'STATUS'),
                        help='Обновить статус фазы (с валидацией фактического выполнения)')
    parser.add_argument('--show-progress', action='store_true',
                        help='Показать текущий прогресс + loop_detector action_history')
    parser.add_argument('--can-start', metavar='PHASE',
                        help='Проверить, можно ли запустить фазу')
    parser.add_argument('--gen-plan', action='store_true',
                        help='Сгенерировать шаблон плана на stdout')
    # Проблема #8: loop detector команды
    parser.add_argument('--record-action', metavar='ACTION',
                        help='Записать действие в loop_detector.action_history и проверить петлю')
    parser.add_argument('--check-loop', action='store_true',
                        help='Проверить loop_detector, exit 0=OK, exit 7=loop detected')

    args = parser.parse_args()

    # Найти шаг в YAML (для большинства команд нужно)
    step = find_step_in_yaml(args.step_id)
    if step is None:
        print(f"ERROR: step '{args.step_id}' not found in any plans/*.yaml",
              file=sys.stderr)
        sys.exit(1)

    if args.print_step:
        print_step(step)
    elif args.init_progress:
        init_progress_file(step, force=args.force, require_ack=not args.no_ack)
    elif args.set_phase:
        set_phase(args.step_id, args.set_phase[0], args.set_phase[1])
    elif args.show_progress:
        show_progress(args.step_id)
    elif args.can_start:
        can_start_phase(args.step_id, args.can_start)
    elif args.gen_plan:
        print(generate_plan(step))
    elif args.record_action:
        record_action(args.step_id, args.record_action)
    elif args.check_loop:
        check_loop(args.step_id)
    else:
        parser.print_help()
        sys.exit(2)


if __name__ == '__main__':
    main()