Human-in-the-Loop UX: Designing Control Surfaces for AI Agents

The Control Problem

AI agents can now browse the web, execute code, send emails, and make purchases. But with autonomy comes risk: a "Sorcerer's Apprentice" scenario where agents spiral out of control.

Why pure autonomy fails in practice: The promise of agents is "set it and forget it"—delegate a task and come back to a finished result. But pure autonomy fails for two reasons: (1) agents make mistakes, and the cost of an uncaught mistake compounds with autonomy level; (2) requirements change mid-task, and users need to steer. A booking agent that autonomously reserves the wrong hotel wastes money. An email agent that sends a draft prematurely damages relationships. HITL isn't about distrust—it's about managing the expected error rate of imperfect systems.

The efficiency paradox: More oversight means fewer mistakes but also less efficiency. The art of HITL design is finding the right trade-off: approve high-stakes actions, observe medium-stakes, ignore low-stakes. This isn't one-size-fits-all—a financial services agent needs more checkpoints than a code formatting agent. The patterns in this post help you calibrate control levels to your specific risk profile.

From UX research: "Most agent actions are asynchronous, meaning traditional synchronous web pages are a poor match. Start, stop, and pause buttons are a good starting point—otherwise you risk runaway automation."

This post covers the design patterns for keeping humans in control of AI agents—without destroying the efficiency that makes agents valuable.

The HITL Spectrum

Human-in-the-loop isn't binary. It's a spectrum of control levels:

Code

Full Manual          Supervised           Autonomous
     │                   │                    │
     ▼                   ▼                    ▼
┌─────────┐        ┌─────────┐         ┌─────────┐
│ Human   │        │ Agent   │         │ Agent   │
│ does    │───────▶│ proposes│────────▶│ executes│
│ task    │        │ Human   │         │ Human   │
│         │        │ approves│         │ monitors│
└─────────┘        └─────────┘         └─────────┘

Trust Level:  Low        Medium            High
Efficiency:   Low        Medium            High
Risk:         Low        Medium            High

Control Levels in Practice

Level	Agent Behavior	Human Role	Use Case
L0: Suggestion	Proposes action	Executes manually	High-risk financial
L1: Confirmation	Prepares action	Clicks "approve"	Email, purchases
L2: Notification	Executes, reports	Reviews async	Data processing
L3: Exception	Executes silently	Intervenes on error	Monitoring, logging
L4: Autonomous	Full independence	Periodic audit	Low-risk automation

Core Design Patterns

1. Interrupt and Resume

The fundamental pattern: agents can be paused mid-execution. This is harder than it sounds—an agent might be mid-way through a multi-step process when the user hits "pause." You need to track state carefully so you can resume exactly where you left off, or roll back to a clean state.

The key insight: treat every action as potentially interruptible. Before each step, check for pause requests. After each step, save enough state to resume. Design for the reality that users will hit pause at the worst possible moments.

Python

from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable

class AgentState(Enum):
    RUNNING = "running"
    PAUSED = "paused"
    WAITING_APPROVAL = "waiting_approval"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class InterruptPoint:
    reason: str
    context: dict
    options: list[str]
    callback: Callable

class InterruptibleAgent:
    def __init__(self):
        self.state = AgentState.RUNNING
        self.pending_interrupt: Optional[InterruptPoint] = None

    async def run(self, task):
        while self.state == AgentState.RUNNING:
            # Check for pause request
            if self.should_pause():
                self.state = AgentState.PAUSED
                await self.notify_paused()
                await self.wait_for_resume()

            # Execute next step
            action = await self.plan_next_action()

            # Check if action requires approval
            if self.requires_approval(action):
                self.state = AgentState.WAITING_APPROVAL
                approval = await self.request_approval(action)

                if not approval.granted:
                    action = approval.alternative or self.replan()

            await self.execute(action)

            if self.is_complete():
                self.state = AgentState.COMPLETED

    def pause(self):
        """External pause request"""
        self._pause_requested = True

    def resume(self):
        """Resume from paused state"""
        if self.state == AgentState.PAUSED:
            self.state = AgentState.RUNNING
            self._pause_requested = False

    async def request_approval(self, action):
        """Create approval request and wait for human response"""
        self.pending_interrupt = InterruptPoint(
            reason=f"Action requires approval: {action.type}",
            context=action.to_dict(),
            options=["Approve", "Modify", "Reject"],
            callback=self._handle_approval
        )

        await self.notify_approval_needed()
        return await self.wait_for_human_input()

