> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cognisafe.uk/llms.txt
> Use this file to discover all available pages before exploring further.

# Human-in-the-Loop Approval Workflows

> Require human approval for high-risk agent actions before they execute

Autonomous agents can take irreversible real-world actions: send emails, delete database records, make payments, execute code, post to social media, grant permissions. Unlike a model that produces text, an agent that calls a tool has side effects in the world. A poorly-aligned response can be corrected in the next turn. A sent wire transfer cannot.

A human-in-the-loop (HITL) gate is the last line of defence between an AI agent and an irreversible action. This reference architecture describes how to build one using Cognisafe's webhook system and your existing approval tooling.

***

## Two variants

This pattern has two variants with different trade-offs:

|                           | Variant 1: Pre-execution gate                    | Variant 2: Post-detection review              |
| ------------------------- | ------------------------------------------------ | --------------------------------------------- |
| **When**                  | Before the dangerous action executes             | After the dangerous action completes          |
| **Effect**                | Prevents the action if rejected                  | Creates an audit trail; informs future policy |
| **Latency**               | Adds human response time (minutes to hours)      | Zero added latency                            |
| **Framework requirement** | Agent must support async tool execution          | None — works with any agent                   |
| **Use for**               | Irreversible actions (payments, deletes, emails) | Monitoring and policy calibration             |

For truly irreversible actions, use Variant 1. For actions that can be unwound or that are lower stakes, Variant 2 is sufficient and imposes no user-visible latency.

***

## Variant 1: Pre-execution gate

### How it works

```text theme={null}
Agent decides to call a dangerous tool
          │
          ▼
Approval-gating wrapper intercepts the tool call
          │
          ▼
Submit approval request to approval service
(ServiceNow workflow / Jira approval / Slack bot / custom FastAPI)
          │
          ▼
Approval service notifies human approver
(email, Slack, PagerDuty, mobile app)
          │
          ├── Approved ──▶  Execute the real tool · return result to agent
          │
          └── Rejected ──▶  Raise PermissionError · agent handles gracefully
                            (or: timeout ──▶ fail-safe reject)

Cognisafe logs the tool call in both paths:
  - Audit trail of what was requested, who approved/rejected, when
  - Fires threat_detected webhook if the tool name matches high-risk classifier
```

### Framework requirements

Variant 1 requires the agent framework to support intercepting tool calls before execution. This works with:

* **LangGraph** — add a conditional edge before the tool node that routes to an approval subgraph
* **Semantic Kernel** — implement a `FunctionInvocationFilter` middleware
* **Custom MCP server** — wrap dangerous tool handlers in the approval gate
* **LangChain custom tools** — override `_run` / `_arun` with the gate wrapper

It does **not** work with frameworks where tool execution is entirely managed by the LLM provider (e.g., OpenAI Assistants with hosted tools). In those cases, use Variant 2 for monitoring and redesign the agent to use a self-hosted tool server.

### Approval-gating wrapper

