Human oversight design

Human Oversight Design

Autonomous AI agents that act without meaningful human control are a governance failure waiting to happen. This skill teaches you to design oversight into the system architecture — not bolt it on afterwards.

The Oversight Spectrum

Before writing any agent code, decide where your system sits:

Full Automation          Supervised Automation       Human-Led + AI-Assisted
─────────────────────────────────────────────────────────────────────────
AI decides & acts        AI decides, human approves  Human decides, AI informs
     │                          │                              │
High risk                   Medium risk                    Low risk
Requires audit trail    Requires override mechanism    Requires transparency

Rule: Systems affecting citizens, finances, health, or legal status must sit at "Supervised Automation" or further right. Argue in writing if you believe otherwise.

---

Step 1: Map Every Action to a Control Level

For every action your agent can take, assign a control level before implementation:

# Define your action registry with explicit oversight requirements
ACTION_REGISTRY = {
    # (action_name): (control_level, rationale)
    'read_database':          ('autonomous',  'Read-only, no external effect'),
    'generate_draft_email':   ('autonomous',  'Draft only, not sent'),
    'search_web':             ('autonomous',  'Read-only, no external effect'),
    'send_email':             ('supervised',  'External communication, irreversible'),
    'write_file':             ('supervised',  'Modifies state, potentially irreversible'),
    'delete_record':          ('supervised',  'Irreversible'),
    'submit_form':            ('supervised',  'External action with consequences'),
    'make_payment':           ('supervised',  'Financial transaction'),
    'update_citizen_record':  ('supervised',  'Affects individual entitlements'),
    'escalate_to_authority':  ('supervised',  'Triggers further action'),
    'take_legal_action':      ('blocked',     'Must not be automated'),
}

CONTROL_LEVELS = {
    'autonomous':  'Agent executes without approval',
    'supervised':  'Agent proposes, human must approve before execution',
    'blocked':     'Agent must not perform this action under any circumstances',
}

---

Step 2: Implement the Approval Gate

Every supervised action MUST pass through a gate:

from typing import Callable, Any
import logging

logger = logging.getLogger(__name__)

class OversightGate:
    """
    Enforces human approval for supervised actions.
    
    In production: integrates with your notification/approval system.
    In testing: can be configured to auto-approve or auto-reject.
    """
    
    def __init__(self, approval_fn: Callable, timeout_seconds: int = 3600):
        self.approval_fn = approval_fn
        self.timeout_seconds = timeout_seconds
    
    def request_approval(
        self,
        action: str,
        parameters: dict,
        agent_reasoning: str,
        risk_level: str = 'medium',
    ) -> bool:
        """
        Request human approval for a supervised action.
        
        Returns True if approved, False if rejected or timed out.
        Logs all decisions to audit trail.
        """
        request = {
            'action': action,
            'parameters': parameters,
            'agent_reasoning': agent_reasoning,
            'risk_level': risk_level,
            'requested_at': datetime.now(timezone.utc).isoformat(),
        }
        
        logger.info(f"OVERSIGHT_REQUEST: {json.dumps(request)}")
        
        approved = self.approval_fn(request, timeout=self.timeout_seconds)
        
        log_entry = {**request, 'approved': approved, 'decided_at': datetime.now(timezone.utc).isoformat()}
        logger.info(f"OVERSIGHT_DECISION: {json.dumps(log_entry)}")
        
        return approved
    
    def is_blocked(self, action: str) -> bool:
        level = ACTION_REGISTRY.get(action, ('supervised', 'unknown'))[0]
        return level == 'blocked'
    
    def requires_approval(self, action: str) -> bool:
        level = ACTION_REGISTRY.get(action, ('supervised', 'unknown'))[0]
        return level == 'supervised'


