Approval Engine

Callback-based approval flow for gating destructive or sensitive operations. Part of the unified CodeGraphHarness service harness.

Table of Contents

Overview

The approval engine decides whether operations (autofix application, pattern fixes, command execution) require explicit user confirmation before proceeding. It supports three policies:

Policy Behavior
auto Auto-approve whitelisted actions, prompt for others
ask_always Require approval for all actions
read_only Decline all write operations automatically

Architecture

flowchart LR
    subgraph Surfaces
        CLI[CLI]
        IDE[IDE]
        REST[REST API]
    end

    subgraph Harness["CodeGraphHarness (singleton)"]
        AE[ApprovalEngine]
        SM[HarnessSessionManager]
    end

    subgraph Policies
        P1["auto -- check whitelist, then prompt"]
        P2["ask_always -- always prompt"]
        P3["read_only -- always decline writes"]
    end

    subgraph Analysis
        AF[Autofix Engine]
        PF[Pattern Fixes]
    end

    CLI <--> AE
    IDE <--> AE
    REST <--> AE
    AE <--> SM
    AE --- Policies
    AF --> AE
    PF --> AE

Module Structure

src/harness/
├── __init__.py          # Public exports
├── approval.py          # ApprovalEngine
├── config.py            # HarnessConfig dataclass
├── core.py              # CodeGraphHarness singleton, get_harness()
├── models.py            # Item, Turn, Thread, ApprovalRequest, ApprovalDecision
└── session.py           # HarnessSessionManager

CodeGraphHarness

CodeGraphHarness is the central singleton that owns all services and exposes them through a uniform internal API. All surfaces (CLI, REST, MCP, ACP) become thin clients.

class CodeGraphHarness:
    def __init__(self, db_path: Optional[str] = None) -> None:

When db_path is None, the harness falls back to ProjectManager.get_active_db_path() for the CLI path.

Lazy-loaded properties:

Property Type Description
cpg CPGQueryService DuckDB CPG query client
gocpg GoCPGClient Go CPG generator client
copilot MultiScenarioCopilot LangGraph orchestrator
vector_store VectorStore ChromaDB vector store
sessions HarnessSessionManager Session manager
approval ApprovalEngine Approval engine

The reset() method clears all cached services (useful for project switches and testing).

get_harness()

def get_harness(db_path: Optional[str] = None) -> CodeGraphHarness
  • db_path=None – returns the global singleton (CLI path). Created on first call.
  • db_path="path" – returns a project-specific harness from an LRU cache (max 5 entries), creating a new one if needed.

reset_harness()

def reset_harness() -> None

Resets the global singleton and all cached instances. Intended for testing.

API Reference

ApprovalRequest

A pending approval request surfaced to the client.

@dataclass
class ApprovalRequest:
    id: str = field(default_factory=_new_id)       # Auto-generated UUID
    thread_id: str = ""                             # Session thread ID
    action_type: str = ""                           # e.g., 'apply_autofix', 'apply_pattern_fix'
    details: Dict[str, Any] = field(default_factory=dict)  # Arbitrary action details
    status: str = "pending"                         # 'pending' | 'accepted' | 'declined' | 'cancelled'

All fields have defaults, so ApprovalRequest() creates a valid instance with an auto-generated id and status="pending".

ApprovalDecision

Client’s response to an approval request.

@dataclass
class ApprovalDecision:
    decision: str = "decline"       # 'accept' | 'accept_for_session' | 'decline' | 'cancel'
    reason: Optional[str] = None    # Optional reason text
    scope: Optional[str] = None     # Scope for 'accept_for_session'

Security default: the decision field defaults to "decline". When an ApprovalDecision is created without explicitly setting decision, the operation is declined. This is a deliberate safety measure to prevent unintended approval of destructive operations.

ApprovalEngine

