Механизм подтверждения операций

Этот модуль управляет подтверждением чувствительных и потенциально опасных действий. Он входит в унифицированный сервисный слой CodeGraphHarness и решает, когда операцию можно выполнить сразу, а когда нужно явное разрешение пользователя.

Содержание

Обзор

Механизм подтверждения определяет, нужно ли для операции явное разрешение пользователя. Это относится к автоисправлениям, шаблонным правкам и выполнению команд. Поддерживаются три базовые политики:

Политика Поведение
auto Автоматическое одобрение операций из белого списка, запрос для остальных
ask_always Запрос подтверждения для всех операций
read_only Автоматический отказ для всех операций записи

Архитектура

flowchart TB
    User["Пользователь / CLI / IDE"]
    AE["ApprovalEngine"]
    Harness["CodeGraphHarness"]
    Modules["Модули анализа"]
    Auto["auto — проверка белого списка, затем запрос"]
    Ask["ask_always — всегда запрос"]
    RO["read_only — всегда отклонять запись"]

    User <--> AE
    AE <--> Harness
    Harness <--> Modules
    AE --> Policies
    subgraph Policies["Политики"]
        Auto
        Ask
        RO
    end

Структура модуля

src/harness/
├── __init__.py          # Публичные экспорты
├── approval.py          # ApprovalEngine
├── config.py            # HarnessConfig (dataclass)
├── core.py              # CodeGraphHarness (синглтон), get_harness()
├── models.py            # Item, Turn, Thread, ApprovalRequest, ApprovalDecision
└── session.py           # HarnessSessionManager

CodeGraphHarness

Центральный синглтон, предоставляющий доступ ко всем подсистемам harness. Создаётся через фабричную функцию get_harness().

from src.harness.core import CodeGraphHarness

class CodeGraphHarness:
    def __init__(self, config: HarnessConfig, db_path: Optional[str] = None):
        ...

Ленивые свойства (инициализируются при первом обращении):

Свойство Тип Описание
cpg CPGQueryService Клиент DuckDB для запросов к CPG
gocpg GoCPGClient Клиент Go CPG-генератора
copilot MultiScenarioCopilot LangGraph-оркестратор
vector_store VectorStore Хранилище векторных представлений
sessions HarnessSessionManager Менеджер сессий
approval ApprovalEngine Механизм подтверждения операций

get_harness()

def get_harness(db_path: Optional[str] = None) -> CodeGraphHarness

Фабричная функция для получения экземпляра CodeGraphHarness:

  • db_path=None — возвращает глобальный синглтон
  • db_path="path" — возвращает LRU-кешированный экземпляр (максимум 5 одновременных)

Пример:

from src.harness import get_harness

# Глобальный синглтон
harness = get_harness()

# Экземпляр для конкретного проекта (multi-tenant)
harness = get_harness(db_path="/data/projects/myproject.duckdb")

reset_harness()

def reset_harness() -> None

Сбрасывает глобальный синглтон. Предназначена для использования в тестах.

Справочник API

ApprovalRequest

@dataclass
class ApprovalRequest:
    id: str = field(default_factory=_new_id)   # Автогенерация UUID
    thread_id: str = ""                         # ID потока сессии
    action_type: str = ""                       # Например, 'apply_autofix', 'command_execution'
    details: Dict[str, Any] = field(default_factory=dict)  # Произвольные детали операции
    status: str = "pending"                     # 'pending' | 'accepted' | 'declined' | 'cancelled'

Все поля имеют значения по умолчанию. Поле id автоматически генерируется как UUID при создании экземпляра.

ApprovalDecision

@dataclass
class ApprovalDecision:
    decision: str = "decline"           # 'accept' | 'accept_for_session' | 'decline' | 'cancel'
    reason: Optional[str] = None        # Необязательная причина
    scope: Optional[str] = None         # Область для 'accept_for_session'

Безопасное значение по умолчанию: при отсутствии явной инициализации поле decision равно "decline", что гарантирует отклонение операции.

ApprovalEngine

