scripts/finalize_step.py - main - ump-team/ump-infra

scripts/finalize_step.py
901 строка44 KB
Dmitry Kochenov
fix: исправить 46 ошибок basedpyright и добавить type hint stubs
09 авг 2026, 15:25
09 авг 2026, 15:258a46eb7
#!/usr/bin/env python3
"""
Автоматический финализатор шага.

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

Выполняет:
1. Проверка docs_target (Phase 2, VERIFIED_CHECKLIST, история ≥ 5)
2. Многосторонняя проверка через verify_step_completion.py
3. Git: commit в feature-ветке → push → создать PR
   ⚠ Merge в main НЕ делается автоматически (политика v31) —
   пользователь merge'ит через GitVerse web UI, затем говорит «merge готов»
4. Обновление STEP_STATE.md (CURRENT_PHASE=awaiting-merge, COMPLETED_STEPS)
5. Feature-ветка НЕ удаляется (для истории и возможного rollback)
6. Обновление снимка файлов (detect_manual_changes.py --update-snapshot)
7. Логирование всех действий

После merge пользователь запускает: make diagnostics STEP=X.Y
(пост-merge диагностика с сохранением лога в _meta/diagnostics/)

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

Exit codes:
    0 — шаг финализирован успешно (PR создан, ожидание ручного merge)
    1 — ошибка (проверки не пройдены, git конфликт, и т.д.)
    2 — шаг не найден
"""
from __future__ import annotations

import argparse
import glob as glob_mod
import os
import re
import subprocess
import sys
from datetime import datetime
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'
# Runtime state (progress, agent.log, snapshots) — ВСЕГДА в BASE/_meta/,
# потому что это состояние шагов пользователя, не infra-константы.
META_DIR = BASE / '_meta'
# Infra-константы (rules-constants.yaml, immutable-snapshot.json, schema) — в INFRA_DIR/_meta/.
INFRA_META_DIR = INFRA_DIR / '_meta' if (INFRA_DIR / '_meta').exists() else BASE / '_meta'

# Импорт функций из agent_log — в submodule-layout лежит в INFRA_DIR/scripts/.
# P0-fix: ранее здесь было `else _scripts_dir` — NameError, если INFRA_DIR/scripts не существует.
# Используем честный fallback: сначала пробуем INFRA_DIR/scripts, затем BASE/scripts,
# затем каталог самого finalize_step.py.
_scripts_dir = INFRA_DIR / 'scripts'
if not _scripts_dir.exists():
    _scripts_dir = BASE / 'scripts'
if not _scripts_dir.exists():
    _scripts_dir = Path(__file__).resolve().parent
sys.path.insert(0, str(_scripts_dir))
import contextlib

from agent_log import add_entry  # noqa: E402

# P0-7 fix: безопасные обёртки над subprocess.run без shell=True.
from subprocess_utils import run_list, run_str  # noqa: E402

# P0-8 fix: enforcement permission-policy.yaml.
# Stubs defined BEFORE try/except so basedpyright sees a single variable type.
from dataclasses import dataclass
from enum import Enum

class _PermissionModeStub2(str, Enum):
    ALLOW = 'allow'
    ASK = 'ask'
    DENY = 'deny'

@dataclass
class _DecisionStub2:
    action: str = ''
    mode: _PermissionModeStub2 = _PermissionModeStub2.ALLOW
    reason: str = 'permission_policy not found'
    policy_section: str = ''

    @property
    def is_allowed(self) -> bool:
        return True

    @property
    def needs_confirmation(self) -> bool:
        return False

def _check_and_enforce_stub2(action: str, phase: str | None = None, *, policy: dict | None = None) -> _DecisionStub2:
    return _DecisionStub2(action=action)

def _check_permission_stub2(action: str, phase: str | None = None) -> _DecisionStub2:
    return _DecisionStub2(action=action)

try:
    from permission_policy import (
        PERMISSION_DENIED_EXIT_CODE,
        PermissionDenied,
        check_and_enforce as _real_check_and_enforce,
        check_permission as _real_check_permission,
    )
except ImportError:
    # permission_policy.py недоступен — enforcement не работает.
    PERMISSION_DENIED_EXIT_CODE = 13
    PermissionDenied = Exception  # type: ignore[assignment, misc]

check_and_enforce = _real_check_and_enforce  # type: ignore[assignment]
check_permission = _real_check_permission  # type: ignore[assignment]