class ApprovalEngine:
    def __init__(self, config: HarnessConfig):
        """Initialize with harness configuration."""

    async def request_approval(
        self,
        action_type: str,
        details: Dict,
        thread_id: str,
    ) -> ApprovalDecision:
        """
        Request approval for an action. Blocks until resolved.

        Auto-approves if action_type is in the whitelist or was
        previously approved for this session.

        Args:
            action_type: Type of action (e.g., 'apply_autofix')
            details: Action context (file, line, description)
            thread_id: Current session thread

        Returns:
            ApprovalDecision with the user's decision
        """

    def resolve_approval(self, request_id: str, decision: ApprovalDecision) -> None:
        """
        Resolve a pending approval request (called by UI/CLI).

        If decision is 'accept_for_session', all future requests of the
        same action_type in this thread are auto-approved.
        """

    def get_pending_request_ids(self) -> list:
        """Return IDs of all pending approval requests."""

    def clear_session_approvals(self, thread_id: str) -> None:
        """Clear session-scoped auto-approvals on session end."""

Session Models

The session hierarchy is: Thread contains Turns, each Turn contains Items.

Item

Atomic unit of work within a Turn.

@dataclass
class Item:
    id: str = field(default_factory=_new_id)
    type: str = "user_message"          # user_message | analysis_result | approval_request | error
    status: str = "started"             # started | in_progress | completed
    content: Any = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    created_at: float = field(default_factory=time.time)

Methods: to_dict() -> dict, from_dict(d: dict) -> Item (classmethod).

Turn

A request-response cycle within a Thread.

@dataclass
class Turn:
    id: str = field(default_factory=_new_id)
    thread_id: str = ""
    items: List[Item] = field(default_factory=list)
    status: str = "in_progress"         # in_progress | completed | interrupted | failed
    started_at: float = field(default_factory=time.time)
    completed_at: Optional[float] = None

Methods: to_dict() -> dict, from_dict(d: dict) -> Turn (classmethod).

Thread

Top-level session container, persisted as one JSON file.

@dataclass
class Thread:
    id: str = field(default_factory=_new_id)
    name: Optional[str] = None
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)
    turns: List[Turn] = field(default_factory=list)
    status: str = "idle"                # idle | active | archived
    config: Dict[str, Any] = field(default_factory=dict)

Methods: to_dict() -> dict, from_dict(d: dict) -> Thread (classmethod).

HarnessSessionManager

Manages harness threads with an in-memory cache and JSON file persistence. Each thread is stored as data/sessions/{thread_id}.json.

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

Accessed via harness.sessions.

Public async methods:

Method Description
create_thread(name=None, config=None, thread_id=None) -> Thread Create a new thread and persist it
get_thread(thread_id) -> Optional[Thread] Get thread by ID (memory first, then disk)
list_threads(status=None) -> List[Thread] List threads, optionally filtered by status
archive_thread(thread_id) -> bool Mark thread as archived, persist, evict from memory
start_turn(thread_id) -> Turn Start a new turn in the given thread
complete_turn(thread_id, turn_id, status="completed") -> Turn Complete a turn and persist
add_item(thread_id, turn_id, item_type, content, metadata=None) -> Item Add an item to a turn
update_item_status(thread_id, turn_id, item_id, status) -> Item Update the status of an existing item

Expired threads (past thread_ttl_hours) are cleaned up automatically when the thread limit is reached.

Configuration

HarnessConfig

@dataclass
class HarnessConfig:
    session_dir: str = "./data/sessions"
    max_threads: int = 100
    thread_ttl_hours: int = 168                       # 7 days

    default_approval_policy: str = "auto"             # auto | ask_always | read_only
    auto_approve_actions: List[str] = field(...)      # see YAML below
    require_approval_actions: List[str] = field(...)  # see YAML below

Note: the dataclass field is named default_approval_policy. In YAML configuration the key is default_policy, which is mapped to this field during config loading.

YAML Configuration