class ApprovalEngine:
    def __init__(self, config: HarnessConfig):
        """Инициализация с конфигурацией harness."""

    async def request_approval(
        self,
        action_type: str,
        details: Dict,
        thread_id: str,
    ) -> ApprovalDecision:
        """
        Запросить подтверждение операции. Блокирует до получения решения.

        Автоматически одобряет, если action_type в белом списке или
        ранее одобрен для данной сессии.

        Аргументы:
            action_type: тип операции (например, 'apply_autofix')
            details: контекст операции (файл, строка, описание)
            thread_id: текущий поток сессии

        Возвращает:
            ApprovalDecision с решением пользователя
        """

    def resolve_approval(self, request_id: str, decision: ApprovalDecision) -> None:
        """
        Разрешить ожидающий запрос (вызывается из UI/CLI).

        При решении 'accept_for_session' все последующие запросы того же
        типа в данном потоке одобряются автоматически.
        """

    def get_pending_request_ids(self) -> list:
        """Получить ID всех ожидающих запросов."""

    def clear_session_approvals(self, thread_id: str) -> None:
        """Очистить сессионные автоодобрения при завершении сессии."""

Модели сессий

Иерархия сессий: Thread содержит Turn, каждый Turn содержит Item.

Item

@dataclass
class Item:
    id: str                             # Уникальный идентификатор элемента
    type: str                           # Тип элемента
    content: str                        # Содержимое
    metadata: Dict[str, Any]            # Метаданные
    status: str                         # Статус элемента
    created_at: str                     # Временная метка создания

Методы: to_dict(), from_dict().

Turn

@dataclass
class Turn:
    id: str                             # Уникальный идентификатор хода
    thread_id: str                      # ID родительского потока
    items: List[Item]                   # Элементы в ходе
    status: str                         # Статус хода
    started_at: str                     # Время начала
    completed_at: Optional[str]         # Время завершения

Методы: to_dict(), from_dict().

Thread

@dataclass
class Thread:
    id: str                             # Уникальный идентификатор потока
    name: str                           # Имя потока
    turns: List[Turn]                   # Ходы в потоке
    status: str                         # Статус потока
    config: Dict[str, Any]              # Конфигурация потока
    created_at: str                     # Время создания
    updated_at: str                     # Время последнего обновления

Методы: to_dict(), from_dict().

HarnessSessionManager

Менеджер сессий, доступный через свойство harness.sessions. Обеспечивает персистенцию потоков в JSON-файлах.

class HarnessSessionManager:
    def __init__(self, config: HarnessConfig):
        ...

Публичные асинхронные методы:

Метод Описание
create_thread() Создать новый поток
get_thread() Получить поток по ID
list_threads() Получить список всех потоков
archive_thread() Архивировать поток
start_turn() Начать новый ход в потоке
complete_turn() Завершить текущий ход
add_item() Добавить элемент к текущему ходу
update_item_status() Обновить статус элемента

Хранение: JSON-файлы в data/sessions/{thread_id}.json. Используется кеш в памяти с синхронизацией на диск. Устаревшие потоки очищаются по TTL (настраивается через thread_ttl_hours).

Конфигурация

HarnessConfig

@dataclass
class HarnessConfig:
    session_dir: str = "./data/sessions"
    max_threads: int = 100
    thread_ttl_hours: int = 168                     # 7 дней
    default_approval_policy: str = "auto"           # auto | ask_always | read_only
    auto_approve: List[str] = field(default_factory=list)
    require_approval: List[str] = field(default_factory=list)

Имя поля в dataclass: default_approval_policy. В YAML-конфигурации используется ключ default_policy, который маппится на это поле при загрузке.

YAML-конфигурация

# config.yaml
harness:
  session_dir: ./data/sessions
  max_threads: 100
  thread_ttl_hours: 168                  # 7 дней
  approval:
    default_policy: auto                 # Маппится на поле default_approval_policy
    auto_approve:                        # Автоматически одобряемые операции
      - codegraph_query
      - codegraph_find_callers
      - codegraph_find_callees
      - codegraph_explain
    require_approval:                    # Операции, требующие подтверждения
      - apply_autofix
      - apply_pattern_fix
      - command_execution