2. Checkpoint Control

Define explicit checkpoints where human review is required:

Python

class CheckpointController:
    def __init__(self):
        self.checkpoints = {
            "before_external_api": CheckpointConfig(
                required=True,
                timeout_seconds=300,
                fallback="abort"
            ),
            "before_payment": CheckpointConfig(
                required=True,
                timeout_seconds=600,
                fallback="abort"
            ),
            "before_email_send": CheckpointConfig(
                required=True,
                timeout_seconds=120,
                fallback="draft"
            ),
            "before_file_delete": CheckpointConfig(
                required=True,
                timeout_seconds=60,
                fallback="abort"
            ),
        }

    async def checkpoint(self, name: str, context: dict) -> CheckpointResult:
        config = self.checkpoints.get(name)

        if not config or not config.required:
            return CheckpointResult(approved=True)

        # Present to human
        request = ApprovalRequest(
            checkpoint=name,
            context=context,
            options=["Approve", "Modify", "Reject"],
            timeout=config.timeout_seconds
        )

        try:
            response = await self.get_human_response(request)
            return response
        except TimeoutError:
            return self.handle_timeout(config.fallback, context)

UI for checkpoints:

Code

┌─────────────────────────────────────────────────────────┐
│  ⚠️  Checkpoint: Before Email Send                       │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  Agent wants to send the following email:               │
│                                                         │
│  To: client@example.com                                 │
│  Subject: Q4 Report Summary                             │
│  Body: [Preview of 500 chars...]                        │
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │ Full email preview (expandable)                 │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
│  [✓ Approve]   [✏️ Edit]   [✗ Reject]   [⏸ Pause Agent] │
│                                                         │
│  ⏱️ Auto-approve in 2:00 (save to drafts if no action)  │
└─────────────────────────────────────────────────────────┘

3. Confidence-Based Routing

Route to humans when the agent is uncertain:

Python

class ConfidenceRouter:
    def __init__(self, confidence_threshold=0.8):
        self.threshold = confidence_threshold

    async def route(self, agent_response):
        confidence = agent_response.confidence_score

        if confidence >= self.threshold:
            # High confidence: proceed automatically
            return Decision(
                action="proceed",
                requires_human=False
            )
        elif confidence >= 0.5:
            # Medium confidence: notify but proceed
            await self.notify_human(
                message=f"Agent proceeding with {confidence:.0%} confidence",
                context=agent_response,
                action_required=False
            )
            return Decision(action="proceed", requires_human=False)
        else:
            # Low confidence: require human decision
            return Decision(
                action="await_human",
                requires_human=True,
                reason=f"Low confidence ({confidence:.0%})",
                suggestions=agent_response.alternatives
            )

    def extract_confidence(self, response):
        """Extract confidence from LLM response"""
        # Option 1: Explicit confidence in response
        if hasattr(response, 'confidence'):
            return response.confidence

        # Option 2: Parse from structured output
        if "<confidence>" in response.text:
            match = re.search(r'<confidence>(\d+\.?\d*)</confidence>', response.text)
            if match:
                return float(match.group(1))

        # Option 3: Calibrated logprobs
        if response.logprobs:
            return self.calibrate_logprobs(response.logprobs)

        return 0.5  # Default medium confidence

4. Progressive Autonomy

Start conservative, increase autonomy as trust builds:

Python