```python theme={null}
import hashlib, hmac, time
import httpx

# Configuration
APPROVAL_SERVICE_URL = "https://approvals.yourco.com"
APPROVAL_SERVICE_TOKEN = "your-approval-service-token"
COGNISAFE_API_URL = "https://api.cognisafe.io"
COGNISAFE_API_KEY = "your-cognisafe-api-key"

# High-risk tools that always require approval
HIGH_RISK_TOOLS = {
    # Financial
    "transfer_funds", "create_payment", "refund_order", "update_billing",
    # Destructive
    "delete_record", "drop_table", "terminate_instance", "purge_queue",
    # External communication
    "send_email", "post_to_social", "create_pr", "merge_pr", "send_sms",
    # Privilege
    "add_user", "grant_role", "update_permissions", "delete_user",
}

def gated_tool(tool_name: str, args: dict, agent_name: str = "unknown") -> dict:
    """
    Wrap any dangerous tool call in a HITL approval gate.
    Blocks until approved, rejected, or timed out.
    """
    if tool_name not in HIGH_RISK_TOOLS:
        # Not a high-risk tool — execute directly
        return execute_tool(tool_name, args)

    # Get the current Cognisafe request ID for audit trail linkage
    cognisafe_request_id = get_current_cognisafe_request_id()

    # 1. Submit approval request
    resp = httpx.post(
        f"{APPROVAL_SERVICE_URL}/requests",
        json={
            "tool": tool_name,
            "args": args,
            "agent": agent_name,
            "cognisafe_request_id": cognisafe_request_id,
            "requested_at": time.time(),
            "context": {
                "why": "Agent requested this action during task execution",
                "risk": describe_risk(tool_name),
                "owasp": "LLM06 - Excessive Agency",
                "review_url": f"{COGNISAFE_API_URL}/requests/{cognisafe_request_id}",
            }
        },
        headers={"Authorization": f"Bearer {APPROVAL_SERVICE_TOKEN}"},
        timeout=10,
    )
    resp.raise_for_status()
    ticket_id = resp.json()["ticket_id"]

    # 2. Poll for decision (max 5 minutes, poll every 10 seconds)
    max_polls = 30
    for attempt in range(max_polls):
        time.sleep(10)
        status_resp = httpx.get(
            f"{APPROVAL_SERVICE_URL}/requests/{ticket_id}",
            headers={"Authorization": f"Bearer {APPROVAL_SERVICE_TOKEN}"},
            timeout=5,
        )
        status = status_resp.json()

        if status["decision"] == "approved":
            # Log approval in Cognisafe audit trail
            log_approval_decision(cognisafe_request_id, ticket_id, "approved", status)
            return execute_tool(tool_name, args)

        elif status["decision"] == "rejected":
            log_approval_decision(cognisafe_request_id, ticket_id, "rejected", status)
            raise PermissionError(
                f"Action '{tool_name}' was rejected by {status.get('approver', 'an approver')}. "
                f"Reason: {status.get('reason', 'No reason provided.')} "
                f"Ticket: {ticket_id}"
            )

    # Timeout: fail-safe reject (never auto-approve on timeout)
    log_approval_decision(cognisafe_request_id, ticket_id, "timeout", {})
    raise TimeoutError(
        f"Approval for '{tool_name}' timed out after {max_polls * 10} seconds. "
        f"Action has been blocked. Ticket: {ticket_id}"
    )


def describe_risk(tool_name: str) -> str:
    risk_descriptions = {
        "transfer_funds":      "Irreversible financial transfer. Cannot be unwound without manual intervention.",
        "delete_record":       "Permanent data deletion. Recoverable only from backup.",
        "send_email":          "External communication sent to real recipients.",
        "grant_role":          "Privilege escalation. Could give a user unauthorised access.",
        "terminate_instance":  "Infrastructure destruction. May cause service outage.",
        "merge_pr":            "Code change merged to production branch.",
    }
    return risk_descriptions.get(tool_name, "High-risk action — review before approving.")


def log_approval_decision(request_id: str, ticket_id: str, decision: str, status: dict) -> None:
    """Post the approval decision back to Cognisafe for the audit trail."""
    try:
        httpx.post(
            f"{COGNISAFE_API_URL}/api/requests/{request_id}/annotations",
            json={
                "type": "hitl_decision",
                "ticket_id": ticket_id,
                "decision": decision,
                "approver": status.get("approver"),
                "reason": status.get("reason"),
            },
            headers={"Authorization": f"Bearer {COGNISAFE_API_KEY}"},
            timeout=5,
        )
    except Exception:
        pass  # Non-blocking — audit log failure must not block the tool gate
```

<Warning>
  Never auto-approve on timeout. The fail-safe must be rejection. An agent that can time out into execution has no gate — it just has a delay. If the approval SLA cannot be met, consider whether the action is suitable for automation at all.
</Warning>

### LangGraph integration