def find_step(step_id: str) -> tuple[dict, str, dict | None] | None:
    """Найти шаг в YAML. Вернуть (step_dict, yaml_filename, next_step_dict).

    P2-4 fix (mypy): аннотация исправлена — next_step это dict (или None),
    не str (как было раньше — это была ошибка в исходной аннотации).
    """
    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 (OSError, yaml.YAMLError):
            # P1-15 fix: точечный except — OSError для файла, YAMLError для парсинга.
            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-команду.

    P0-7 fix: ранее использовался subprocess.run(cmd, shell=True, ...), что
    позволяло инъекции через интерполируемые переменные (branch name из YAML,
    file path из plans, и т.д.). Теперь — через subprocess_utils.run_str,
    который парсит cmd через shlex.split() и вызывает subprocess.run в list-form.

    Если в cmd есть shell-метасимволы (|, >, ||, *, ...), run_str печатает
    WARNING и fallback на shell=True. Весь код UMP переписан так, чтобы
    избегать таких конструкций — см. git_*() ниже (используют run_list).
    """
    if cwd is None:
        cwd = BASE
    return run_str(cmd, check=check, cwd=cwd, timeout=120, capture=True)


def atomic_write_text(filepath: Path, content: str) -> None:
    """FIX-2.4: атомарная запись текстового файла через temp + os.replace."""
    import tempfile
    fd, tmp_path = tempfile.mkstemp(dir=str(filepath.parent), prefix='.tmp-', suffix=filepath.suffix or '.tmp')
    try:
        with os.fdopen(fd, 'w', encoding='utf-8') as f:
            f.write(content)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp_path, filepath)
    except Exception:
        with contextlib.suppress(OSError):
            os.unlink(tmp_path)
        raise


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.


    Phase 2 маркер и 11 обязательных секций. Достаточно:
      - VERIFIED_CHECKLIST закрыт
      - История ≥ 3 записей
    """
    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')

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

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

        # История ≥ 3 записей
        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) < 3:
                return False, f'история действий: {len(data_rows)} записей (нужно ≥ 3 для bootstrap)'
        else:
            return False, 'секция История действий не найдена'

        return True, 'OK (bootstrap: VERIFIED_CHECKLIST + история ≥ 3)'

    # Для остальных шагов — полная проверка
    # Phase 2 маркер
    if '> Phase 1 — инструкция для AI-агента' in content and '> Обучающая документация' not in content:
        return False, '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_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 _prepare_safe_checkout_main(step_id: str) -> None:
    """: обработать untracked immutable-snapshot.json перед `git checkout main`.

    Проблема: `_meta/immutable-snapshot.json` коммитится в репозитории (v26.1 fix),
    но feature-ветка может не иметь коммита с этим файлом. Тогда git checkout main
    FAIL с "untracked files would be overwritten". Также файл мог быть модифицирован
    protect_files.py --update во время выполнения шага.

    Решение: перед checkout main — если файл tracked в текущей ветке, сбросить его
    к HEAD-версии (`git checkout -- <file>`); если untracked — удалить локально,
    чтобы main восстановил его из своего коммита.
    """
    snapshot_rel = '_meta/immutable-snapshot.json'
    snapshot_path = META_DIR / 'immutable-snapshot.json'
    # P0-7 fix: list-form без shell=True.
    r = run_list(['git', 'ls-files', '--error-unmatch', snapshot_rel], check=False, cwd=BASE)
    if r.returncode == 0:
        # Файл tracked — git checkout -- восстановит HEAD-версию
        run_list(['git', 'checkout', '--', snapshot_rel], check=False, cwd=BASE)
        log('immutable-snapshot.json (tracked) сброшен к HEAD для safe checkout', step_id, 'action')
    else:
        # Файл untracked — удалить локально, main восстановит из своего коммита
        if snapshot_path.exists():
            snapshot_path.unlink()
            log('immutable-snapshot.json (untracked) удалён для safe checkout', step_id, 'action')