class ProgressiveAutonomy:
    def __init__(self, user_id: str):
        self.user_id = user_id
        self.trust_score = self.load_trust_score()

    def load_trust_score(self) -> float:
        """Load historical trust score for user"""
        history = self.db.get_user_history(self.user_id)

        if not history:
            return 0.0  # New user starts at lowest autonomy

        # Calculate based on past interactions
        successful = sum(1 for h in history if h.outcome == "success")
        total = len(history)
        rejection_rate = sum(1 for h in history if h.human_rejected) / total

        return (successful / total) * (1 - rejection_rate)

    def get_autonomy_level(self, action_type: str) -> int:
        """Determine autonomy level for action type"""
        base_level = self.trust_score * 4  # 0-4 scale

        # Adjust by action risk
        risk_adjustments = {
            "read_file": 1,
            "search_web": 1,
            "write_file": -1,
            "send_email": -2,
            "make_payment": -3,
            "delete_data": -3,
        }

        adjustment = risk_adjustments.get(action_type, 0)
        final_level = max(0, min(4, base_level + adjustment))

        return int(final_level)

    def update_trust(self, action, outcome, human_feedback):
        """Update trust score based on interaction outcome"""
        if outcome == "success" and human_feedback == "approved":
            self.trust_score = min(1.0, self.trust_score + 0.02)
        elif outcome == "success" and human_feedback == "modified":
            self.trust_score = min(1.0, self.trust_score + 0.01)
        elif human_feedback == "rejected":
            self.trust_score = max(0.0, self.trust_score - 0.05)
        elif outcome == "failed":
            self.trust_score = max(0.0, self.trust_score - 0.1)

        self.save_trust_score()

UI for progressive autonomy:

Code

┌─────────────────────────────────────────────────────────┐
│  Agent Autonomy Settings                                │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  Current Trust Level: ████████░░ 78%                    │
│  Based on 156 successful interactions                   │
│                                                         │
│  Action Permissions:                                    │
│                                                         │
│  Web Search        [████████████] Autonomous            │
│  Read Files        [████████████] Autonomous            │
│  Write Files       [████████░░░░] Notify                │
│  Send Emails       [████░░░░░░░░] Confirm               │
│  Make Purchases    [░░░░░░░░░░░░] Always Ask            │
│                                                         │
│  [Override All: Ask Before Everything]                  │
│                                                         │
└─────────────────────────────────────────────────────────┘

5. Kill Switch and Rollback

The "oh no" button. When an agent goes off the rails—sending wrong emails, deleting files, making bad API calls—users need immediate control. A kill switch must be fast (sub-second), reliable (works even if the agent is in a bad state), and comprehensive (stops everything the agent might be doing).

Rollback is trickier. Not all actions are reversible: you can't unsend an email that's been delivered, can't undo a purchase that's been processed. Track which actions are reversible, maintain backups/snapshots for file operations, and be honest with users about what can and can't be undone.

Python

class AgentKillSwitch:
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.action_log = []

    async def emergency_stop(self):
        """Immediately halt agent and all spawned processes"""
        # 1. Set global stop flag
        await self.redis.set(f"agent:{self.agent_id}:stop", "1")

        # 2. Kill any running subprocesses
        for pid in self.tracked_processes:
            os.kill(pid, signal.SIGTERM)

        # 3. Revoke API credentials
        await self.revoke_credentials()

        # 4. Log emergency stop
        await self.log_event("emergency_stop", {
            "timestamp": datetime.utcnow(),
            "pending_actions": self.get_pending_actions(),
            "triggered_by": "user"
        })

        return StopResult(
            stopped=True,
            rollback_available=len(self.action_log) > 0,
            pending_actions_cancelled=len(self.get_pending_actions())
        )

    async def rollback(self, steps: int = 1):
        """Undo recent agent actions"""
        rollback_actions = self.action_log[-steps:]

        for action in reversed(rollback_actions):
            if action.reversible:
                await self.reverse_action(action)
            else:
                await self.notify_user(
                    f"Cannot reverse: {action.description}"
                )

    async def reverse_action(self, action):
        """Reverse a specific action"""
        reversals = {
            "file_write": lambda a: self.restore_file_backup(a.path),
            "file_delete": lambda a: self.restore_from_trash(a.path),
            "api_call": lambda a: self.call_undo_endpoint(a),
            "email_send": lambda a: self.recall_email(a) if a.recallable else None,
        }

        reversal = reversals.get(action.type)
        if reversal:
            await reversal(action)

What can go wrong:

Email recall fails: Gmail/Outlook recall only works within seconds and only if the recipient hasn't read it. After that, the email is permanent. Solution: hold emails in a "pending" queue for 30 seconds before actual send, giving users a reliable undo window.
API calls with side effects: A "delete user" API call can't be undone by calling it again. Solution: prefer soft-delete APIs, or implement your own undo log that can restore state.
File backup not found: If the agent deleted a file before you had a chance to back it up, rollback fails. Solution: pre-backup before any destructive operation, even if it adds latency.
Partial rollback: Rolling back step 3 of 5 might leave the system in an inconsistent state. Solution: treat multi-step operations as transactions—rollback everything or nothing.