```python theme={null}
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    pending_tool_call: dict | None
    approval_status: str | None

def should_gate(state: AgentState) -> str:
    """Route to approval gate if the pending tool call is high-risk."""
    pending = state.get("pending_tool_call")
    if pending and pending["name"] in HIGH_RISK_TOOLS:
        return "approval_gate"
    return "execute_tool"

def approval_gate_node(state: AgentState) -> AgentState:
    """Block and wait for human approval."""
    pending = state["pending_tool_call"]
    try:
        result = gated_tool(
            tool_name=pending["name"],
            args=pending["args"],
            agent_name="my-agent",
        )
        return {**state, "approval_status": "approved", "pending_tool_call": None}
    except PermissionError as e:
        return {**state, "approval_status": f"rejected: {e}", "pending_tool_call": None}
    except TimeoutError as e:
        return {**state, "approval_status": f"timeout: {e}", "pending_tool_call": None}

# Build the graph
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("approval_gate", approval_gate_node)
graph.add_node("tool_executor", ToolNode(safe_tools))

graph.add_conditional_edges("agent", should_gate, {
    "approval_gate": "approval_gate",
    "execute_tool": "tool_executor",
})
graph.add_edge("approval_gate", "agent")  # Return to agent after decision
graph.add_edge("tool_executor", "agent")
```

***

### Slack approval bot

The simplest approval service for many teams is a Slack bot with Approve/Reject buttons. This avoids deploying a separate service and integrates with where engineers already work.

````python theme={null}
# approval_bot.py — FastAPI-based Slack approval bot
from fastapi import FastAPI, Request
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
import json, uuid

app = FastAPI()
slack = WebClient(token="xoxb-your-slack-bot-token")

# In-memory store (use Redis in production for multi-instance deployments)
pending_approvals: dict[str, dict] = {}

APPROVAL_CHANNEL = "#ai-hitl-approvals"

@app.post("/requests")
async def create_approval_request(request: Request) -> dict:
    body = await request.json()
    ticket_id = str(uuid.uuid4())

    pending_approvals[ticket_id] = {
        "ticket_id": ticket_id,
        "decision": "pending",
        "approver": None,
        "reason": None,
        **body,
    }

    # Build Slack Block Kit message
    blocks = [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": ":warning: Agent Action Requires Approval"}
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*Tool:* `{body['tool']}`"},
                {"type": "mrkdwn", "text": f"*Agent:* `{body['agent']}`"},
                {"type": "mrkdwn", "text": f"*Risk:* {body.get('context', {}).get('risk', 'Unknown')}"},
                {"type": "mrkdwn", "text": f"*OWASP:* {body.get('context', {}).get('owasp', 'LLM06')}"},
            ]
        },
        {
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"*Arguments:*\n```{json.dumps(body['args'], indent=2)[:500]}```"}
        },
        {
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"<{body.get('context', {}).get('review_url', '')}|View full request in Cognisafe>"}
        },
        {
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Approve"},
                    "style": "primary",
                    "value": ticket_id,
                    "action_id": "approve_action",
                },
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Reject"},
                    "style": "danger",
                    "value": ticket_id,
                    "action_id": "reject_action",
                }
            ]
        }
    ]

    slack.chat_postMessage(channel=APPROVAL_CHANNEL, blocks=blocks)
    return {"ticket_id": ticket_id, "status": "pending"}


@app.get("/requests/{ticket_id}")
async def get_approval_status(ticket_id: str) -> dict:
    approval = pending_approvals.get(ticket_id)
    if not approval:
        return {"ticket_id": ticket_id, "decision": "not_found"}
    return approval