Использование

Запрос подтверждения

from src.harness import get_harness

harness = get_harness()

decision = await harness.approval.request_approval(
    action_type="apply_autofix",
    details={
        "file": "main.c",
        "line": 42,
        "fix": "Заменить strcpy на strncpy",
        "cwe": "CWE-120"
    },
    thread_id="session_abc"
)

if decision.decision == "accept":
    apply_fix(...)
elif decision.decision == "decline":
    log(f"Исправление отклонено: {decision.reason}")

Разрешение из UI

# Вызывается из CLI/IDE при решении пользователя
from src.harness.models import ApprovalDecision

harness.approval.resolve_approval(
    request_id="req_abc123",
    decision=ApprovalDecision(decision="accept")
)

Сессионное подтверждение

При выборе accept_for_session все последующие запросы того же action_type в текущем потоке одобряются автоматически:

# Пользователь одобряет однократно на всю сессию
decision = ApprovalDecision(decision="accept_for_session")
harness.approval.resolve_approval(request_id, decision)

# Последующие запросы autofix одобряются автоматически
decision = await harness.approval.request_approval(
    action_type="apply_autofix",  # Тот же тип — автоодобрение
    details={...},
    thread_id="session_abc"       # Тот же поток
)
# decision.decision == "accept" (автоматически)

Логика принятия решений

flowchart TD
    Start["request_approval(action_type, details, thread_id)"]
    CheckAuto{"action_type в списке auto_approve?"}
    CheckSession{"action_type в session_approvals для thread_id?"}
    CheckReadOnly{"default_policy == read_only?"}
    Accept1["Вернуть accept"]
    Accept2["Вернуть accept (сессионное)"]
    Decline["Вернуть decline (политика read_only)"]
    Wait["Создать ожидающий запрос, ждать решения"]
    ResAccept["accept — выполнить операцию"]
    ResSession["accept_for_session — выполнить + запомнить"]
    ResDecline["decline — пропустить операцию"]
    ResCancel["cancel — отменить операцию"]

    Start --> CheckAuto
    CheckAuto -- Да --> Accept1
    CheckAuto -- Нет --> CheckSession
    CheckSession -- Да --> Accept2
    CheckSession -- Нет --> CheckReadOnly
    CheckReadOnly -- Да --> Decline
    CheckReadOnly -- Нет --> Wait
    Wait --> ResAccept
    Wait --> ResSession
    Wait --> ResDecline
    Wait --> ResCancel

Классификация операций по умолчанию

Автоодобряемые (только чтение) Требуют подтверждения (запись)
codegraph_query apply_autofix
codegraph_find_callers apply_pattern_fix
codegraph_find_callees command_execution
codegraph_explain

Точки интеграции

Механизм подтверждения ApprovalEngine используется в следующих компонентах:

Автоисправления

Файл: src/analysis/autofix/engine.py

Метод AutofixEngine.apply_with_approval() вызывает harness.approval.request_approval() с типом операции apply_autofix перед применением исправлений безопасности.

decision = await harness.approval.request_approval(
    action_type="apply_autofix",
    details={"file": path, "fix": description, "cwe": cwe_id},
    thread_id=thread_id
)

Шаблоны CLI

Файл: src/cli/patterns_command.py

Функция _run_fix() вызывает harness.approval.request_approval() с типом операции apply_pattern_fix при исправлении структурных нарушений через командную строку.

Шаблоны REST

Файл: src/api/routers/patterns.py

Маршрут применения исправлений по шаблонам использует механизм подтверждения в контексте REST API.

ACP (отдельная система)

Протокол ACP использует отдельную реализацию подтверждения — ACPApprovalBridge (src/acp/server/approval_bridge.py), а не ApprovalEngine из harness. Этот компонент имеет собственный API:

  • Метод request_file_change_approval() для запроса подтверждения изменения файлов
  • Таймаут 300 секунд
  • Уведомления через item/fileChange/requestApproval

Подробности в Интеграция ACP.

Связанная документация

Модуль: src/harness/ Последнее обновление: март 2026