Kill switch UI:

Code

┌─────────────────────────────────────────────────────────┐
│  🔴 EMERGENCY CONTROLS                                  │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  Agent Status: ● Running (Task: "Process invoices")     │
│  Runtime: 4m 23s | Actions: 12 | Pending: 3             │
│                                                         │
│  ┌───────────────┐  ┌───────────────┐                   │
│  │  ⏸️ PAUSE     │  │  🛑 STOP      │                   │
│  │  (Resumable)  │  │  (Immediate)  │                   │
│  └───────────────┘  └───────────────┘                   │
│                                                         │
│  ┌───────────────────────────────────────────────┐     │
│  │  ↩️ ROLLBACK LAST ACTION                       │     │
│  │  Last: "Sent email to vendor@example.com"     │     │
│  └───────────────────────────────────────────────┘     │
│                                                         │
│  Recent Actions:                                        │
│  ✓ 4:23 - Read invoice_march.pdf                       │
│  ✓ 4:21 - Extracted line items                         │
│  ⚠️ 4:20 - Sent email to vendor@example.com            │
│  ✓ 4:18 - Created expense entry                        │
│                                                         │
└─────────────────────────────────────────────────────────┘

UI/UX Design Principles

1. Status Visibility

Always show what the agent is doing:

Code

┌─────────────────────────────────────────────────────────┐
│  Agent: Research Assistant                              │
│  Task: "Find competitor pricing for Q1 report"          │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ● Currently: Searching competitor websites             │
│                                                         │
│  Progress:                                              │
│  ✓ Identified 5 competitors                             │
│  ✓ Scraped pricing from CompetitorA.com                │
│  ● Scraping CompetitorB.com (2/5 pages)                │
│  ○ CompetitorC.com                                      │
│  ○ CompetitorD.com                                      │
│  ○ Compile comparison report                            │
│                                                         │
│  Estimated completion: ~3 minutes                       │
│                                                         │
│  [⏸️ Pause]  [🛑 Stop]  [📋 View Details]               │
│                                                         │
└─────────────────────────────────────────────────────────┘

2. Meaningful Interruptions

Don't interrupt for trivial decisions:

Python

class InterruptionPolicy:
    """Decide what warrants interrupting the user"""

    def should_interrupt(self, action, context) -> tuple[bool, str]:
        # Never interrupt for read-only operations
        if action.type in ["read", "search", "analyze"]:
            return False, "read_only"

        # Always interrupt for irreversible actions
        if action.irreversible:
            return True, "irreversible"

        # Interrupt based on value threshold
        if action.estimated_cost > self.user_preferences.cost_threshold:
            return True, "cost_threshold"

        # Interrupt for external communications
        if action.type in ["email", "message", "post"]:
            return True, "external_communication"

        # Batch low-priority approvals
        if action.priority == "low":
            self.queue_for_batch(action)
            return False, "batched"

        return False, "default_allow"

3. Context-Rich Approval Requests

When asking for approval, provide full context:

Code

┌─────────────────────────────────────────────────────────┐
│  🔔 Approval Required                                   │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  The agent wants to: Purchase software license          │
│                                                         │
│  Details:                                               │
│  ├─ Vendor: Acme Software Inc.                         │
│  ├─ Product: Enterprise Analytics Suite                 │
│  ├─ Cost: $2,499/year                                  │
│  ├─ Payment: Corporate card ending in 4242             │
│  └─ Renewal: Auto-renew annually                       │
│                                                         │
│  Why: User requested "set up analytics for Q1"         │
│  Agent reasoning: "This tool best matches requirements  │
│  for dashboard creation and data visualization..."      │
│                                                         │
│  Alternatives considered:                               │
│  • Tool B ($1,999/yr) - Missing real-time features     │
│  • Tool C ($3,200/yr) - Over budget                    │
│                                                         │
│  [✓ Approve]  [✏️ Choose Alternative]  [✗ Reject]      │
│                                                         │
└─────────────────────────────────────────────────────────┘

4. Async Notification Patterns