def _ensure_feature_branch(step_id: str, step_branch: str) -> tuple[bool, str]:
    """Убедиться, что мы на feature-ветке. Создать при необходимости.

    P2-refactor (v0.0.5 audit): вынесено из git_operations() для снижения CC.

    Returns:
        (True, '') — на feature-ветке.
        (False, msg) — не удалось переключиться/создать.
    """
    branch = run_list(['git', 'branch', '--show-current'], cwd=BASE).stdout.strip()
    if branch == step_branch:
        return True, ''
    log(f'Текущая ветка: {branch}, нужно: {step_branch}', step_id, 'warning')

    # Проверить, существует ли feature-ветка
    r = run_list(['git', 'rev-parse', '--verify', step_branch], check=False, cwd=BASE)
    if r.returncode == 0:
        # Ветка существует — переключаемся
        log(f'Переключение на {step_branch}', step_id, 'action')
        run_list(['git', 'stash'], check=False, cwd=BASE)
        r = run_list(['git', 'checkout', step_branch], cwd=BASE)
        run_list(['git', 'stash', 'pop'], check=False, cwd=BASE)
        if r.returncode != 0:
            return False, f'Не удалось переключиться на {step_branch}: {r.stderr}'
    else:
        # Ветка не существует — создаём от main
        log(f'Создание {step_branch} от main', step_id, 'action')
        run_list(['git', 'stash'], check=False, cwd=BASE)
        r = run_list(['git', 'rev-parse', '--verify', 'main'], check=False, cwd=BASE)
        if r.returncode == 0:
            _prepare_safe_checkout_main(step_id)
            run_list(['git', 'checkout', 'main'], cwd=BASE)
        run_list(['git', 'checkout', '-b', step_branch], cwd=BASE)
        run_list(['git', 'stash', 'pop'], check=False, cwd=BASE)

    # Проверить повторно
    branch = run_list(['git', 'branch', '--show-current'], cwd=BASE).stdout.strip()
    if branch != step_branch:
        return False, f'Не на feature-ветке (current={branch}, expected={step_branch})'
    return True, ''


def _unstage_protected_files(step_id: str) -> None:
    """Снять защищённые файлы из git index (кроме STEP_STATE.md).

    P2-refactor (v0.0.5 audit): вынесено из git_operations().
    """
    sys.path.insert(0, str(_scripts_dir))
    try:
        from protect_files import load_immutable_list
        protected_to_unstage = load_immutable_list()
    except ImportError:
        protected_to_unstage = [
            '.agent/rules/', '.agent/prompts/', '.agent/cards/', '.agent/immutable.txt',
            'scripts/', 'plans/',
            '00-manifest.md', 'AGENTS.md', 'Makefile', 'pyproject.toml',
            'package.json', '.gitignore', '.nvmrc', 'STEP_STATE.template.md',
        ]
    staged_result = run_list(['git', 'diff', '--cached', '--name-only'], cwd=BASE)
    staged = staged_result.stdout.strip().split('\n') if staged_result.stdout.strip() else []
    unstaged_count = 0
    for f in staged:
        for prot in protected_to_unstage:
            if f.startswith(prot) and f != 'STEP_STATE.md':
                run_list(['git', 'reset', 'HEAD', '--', f], check=False, cwd=BASE)
                unstaged_count += 1
                break
    if unstaged_count > 0:
        log(f'Unstaged {unstaged_count} protected files', step_id, 'action')


def _remove_runtime_index_files(step_id: str) -> None:
    """Удалить runtime-файлы из git index (кроме STEP_STATE.md).

    P2-refactor (v0.0.5 audit): вынесено из git_operations().
    """
    _runtime_index_patterns = [
        '_meta/agent.log', '_meta/agent.log.*',
        '_meta/agent-snapshot.json', '_meta/manual-changes.log',
        '_meta/step-progress.yaml', '_meta/step-progress-*.yaml',
        '_meta/step-progress-*.done.yaml', '_meta/.subagent-lock-*',
        '_meta/step-*.ack', '_meta/commit-msg-*.txt',
        '_meta/user-settings.local.yaml', '_meta/errors.json',
        '_meta/errors.log', '_meta/immutable-snapshot.json',
        '_meta/.allow-protected-edit',
    ]
    for pattern in _runtime_index_patterns:
        if '*' in pattern or '?' in pattern or '[' in pattern:
            matches = glob_mod.glob(str(BASE / pattern))
            for matched_path in matches:
                rel = os.path.relpath(matched_path, BASE)
                run_list(['git', 'rm', '--cached', '--', rel], check=False, cwd=BASE)
        else:
            run_list(['git', 'rm', '--cached', '--', pattern], check=False, cwd=BASE)