@app.post("/slack/interactions")
async def handle_slack_interaction(request: Request) -> dict:
    form = await request.form()
    payload = json.loads(form["payload"])

    action = payload["actions"][0]
    ticket_id = action["value"]
    action_id = action["action_id"]
    approver = payload["user"]["name"]

    if ticket_id not in pending_approvals:
        return {"text": "Approval request not found or already resolved."}

    if action_id == "approve_action":
        pending_approvals[ticket_id]["decision"] = "approved"
        pending_approvals[ticket_id]["approver"] = approver
        message = f":white_check_mark: Approved by {approver}"
    else:
        pending_approvals[ticket_id]["decision"] = "rejected"
        pending_approvals[ticket_id]["approver"] = approver
        message = f":x: Rejected by {approver}"

    # Update the Slack message to show the decision
    slack.chat_update(
        channel=payload["channel"]["id"],
        ts=payload["message"]["ts"],
        text=message,
        blocks=[{
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"{message}\nTool: `{pending_approvals[ticket_id]['tool']}`"}
        }]
    )

    return {"status": "ok"}
````

***

## Variant 2: Post-detection review

When pre-execution gating is not feasible, configure Cognisafe to detect and alert on high-risk tool calls after they complete. This creates an audit trail and informs future policy even if it does not block the action.

### Configure the custom scorer

```yaml theme={null}
# evals/scorers.yaml
scorers:
  - name: dangerous_tool_detector
    type: keyword_list
    severity: critical
    description: "Detects calls to high-risk tools in agent output"
    keywords:
      # Financial
      - "transfer_funds"
      - "create_payment"
      - "refund_order"
      # Destructive
      - "delete_record"
      - "drop_table"
      - "terminate_instance"
      # External communication
      - "send_email"
      - "post_to_social"
      - "create_pr"
      # Privilege
      - "add_user"
      - "grant_role"
      - "update_permissions"
    # Fire threat_detected webhook on any match
    webhook_on_match: true
```

### Webhook receiver: post-detection review ticket

```python theme={null}
async def handle_post_detection(event: dict) -> None:
    """
    Create a review ticket for a completed high-risk action.
    This is audit trail + policy feedback, not a gate.
    """
    data = event["data"]

    # Create a Jira review ticket
    await create_jira_ticket(event, {
        "summary": f"[HITL Review] {data['agent_name']} executed high-risk tool",
        "description": (
            f"A high-risk tool call was detected in a completed agent interaction.\n\n"
            f"This ticket is for review and policy purposes — the action has already executed.\n\n"
            f"Tool pattern detected: {data.get('prompt_snippet', 'see report')}\n"
            f"Agent: {data['agent_name']} ({data.get('agent_tag', 'untagged')})\n"
            f"Full interaction: {data.get('report_url', 'N/A')}\n\n"
            f"Action: Review the interaction. If the action was inappropriate, "
            f"add a pre-execution gate for this tool using Variant 1 of the HITL pattern.\n"
            f"Docs: https://docs.cognisafe.io/reference-architectures/hitl-approval"
        ),
        "labels": ["hitl-review", "ai-security", "owasp-llm06"],
    })
```

***

## High-risk tool classification

Not all tools need HITL. Apply the gate selectively to preserve agent velocity for low-risk operations. Classify tools by the reversibility of their side effects and the blast radius of a mistake.

| Category               | Tools                                                                | HITL required | Rationale                                                     |
| ---------------------- | -------------------------------------------------------------------- | ------------- | ------------------------------------------------------------- |
| Financial              | `transfer_funds`, `create_payment`, `refund_order`, `update_billing` | Always        | Irreversible without manual intervention; regulatory exposure |
| Destructive            | `delete_record`, `drop_table`, `terminate_instance`, `purge_queue`   | Always        | Data or infrastructure loss; recovery from backup only        |
| External communication | `send_email`, `post_to_social`, `create_pr`, `merge_pr`, `send_sms`  | Always        | Reputational risk; cannot un-send                             |
| Privilege escalation   | `add_user`, `grant_role`, `update_permissions`, `delete_user`        | Always        | Security boundary violation; enables lateral movement         |
| External API calls     | `call_external_api`, `submit_form`, `place_order`                    | Depends       | Gate if the API has side effects; safe if read-only           |
| Read-only operations   | `search_db`, `list_files`, `get_record`, `fetch_url`                 | Never         | No side effects; no gating needed                             |