Not everything needs immediate attention:

Python

class NotificationManager:
    def __init__(self, user_preferences):
        self.prefs = user_preferences

    async def notify(self, event: AgentEvent):
        urgency = self.classify_urgency(event)

        if urgency == "immediate":
            # Push notification + sound
            await self.push_notification(event, sound=True)

        elif urgency == "soon":
            # Push notification, no sound
            await self.push_notification(event, sound=False)

        elif urgency == "batched":
            # Add to digest
            self.add_to_digest(event)

        elif urgency == "log_only":
            # Just log, no notification
            self.log(event)

    def classify_urgency(self, event):
        if event.requires_action and event.timeout_minutes < 5:
            return "immediate"
        if event.requires_action:
            return "soon"
        if event.type in ["error", "warning"]:
            return "soon"
        if event.type == "completion":
            return "batched"
        return "log_only"

Implementation: Amazon Bedrock Example

AWS provides built-in HITL for Bedrock Agents:

Python

import boto3

bedrock_agent = boto3.client('bedrock-agent-runtime')

def invoke_agent_with_hitl(agent_id, session_id, prompt):
    response = bedrock_agent.invoke_agent(
        agentId=agent_id,
        agentAliasId='TSTALIASID',
        sessionId=session_id,
        inputText=prompt,
        enableTrace=True
    )

    for event in response['completion']:
        if 'returnControl' in event:
            # Agent is requesting human input
            invocation = event['returnControl']['invocationInputs'][0]

            if invocation['invocationType'] == 'ACTION_GROUP_INVOCATION':
                # Present action to user
                action = invocation['actionGroupInvocationInput']
                user_decision = present_approval_ui(action)

                # Continue with user's decision
                return continue_agent(
                    session_id,
                    invocation['invocationId'],
                    user_decision
                )

        elif 'chunk' in event:
            # Normal response chunk
            yield event['chunk']['bytes'].decode()

Measuring HITL Effectiveness

Track these metrics:

Python

class HITLMetrics:
    def track(self):
        return {
            # Efficiency metrics
            "approval_rate": self.approvals / self.total_requests,
            "modification_rate": self.modifications / self.total_requests,
            "rejection_rate": self.rejections / self.total_requests,

            # Time metrics
            "avg_approval_time_seconds": self.avg_approval_time,
            "timeout_rate": self.timeouts / self.total_requests,

            # Quality metrics
            "post_approval_success_rate": self.successful_after_approval / self.approvals,
            "caught_errors": self.errors_caught_by_human,
            "missed_errors": self.errors_after_approval,

            # User experience
            "interruption_frequency": self.interruptions_per_hour,
            "user_override_rate": self.user_overrides / self.agent_suggestions,
        }

Target benchmarks:

Approval rate: 85-95% (if lower, agent is too cautious)
Avg approval time: <30 seconds (if higher, UI needs work)
Timeout rate: <5% (if higher, timeouts too short)
Post-approval success: >98% (if lower, approval context insufficient)

Conclusion

Human-in-the-loop isn't about limiting AI agents—it's about building trust and safety into autonomous systems:

Design for the spectrum from full supervision to full autonomy
Make interruptions meaningful with rich context
Enable recovery with pause, stop, and rollback
Build trust progressively based on demonstrated reliability
Measure and optimize the human-agent interaction

The best HITL systems are invisible when things go well and immediately available when intervention is needed.

Human-in-the-Loop UX: Designing Control Surfaces for AI Agents

Table of Contents

The Control Problem

The HITL Spectrum

Control Levels in Practice

Core Design Patterns

1. Interrupt and Resume

2. Checkpoint Control

3. Confidence-Based Routing

4. Progressive Autonomy

5. Kill Switch and Rollback

UI/UX Design Principles

1. Status Visibility

2. Meaningful Interruptions

3. Context-Rich Approval Requests

4. Async Notification Patterns

Implementation: Amazon Bedrock Example

Measuring HITL Effectiveness

Conclusion

Frequently Asked Questions

Enrico Piovano, PhD

Related Articles

Building Agentic AI Systems: A Complete Implementation Guide

Agentic AI Compliance: Liability, Legal Frameworks, and Risk Management

Advanced Chatbot Architectures: Beyond Simple Q&A