def _add_step_files(step: dict, step_id: str) -> tuple[bool, str]:
    """git add docs_target + files_to_commit (с проверкой immutable).

    P2-refactor (v0.0.5 audit): вынесено из git_operations().
    """
    docs_target = step.get('docs_target', '')
    files_to_commit = step.get('files_to_commit', [])

    # Добавить docs_target
    if docs_target and (BASE / docs_target).exists():
        r = run_list(['git', 'add', '--', docs_target], check=False, cwd=BASE)
        if r.returncode == 0:
            log(f'git add {docs_target}', step_id, 'action')
        else:
            log(f'git add {docs_target} failed: {r.stderr[:200]}', step_id, 'warning')

    # Проверить files_to_commit на immutable
    sys.path.insert(0, str(_scripts_dir))
    try:
        from protect_files import expand_paths, load_immutable_list
        immutable_files = {str(f.relative_to(BASE)) for f in expand_paths(load_immutable_list())}
        for f in files_to_commit:
            if f in immutable_files:
                log(f'REFUSED: file_to_commit {f} is immutable!', step_id, 'error')
                return False, f'file_to_commit {f} is immutable — cannot finalize step that modifies protected file via YAML'
    except ImportError:
        pass

    # Добавить files_to_commit
    for f in files_to_commit:
        if (BASE / f).exists():
            r = run_list(['git', 'add', '--', f], check=False, cwd=BASE)
            if r.returncode == 0:
                log(f'git add {f} (из files_to_commit)', step_id, 'action')
    return True, ''


def _git_commit_step(step: dict, step_id: str) -> tuple[bool, str]:
    """git commit в feature-ветке с permission enforcement.

    P2-refactor (v0.0.5 audit): вынесено из git_operations().
    """
    try:
        check_and_enforce('git_commit', phase='FINALIZATION')
    except PermissionDenied as e:
        log(f'git_commit denied: {e}', step_id, 'error')
        return False, f'PERMISSION_DENIED: git_commit — {e}'

    step_title = step.get('title', '')
    commit_msg = f'feat(step-{step_id}): {step_title}\n\nДокументация: {step.get("docs_target", "")}'
    msg_file = META_DIR / 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('git commit в feature-ветке', step_id, 'commit')
    r = run_list(['git', 'commit', '-F', str(msg_file)], cwd=BASE)
    if r.returncode != 0 and 'nothing to commit' not in r.stdout and 'nothing to commit' not in r.stderr:
        log('git commit -F failed, trying -m', step_id, 'warning')
        short_msg = f'feat(step-{step_id}): {step_title}'
        r = run_list(['git', 'commit', '-m', short_msg], cwd=BASE)
        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')
    return True, ''