<Tip>
  Start with a conservative list and expand it. It is better to gate too many tools initially and relax the list based on analyst feedback than to discover after an incident that a tool was unguarded. Review the high-risk tool list quarterly with the AI system owners and the security team.
</Tip>

***

## Custom scorer for dangerous tool detection

```yaml theme={null}
# evals/scorers.yaml — production configuration
scorers:
  - name: hitl_tool_detector
    type: keyword_list
    severity: critical
    description: "Detects high-risk tool names in agent interactions"
    # Matches against both the request body (tool call) and response body (tool result)
    keywords:
      - transfer_funds
      - create_payment
      - refund_order
      - delete_record
      - drop_table
      - terminate_instance
      - purge_queue
      - send_email
      - post_to_social
      - create_pr
      - merge_pr
      - add_user
      - grant_role
      - update_permissions
      - delete_user
    webhook_on_match: true
    # Exclude red team runs from creating P1 alerts
    exclude_tags:
      - redteam
      - staging
      - test
```

***

## Compliance value

### OWASP LLM06 — Excessive Agency

HITL is the primary control for LLM06. The OWASP guidance states: "Limit the permissions of LLM-integrated systems and implement human-in-the-loop controls for actions that are irreversible or have significant impact." Document the approval gate as the LLM06 control in your security posture assessment.

### EU AI Act Article 14 — Human Oversight

For AI systems classified as "high-risk" under the EU AI Act (Annex III), Article 14 requires that natural persons are able to "oversee the functioning" of the AI system and "intervene or interrupt" it. A pre-execution HITL gate satisfies this requirement for the specific actions it covers. Document each tool category in your AI Act technical documentation.

### SOC 2 CC6.1 — Logical Access Controls

The approval service functions as a logical access control: the AI agent cannot execute a privileged action without an authorised human granting access for that specific invocation. Include the HITL gate in your CC6.1 control evidence:

> "High-risk tool calls made by AI agents require approval from an authorised human operator before execution. Approvals are logged in the Cognisafe audit trail with the approver identity, timestamp, and decision. The approval gate is implemented as a synchronous block in the agent tool wrapper and defaults to rejection on timeout."

***

## Operational considerations

### Latency

HITL adds human response time. An approval workflow that requires a human to click Approve adds seconds to minutes to hours of latency to the agent task. Design agent tasks accordingly:

* Set agent timeout longer than your approval SLA: if your Slack approval SLA is 5 minutes, set the agent task timeout to 10 minutes
* Inform end users that high-risk actions are queued for human review: "I need to transfer funds — this action is pending approval and will complete within 5 minutes"
* Define a business-hours policy: if approvals are only reviewed during working hours, agents should not initiate high-risk actions outside those hours unless there is an on-call roster

### Approval fatigue

If the approval queue receives too many requests, approvers start rubber-stamping without reviewing. Counter this by:

1. Calibrating the high-risk tool list — do not gate low-stakes operations
2. Providing maximum context in the approval message so the review takes 30 seconds, not 5 minutes
3. Setting a volume alert: if more than 10 approvals are pending simultaneously, something is wrong with the agent

### Failure mode: approval service unavailable

If the approval service is unavailable, the gate wrapper will raise an exception when it cannot submit the request. The agent will receive a tool error and should surface this to the user. This is the correct behaviour — unavailability of the approval service must not result in ungated execution. Design the gate wrapper to fail closed.

```python theme={null}
def gated_tool(tool_name: str, args: dict, agent_name: str = "unknown") -> dict:
    if tool_name not in HIGH_RISK_TOOLS:
        return execute_tool(tool_name, args)

    try:
        resp = httpx.post(f"{APPROVAL_SERVICE_URL}/requests", ...)
        resp.raise_for_status()
    except (httpx.ConnectError, httpx.TimeoutException, httpx.HTTPStatusError) as e:
        # Approval service unavailable — fail closed, never execute
        raise RuntimeError(
            f"Approval service unavailable for '{tool_name}'. "
            f"Action blocked. Contact the platform team. Error: {e}"
        ) from e
```