# config.yaml
harness:
  session_dir: ./data/sessions
  max_threads: 100
  thread_ttl_hours: 168            # 7 days
  approval:
    default_policy: auto             # auto | ask_always | read_only
    auto_approve:                    # Actions auto-approved (no prompt)
      - codegraph_query
      - codegraph_find_callers
      - codegraph_find_callees
      - codegraph_explain
    require_approval:                # Actions that always require approval
      - apply_autofix
      - apply_pattern_fix
      - command_execution

Usage

Request Approval

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": "Replace strcpy with strncpy",
        "cwe": "CWE-120"
    },
    thread_id="session_abc"
)

if decision.decision == "accept":
    apply_fix(...)
elif decision.decision == "decline":
    log(f"Fix rejected: {decision.reason}")

Resolve from UI

# Called by CLI/IDE when user makes a decision
harness.approval.resolve_approval(
    request_id="req_abc123",
    decision=ApprovalDecision(decision="accept")
)

Session-Scoped Approval

When a user selects accept_for_session, all subsequent requests of the same action_type within the same thread are auto-approved:

# User approves once for the whole session
decision = ApprovalDecision(decision="accept_for_session")
harness.approval.resolve_approval(request_id, decision)

# Subsequent autofix requests auto-approved (no prompt)
decision = await harness.approval.request_approval(
    action_type="apply_autofix",  # Same type -> auto-approved
    details={...},
    thread_id="session_abc"       # Same thread
)
# decision.decision == "accept" (automatic)

Decision Flow

flowchart TD
    A["request_approval(action_type, details, thread_id)"] --> B{action_type in<br/>auto_approve list?}
    B -- YES --> C["return accept immediately"]
    B -- NO --> D{action_type in<br/>session_approvals for thread?}
    D -- YES --> E["return accept (session-scoped)"]
    D -- NO --> F{default_policy<br/>== read_only?}
    F -- YES --> G["return decline('read_only policy')"]
    F -- NO --> H["Create pending request,<br/>await user decision"]
    H --> I["accept -- apply action"]
    H --> J["accept_for_session -- apply + remember"]
    H --> K["decline -- skip action"]
    H --> L["cancel -- abort operation"]

Default Action Classification

Auto-Approved (read-only) Requires Approval (write)
codegraph_query apply_autofix
codegraph_find_callers apply_pattern_fix
codegraph_find_callees command_execution
codegraph_explain

Integration Points

The approval engine is used by three subsystems. ACP has its own separate approval bridge.

Autofix Engine

src/analysis/autofix/engine.py – AutofixEngine.apply_with_approval() requests approval before applying security fixes:

decision = await approval_engine.request_approval(
    action_type="apply_autofix",
    details={
        "file": result.fix.file_path,
        "description": result.fix.description,
        "diff_preview": result.fix.diff_patch[:500],
    },
    thread_id=thread_id,
)

Pattern Fixes CLI

src/cli/patterns_command.py – _run_fix() requests approval before applying pattern-based code rewrites:

harness = get_harness()
decision = await harness.approval.request_approval(
    action_type="apply_pattern_fix",
    details={
        "total_matches": preview.total_matches,
    },
    thread_id=thread_id,
)

Pattern Fixes REST API

src/api/routers/patterns.py – the fix endpoint requests approval when dry_run=False:

harness = get_harness()
decision = await harness.approval.request_approval(
    action_type="apply_pattern_fix",
    details={"rule_id": request.rule_id, "dry_run": False},
    thread_id="api",
)
if decision.decision not in ("accept", "accept_for_session"):
    raise HTTPException(status_code=403, detail="Fix rejected by approval engine")

ACP Approval Bridge (separate system)

ACP uses ACPApprovalBridge (src/acp/server/approval_bridge.py), which is a separate implementation from ApprovalEngine. It provides a different API tailored for IDE integration:

  • Method: request_file_change_approval()
  • Timeout: 300 seconds
  • Notifications via item/fileChange/requestApproval

ACPApprovalBridge does not use ApprovalEngine internally. See the ACP documentation for details.

Module: src/harness/ Last updated: March 2026