def _push_feature_branch(step_id: str, step_branch: str) -> tuple[bool, str]:
    """Push feature-ветки с fallback на --set-upstream.

    P2-refactor (v0.0.5 audit): вынесено из git_operations().
    """
    try:
        check_and_enforce('git_push', phase='FINALIZATION')
    except PermissionDenied as e:
        log(f'git_push denied: {e}', step_id, 'error')
        return False, f'PERMISSION_DENIED: git_push — {e}'
    log(f'git push origin {step_branch}', step_id, 'action')
    r = run_list(['git', 'push', 'origin', step_branch], check=False, cwd=BASE)
    if r.returncode != 0:
        r = run_list(['git', 'push', '--set-upstream', 'origin', step_branch], check=False, cwd=BASE)
        if r.returncode != 0:
            log(f'git push failed: {r.stderr[:200]}', step_id, 'error')
            log('Продолжаем без push (возможно нет remote)', step_id, 'warning')
        else:
            log('git push OK (set-upstream)', step_id, 'action')
    else:
        log('git push OK', step_id, 'action')
    return True, ''


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

    P2-refactor (v0.0.5 audit): разбита на 6 helper-функций для снижения CC.
    Ранее CC=55 — теперь функция оркеструет вызовы helpers.

    - Один шаг = одна ветка = один PR = РУЧНОЙ merge в main через GitVerse web UI
    - Если агент не на feature-ветке — переключиться на неё
    - После тестов → commit → push feature-ветки → создать PR
     - ⚠ Merge в main НЕ делается автоматически — пользователь merge'ит через
      GitVerse web UI.
     """

    # 1. Убедиться, что мы на feature-ветке
    ok, msg = _ensure_feature_branch(step_id, step_branch)
    if not ok:
        return False, msg

    # 2. Unstage protected + runtime files
    log('Умный git add (только файлы шага)', step_id, 'action')
    _unstage_protected_files(step_id)
    _remove_runtime_index_files(step_id)

    # 3. Добавить docs_target + files_to_commit
    ok, msg = _add_step_files(step, step_id)
    if not ok:
        return False, msg

    # 4. git commit
    ok, msg = _git_commit_step(step, step_id)
    if not ok:
        return False, msg

    # 5. Push feature-ветки
    ok, msg = _push_feature_branch(step_id, step_branch)
    if not ok:
        return False, msg

    # 6. Создать PR (через gh CLI)
    pr_created, pr_url, pr_msg = _create_pull_request(step, step_id, step_branch)
    if pr_msg:
        log(pr_msg, step_id, 'action')
    return True, pr_url if pr_created else 'OK'


def _create_pull_request(step: dict, step_id: str, step_branch: str) -> tuple[bool, str, str]:
    """Создать PR через gh CLI. Возвращает (created, url, message).

    P2-refactor (v0.0.5 audit): вынесено из git_operations().
    """
    log(f'Создание PR для шага {step_id}', step_id, 'action')
    # Проверить, есть ли уже PR
    r = run_list(
        ['gh', 'pr', 'list', '--head', step_branch, '--json', 'number', '--jq', '.[0].number'],
        check=False, cwd=BASE,
    )
    existing_pr = r.stdout.strip()
    if existing_pr and existing_pr.isdigit():
        return True, f'#{existing_pr}', f'PR уже существует: #{existing_pr} (переиспользуем)'
    # Создать новый PR
    step_title = step.get('title', '')
    pr_body = f'Step {step_id}: {step_title}'
    r = run_list(
        ['gh', 'pr', 'create', '--title', f'step-{step_id}: {step_title}', '--body', pr_body, '--head', step_branch],
        check=False, cwd=BASE,
    )
    if r.returncode == 0 and r.stdout.strip():
        return True, r.stdout.strip(), f'PR создан: {r.stdout.strip()}'
    return False, '', f'gh pr create failed: {r.stderr[:200]}'


def _update_current_step_fields(
    content: str, step_id: str, step: dict, next_step: dict | None
) -> str:
    """Обновить CURRENT_STEP / LAST_COMPLETED_STEP / CURRENT_PHASE в STEP_STATE.md.

    P2-refactor (v0.0.5 audit): вынесено из update_step_state.
    """
    # 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:
        next_step_id = next_step.get('id', step_id)
        next_step_docs = next_step.get('docs_target', '')
        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)
    return content


def _add_completed_step_row(
    content: str, step_id: str, step: dict, now: str
) -> str:
    """Добавить строку в COMPLETED_STEPS (идемпотентно).

    P2-refactor (v0.0.5 audit): вынесено из update_step_state.
    """
    if f'| {step_id} |' in content:
        log(f'STEP_STATE.md уже содержит запись о шаге {step_id} — пропуск', step_id, 'warning')
        return content
    step_title = step.get('title', '')
    docs_target = step.get('docs_target', '')
    new_row = f'| {step_id} | {step_title} | {now} | [{docs_target}]({docs_target}) |'
    content = re.sub(
        r'(\| Шаг \| Название \| Дата \| Документация \|\n\|-----\|----------\|------\|--------------\|\n)',
        rf'\1{new_row}\n',
        content,
        count=1,
    )
    return content


def _extract_adr_files_from_step(docs_target_path: Path) -> set:
    """Найти ADR-файлы, упомянутые в docs_target шага.

    P2-refactor (v0.0.5 audit): вынесено из update_step_state.
    """
    if not docs_target_path or not docs_target_path.exists():
        return set()
    try:
        dt_content = docs_target_path.read_text(encoding='utf-8')
    except Exception:
        return set()
    adr_pattern = re.compile(r'docs/для-разработчиков/adr/(\d{4})-([a-z0-9-]+)\.md')
    return set(adr_pattern.findall(dt_content))


def update_step_state(step: dict, step_id: str, next_step: dict | None):
    """Обновить STEP_STATE.md.

    P2-refactor (v0.0.5 audit): функция разбита на 3 helper'а.
    Ранее CC=44 — теперь оркестратор с CC≈10.
    """
    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')

    # 1. Обновить CURRENT_STEP / LAST_COMPLETED_STEP / CURRENT_PHASE
    content = _update_current_step_fields(content, step_id, step, next_step)

    # 2. Добавить строку в COMPLETED_STEPS
    content = _add_completed_step_row(content, step_id, step, now)

    # 3. Перенос новых ADR в ADR_REGISTRY
    docs_target = step.get('docs_target', '')
    docs_target_path: Path | None = BASE / docs_target if docs_target else None
    adr_files_in_step = _extract_adr_files_from_step(docs_target_path) if docs_target_path is not None else set()
    if adr_files_in_step:
        content = _add_adr_registry_entries(content, step_id, adr_files_in_step)

    # 4. Записать обновлённый STEP_STATE.md
    atomic_write_text(step_state, content)
    log(f'STEP_STATE.md обновлён: step {step_id} завершён', step_id, 'action')


def _add_adr_registry_entries(
    content: str, step_id: str, adr_files: set
) -> str:
    """Добавить ADR-записи в ADR_REGISTRY секцию STEP_STATE.md.

    P2-refactor (v0.0.5 audit): вынесено из update_step_state.
    """
    adr_registry_match = re.search(
        r'(## ADR_REGISTRY\s*\n\| ADR \|.*?\|\n\|[-\s|]+\|\n)',
        content,
    )
    if not adr_registry_match:
        return content
    for adr_num, adr_slug in sorted(adr_files):
        adr_path = f'docs/для-разработчиков/adr/{adr_num}-{adr_slug}.md'
        adr_full = BASE / adr_path
        if not adr_full.exists():
            continue
        try:
            adr_content = adr_full.read_text(encoding='utf-8')
        except Exception:
            continue
        # Извлечь название ADR из первой строки заголовка
        title_match = re.search(r'^#\s+(.+)$', adr_content, re.MULTILINE)
        adr_title = title_match.group(1) if title_match else adr_slug
        new_row = f'| {adr_num} | {adr_title} | Accepted | [{adr_slug}]({adr_path}) |'
        if f'| {adr_num} |' not in content:
            content = re.sub(
                r'(## ADR_REGISTRY\s*\n\| ADR \|.*?\|\n\|[-\s|]+\|\n)',
                rf'\1{new_row}\n',
                content,
                count=1,
            )
    return content



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', '')

    # FIX-2.7: pre-check — если маркер уже существует, это аномалия
    allow_marker = META_DIR / '.allow-protected-edit'
    if allow_marker.exists():
        print(f'ERROR: обнаружен остаточный маркер {allow_marker}', file=sys.stderr)
        print('Это означает, что предыдущий запуск finalize_step.py был прерван.', file=sys.stderr)
        print(f'Удалите маркер вручную: rm {allow_marker}', file=sys.stderr)
        print(f'Или проверьте содержимое: cat {allow_marker}', file=sys.stderr)
        sys.exit(1)

    # FIX-2.11: flock для сериализации одновременных запусков finalize_step.
    # P11-A2-35 fix (фаза 11, Неделя 3, P1): добавлен try/except ImportError
    # вокруг `import fcntl` для Windows-совместимости. Ранее `import fcntl`
    # без guard приводил к ImportError с нечитаемым traceback на Windows до
    # того, как пользователь увидит осмысленное сообщение. Соответствует
    # паттерну в agent_log.py:147-160 и protect_files.py:122-133 (после C3 fix).
    # См. ADR-1 (Linux/macOS-only): на Windows блокировка отключается с warning.
    try:
        import fcntl
    except ImportError:
        print('WARN: fcntl недоступен (Windows?) — параллельный запуск '
              'finalize_step.py не сериализуется. См. ADR-1 (POSIX-only).',
              file=sys.stderr)
        fcntl = None  # type: ignore[assignment]
    lock_file = META_DIR / '.finalize.lock'
    lock_file.parent.mkdir(parents=True, exist_ok=True)
    lock_fd = open(lock_file, 'w')
    if fcntl is not None:
        try:
            fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            print('ERROR: другой процесс finalize_step.py уже запущен. Дождитесь завершения.', file=sys.stderr)
            sys.exit(1)
    # Держим lock_fd открытым до конца main()

    # FIX-2.12: обернуть основной код в try/except KeyboardInterrupt
    try:
        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')
        print('\n--- Проверка immutable-файлов (protect_files.py) ---')
        protect_script = _scripts_dir / 'protect_files.py'
        if protect_script.exists():
            # : использовать sys.executable для надёжного наследования venv
            # : в submodule-layout protect_files.py в .ump/scripts/ — вызываем через абсолютный путь
            # P0-7 fix: list-form без shell=True (shlex.quote не нужен — list-form сам экранирует).
            r = run_list([sys.executable, str(protect_script), '--check'], check=False, cwd=BASE)
            print(r.stdout)
            if r.returncode != 0:
                print('\n❌ Защищённые файлы были изменены без разрешения!')
                print('См. .agent/immutable.txt и 06-interactive-protocol.md §7 (PROTECT-ГЕЙТ).')
                log('Финализация FAIL: protect_files.py обнаружил несанкционированные изменения', args.step_id, 'error')
                sys.exit(1)
            log('protect_files.py PASSED — immutable-файлы не изменены', args.step_id, 'test')
        else:
            print('  (protect_files.py не найден — пропуск)')

        # 2. Многосторонняя проверка (если не --skip-verify)
        if not args.skip_verify:
            print('\n--- Многосторонняя проверка (verify_step_completion.py) ---')
            # : sys.executable
            # : в submodule-layout verify_step_completion.py в .ump/scripts/
            # P0-7 fix: list-form без shell=True.
            verify_script = str(_scripts_dir / 'verify_step_completion.py')
            r = run_list([sys.executable, verify_script, args.step_id], check=False, cwd=BASE)
            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-операции ВНАЧАЛЕ (: сначала git, потом STEP_STATE.md)
        # Если git_operations упадёт (merge conflict) — STEP_STATE.md останется
        # неповреждённым, шаг не будет отмечен как completed преждевременно.
        # Альтернатива из спецификации: STEP_STATE.md помечен assume-unchanged
        # и не входит в git-коммит — обновляем его ПОСЛЕ успешных git-операций.
        print('\n--- Git-операции ---')
        allow_marker.parent.mkdir(parents=True, exist_ok=True)
        # : маркер как YAML manifest с конкретным списком разрешённых файлов.
        # Раньше маркер был свободным текстом и protect_files.py мог интерпретировать
        # его как «allowed: ['*']». Теперь — ограничение только STEP_STATE.md + docs_target.
        _docs_target = step.get('docs_target', '')
        manifest_content = (
            '# Auto-created by finalize_step.py\n'
            'allowed:\n'
            '  - STEP_STATE.md\n'
            f'  - {_docs_target}\n'
            f'reason: finalize_step.py для шага {args.step_id}\n'
            f'created_at: {datetime.now().isoformat()}\n'
            'session: finalize_step.py\n'
            'expires: null\n'
        )
        allow_marker.write_text(manifest_content, encoding='utf-8')
        log('Создан маркер _meta/.allow-protected-edit (YAML manifest, )', args.step_id, 'action')
        try:
            ok, msg = git_operations(step, args.step_id, step_branch)
        finally:
            # Удалить маркер в любом случае (даже при ошибке)
            if allow_marker.exists():
                allow_marker.unlink()
                log('Удалён маркер _meta/.allow-protected-edit', args.step_id, 'action')
        print(f'  [{"✓" if ok else "✗"}] {msg}')
        if not ok:
            log(f'Финализация FAIL: git — {msg} (STEP_STATE.md НЕ обновлён — откат безопасен)', args.step_id, 'error')
            sys.exit(1)

        # 4. ОБНОВЛЕНИЕ STEP_STATE.md ПОСЛЕ успешных git-операций ()
        # STEP_STATE.md помечен assume-unchanged, не входит в git-коммит.
        # Обновляется только после успеха git — чтобы при rollback не было ложного completed.
        # /1.2/1.3/1.5 также выполняются внутри update_step_state():
        #   - перенос ADR в ADR_REGISTRY
        #   - обновление NEXT_ACTIONS
        #   - создание STAGE_XX_SUMMARY.md (если generates_stage_summary)
        #   - сбор TODO/FIXME в TECHNICAL_DEBT
        print('\n--- Обновление STEP_STATE.md (после git-операций) ---')
        update_step_state(step, args.step_id, next_step)
        print('  ✓ STEP_STATE.md обновлён')

        # 5. Обновить снимок файлов
        print('\n--- Обновление снимка файлов ---')
        # : sys.executable
        # : в submodule-layout detect_manual_changes.py в .ump/scripts/
        # P0-7 fix: list-form без shell=True. `2>/dev/null || true` не нужно —
        # capture_output уже подавляет stderr, check=False не падает на ненулевом exit.
        detect_script = str(_scripts_dir / 'detect_manual_changes.py')
        run_list([sys.executable, detect_script, '--update-snapshot', '--quiet'], check=False, cwd=BASE)
        print('  ✓ Снимок обновлён')

        # 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} ФИНАЛИЗИРОВАН (commit + push + PR + tag)')
        print(f'{"=" * 70}')
        print(f'\nGit: feature-ветка {step_branch} запушена, PR создан, тег step-{args.step_id} создан.')
        print('⚠ Merge в main НЕ выполнен — ожидание ручного merge через GitVerse web UI (политика v31).')
        print('   После merge скажите: «merge готов» — запустится пост-merge диагностика.')
        print('   Затем «продолжить» — для запуска следующего шага.')
        print(f'Документация: {step.get("docs_target", "")}')
        print(f'\nСледующий шаг: {next_step_id} — {next_step_title}')
        print()

        # Переименовать progress-файл (: per-step file)
        safe_step_id = args.step_id.replace('.', '-')
        progress_file = META_DIR / f'step-progress-{safe_step_id}.yaml'
        done_file = META_DIR / f'step-progress-{safe_step_id}.done.yaml'

        # P0-6 fix: обновить прогресс перед переименованием в .done.yaml
        # Раньше finished_at=None и current_phase=FINALIZATION оставались в .done файле,
        # что нарушало идемпотентность и запутывало повторный анализ.
        if progress_file.exists():
            try:
                import yaml as _yaml
                prog_data = _yaml.safe_load(progress_file.read_text(encoding='utf-8')) or {}
                prog_data['current_phase'] = 'FINALIZATION'
                prog_data['finished_at'] = datetime.now().strftime('%Y-%m-%d %H:%M')
                if 'phases' in prog_data and 'FINALIZATION' in prog_data['phases']:
                    prog_data['phases']['FINALIZATION']['status'] = 'completed'
                    prog_data['phases']['FINALIZATION']['finished_at'] = prog_data['finished_at']
                    prog_data['phases']['FINALIZATION']['last_error'] = None
                # P11-A2-36 fix (фаза 11, Неделя 3, P1): ранее progress_file.write_text
                # был неатомарным — crash mid-write мог повредить state file, после
                # чего orchestrate_step.check_state_sync сообщал "state machines
                # desynchronized" и шаг зависал. Теперь используется atomic_write_text
                # (определён в этом же модуле, строка 136), который пишет через
                # tempfile + os.replace — atomic on POSIX.
                atomic_write_text(
                    progress_file,
                    _yaml.safe_dump(prog_data, allow_unicode=True, sort_keys=False, default_flow_style=False),
                )
                log('Progress-файл обновлён: FINALIZATION=completed, finished_at set', args.step_id, 'action')
            except Exception as e:
                log(f'WARNING: не удалось обновить progress перед переименованием: {e}', args.step_id, 'warning')

        if progress_file.exists():
            progress_file.rename(done_file)
            print(f'Progress-файл: {done_file.name}')
        else:
            # Legacy fallback: если остался единый step-progress.yaml — переименовать его
            legacy_file = META_DIR / 'step-progress.yaml'
            if legacy_file.exists():
                legacy_file.rename(done_file)
                print(f'Progress-файл (legacy): {done_file.name}')

        #  [P0, NEW-P-003] — Запустить gen_progress.py для обновления docs/progress.md (§2.8)
        gen_progress_script = _scripts_dir / 'gen_progress.py'
        if gen_progress_script.exists():
            r = subprocess.run(
                # : sys.executable — гарантированно тот же интерпретатор
                [sys.executable, str(gen_progress_script)],
                capture_output=True, text=True, cwd=BASE, timeout=30
            )
            if r.returncode == 0:
                log('docs/progress.md обновлён через gen_progress.py', args.step_id, 'action')
                print('  ✓ docs/progress.md обновлён')
            else:
                log(f'gen_progress.py failed: {r.stderr[:200]}', args.step_id, 'warning')
                print('  ⚠ gen_progress.py failed (см. лог)')
        else:
            log('gen_progress.py не найден — пропуск обновления docs/progress.md',
                args.step_id, 'warning')

        sys.exit(0)
    except KeyboardInterrupt:
        print('\nПрервано пользователем (Ctrl+C)', file=sys.stderr)
        log('Прервано пользователем в фазе финализации', args.step_id, 'error')
        # Удалить маркер, если создан
        if allow_marker.exists():
            allow_marker.unlink()
        sys.exit(130)


if __name__ == '__main__':
    main()