def execute_with_oversight(
    gate: OversightGate,
    action: str,
    parameters: dict,
    action_fn: Callable,
    agent_reasoning: str,
) -> Any:
    """
    Execute an agent action with appropriate oversight controls.
    
    This is the ONLY way agents should execute actions.
    Do not bypass this function.
    """
    if gate.is_blocked(action):
        raise PermissionError(
            f"Action '{action}' is blocked. Agents must not perform this action. "
            f"Escalate to a human operator."
        )
    
    if gate.requires_approval(action):
        approved = gate.request_approval(
            action=action,
            parameters=parameters,
            agent_reasoning=agent_reasoning,
        )
        if not approved:
            logger.warning(f"OVERSIGHT_REJECTED: action={action}")
            return None  # agent must handle rejection gracefully
    
    # Execute and log
    logger.info(f"AGENT_ACTION: {action} with params {parameters}")
    result = action_fn(**parameters)
    logger.info(f"AGENT_RESULT: {action} completed")
    return result

---

Step 3: Design for Corrigibility

Your agent must be stoppable at any point:

import threading

class AgentController:
    """Controls agent execution with pause and stop capability."""
    
    def __init__(self):
        self._stop_event = threading.Event()
        self._pause_event = threading.Event()
        self._pause_event.set()  # start unpaused
    
    def stop(self) -> None:
        """Immediately stop the agent after current step completes."""
        self._stop_event.set()
        logger.warning("AGENT_STOP_REQUESTED")
    
    def pause(self) -> None:
        """Pause the agent until resumed."""
        self._pause_event.clear()
        logger.info("AGENT_PAUSED")
    
    def resume(self) -> None:
        self._pause_event.set()
        logger.info("AGENT_RESUMED")
    
    def should_stop(self) -> bool:
        return self._stop_event.is_set()
    
    def check_pause(self) -> None:
        """Call this between every agent step. Blocks if paused."""
        self._pause_event.wait()  # blocks until resumed
    
    def run_step(self, step_fn: Callable) -> Any:
        """Execute one agent step with stop/pause checking."""
        if self.should_stop():
            raise AgentStoppedException("Agent was stopped by operator.")
        self.check_pause()
        return step_fn()

---

Step 4: Audit Trail — Every Action, Every Decision

class AgentAuditLog:
    """Immutable audit log for all agent actions and decisions."""
    
    def log_step(
        self,
        step_id: str,
        action: str,
        reasoning: str,
        inputs: dict,
        outputs: Any,
        human_approved: bool,
        approver: str | None = None,
    ) -> None:
        entry = {
            'step_id': step_id,
            'timestamp': datetime.now(timezone.utc).isoformat(),
            'action': action,
            'reasoning': reasoning,
            'inputs_summary': str(inputs)[:500],
            'outputs_summary': str(outputs)[:500],
            'human_approved': human_approved,
            'approver': approver,
        }
        # Write to append-only log — never delete entries
        with open('agent_audit.jsonl', 'a') as f:
            f.write(json.dumps(entry) + '\n')

---

Step 5: Override Rate Monitoring

Track how often humans override or reject agent proposals. A very low override rate may indicate rubber-stamping (automation bias), not good performance:

def compute_override_rate(audit_log_path: str, window_days: int = 30) -> float:
    """Compute the human override rate over the last N days."""
    ...
    # Alert if override rate drops below 5% for supervised actions
    # This may indicate humans are not exercising meaningful oversight
    if override_rate < 0.05:
        logger.warning(
            f"OVERSIGHT_ALERT: Override rate is {override_rate:.1%} — "
            f"possible automation bias. Review oversight process."
        )
    return override_rate

---

Completion Checklist

• All agent actions classified in ACTION_REGISTRY with control level and rationale
• OversightGate implemented for all supervised actions
• Agent corrigibility implemented (stop/pause/resume)
• Audit log captures every step, decision, and human approval
• Override rate monitoring set up
• System prompt or agent instructions explicitly state agent's boundaries
• Tested: agent stops correctly when stop signal received
• Tested: agent requests approval before executing supervised actions

Proceed to governance-documentation before deploying this system.