LeetLLM
My PlanLearnGlossaryTracksPracticeBlog
LeetLLM

Your go-to resource for mastering AI & LLM systems.

Product

  • Learn
  • Glossary
  • Tracks
  • Practice
  • Blog
  • RSS

Legal

  • Terms of Service
  • Privacy Policy

© 2026 LeetLLM. All rights reserved.

All Topics
Your Progress
0%

0 of 158 articles completed

🛠️Computing Foundations0/9
Git, Shell, Linux for AIDocker for Reproducible AIPython for AI EngineeringNumPy and Tensor ShapesCUDA for ML TrainingMPS & Metal for ML on MacData Structures for AISQL and Data ModelingAlgorithms for ML Engineers
📊Math & Statistics0/8
Gradients and BackpropVectors, Matrices & TensorsLinear Algebra for MLAdam, Momentum, SchedulersProbability for Machine LearningStatistics and UncertaintyDistributions and SamplingHypothesis Tests, Intervals, and pass@k
📚Preparation & Prerequisites0/13
Neural Networks from ScratchCNNs from ScratchTraining & BackpropagationSoftmax, Cross-Entropy & OptimizationRNNs, LSTMs, GRUs, and Sequence ModelingAutoencoders and VAEsThe Transformer Architecture End-to-EndLanguage Modeling & Next TokensFrom GPT to Modern LLMsPrompt Engineering FundamentalsCalling LLM APIs in ProductionFirst AI App End-to-EndThe LLM Lifecycle
🧮ML Algorithms & Evaluation0/11
Linear Regression from ScratchLogistic Regression and MetricsDecision Trees, Forests, and BoostingReinforcement Learning BasicsValidation and LeakageClustering and PCACore Retrieval AlgorithmsDecoding AlgorithmsExperiment Design and A/B TestingPyTorch Training LoopsDataset Pipelines and Data Quality
📦Production ML Systems0/6
Feature Engineering for Production MLBatch and Streaming Feature PipelinesGradient Boosted Trees in ProductionRanking and Recommendation SystemsForecasting and Anomaly DetectionMonitoring Predictive Models
🧪Core LLM Foundations0/8
The Bitter Lesson & ComputeBPE, WordPiece, and SentencePieceStatic to Contextual EmbeddingsPerplexity & Model EvaluationFile Ingestion for AIChunking StrategiesLLM Benchmarks & LimitationsInstruction Tuning & Chat Templates
🧰Applied LLM Engineering0/24
Dimensionality Reduction for EmbeddingsCoT, ToT & Self-Consistency PromptingFunction Calling & Tool UseMCP & Tool Protocol StandardsContext EngineeringPrompt Injection DefenseResponsible AI GovernanceData Labeling and Human FeedbackEvaluating AI AgentsProduction RAG PipelinesHybrid Search: Dense + SparseReranking and Cross-Encoders for RAGRAG Evaluation for Reliable AnswersLLM-as-a-Judge EvaluationBias & Fairness in LLMsHallucination Detection & MitigationLLM Observability & MonitoringExperiment Tracking with MLflow and W&BPrompt Optimization with DSPyModel Versioning & DeploymentSemantic Caching & Cost OptimizationLLM Cost Engineering & Token EconomicsModel Gateways, Routing, and FallbacksDesign an Automated Support Agent
🎓Portfolio Capstones0/8
Capstone: Delivery ETA PredictionCapstone: Product RankingCapstone: Demand ForecastingCapstone: Image Damage ClassifierCapstone: Production ML PipelineCapstone: Document QACapstone: Eval DashboardCapstone: Fine-Tuned Classifier
🧠Transformer Deep Dives0/8
Sentence Embeddings & Contrastive LossEmbedding Similarity & QuantizationScaled Dot-Product AttentionVision Transformers and Image EncodersPositional Encoding: RoPE & ALiBiLayer Normalization: Pre-LN vs Post-LNMechanistic InterpretabilityDecoding Strategies: Greedy to Nucleus
🧬Advanced Training & Adaptation0/15
Scaling Laws & Compute-Optimal TrainingPre-training Data at ScaleBuild GPT from Scratch LabContinued Pretraining for Domain ShiftSynthetic Data PipelinesSupervised Fine-Tuning PipelineMixed Precision TrainingDistributed Training: FSDP & ZeROLoRA & Parameter-Efficient TuningReward Modeling from Preference DataRLHF & DPO AlignmentConstitutional AI & Red TeamingRLVR & Verifiable RewardsKnowledge Distillation for LLMsModel Merging and Weight Interpolation
🤖Advanced Agents & Retrieval0/16
Vector DB Internals: HNSW & IVFAdvanced RAG: HyDE & Self-RAGGraphRAG & Knowledge GraphsRAG Security & Access ControlStructured Output GenerationReAct & Plan-and-ExecuteGuardrails & Safety FiltersCode Generation & SandboxingComputer-Use / GUI / Browser AgentsHuman-in-the-Loop Agent ArchitectureAI Coding Workflow with AgentsAgent Memory & PersistenceAgent Failure & RecoveryRecursive Language Models (RLM)Multi-Agent OrchestrationCapstone: Production Agent
⚡Inference & Production Scale0/19
Inference: TTFT, TPS & KV CacheMulti-Query & Grouped-Query AttentionKV Cache & PagedAttentionPrefix Caching and Prompt CachingFlashAttention & Memory EfficiencyContinuous Batching & SchedulingScaling LLM InferenceModel Parallelism for LLM InferenceModel Quantization: GPTQ, AWQ & GGUFLocal LLM DeploymentSLM Specialization & Edge DeploymentSpeculative DecodingLong Context Window ManagementMixture of Experts ArchitectureMamba & State Space ModelsReasoning & Test-Time ComputeAdvanced MLOps & DevOps for AIGPU Serving & AutoscalingA/B Testing for LLMs
🏗️System Design Capstones0/9
Content Moderation SystemCode Completion SystemMulti-Tenant LLM PlatformLLM-Powered Search EngineVision-Language Models & CLIPMultimodal LLM ArchitectureDiffusion Models: Images & TextReal-Time Voice AI AgentReasoning & Test-Time Compute
🎤AI Lab Interviewing0/4
AI Lab Coding Interview: Python SystemsAI Lab System Design InterviewAI Lab Behavioral InterviewAI Lab Technical Presentation
Back to Topics
LearnAdvanced Agents & RetrievalHuman-in-the-Loop Agent Architecture
🤖HardLLM Agents & Tool Use

Human-in-the-Loop Agent Architecture

Build approval gates, durable checkpoints, and guarded resumes for agent actions that change external state.

39 min read
Learning path
Step 120 of 158 in the full curriculum
Computer-Use / GUI / Browser AgentsAI Coding Workflow with Agents

Computer-use agents can click buttons, fill forms, and operate real software. That makes the next production question unavoidable: what happens when the agent is about to spend money, change user data, send an external message, or deploy code?

Human-in-the-Loop (HITL) architecture adds a hard execution boundary for those moments. An agent can propose a side effect, but a downstream policy service pauses it until the required reviewer authorizes the exact action.

This addresses the opaque execution problem: even capable models can produce confident but incorrect outputs, and in production AI systems, one bad model promotion or rollback can degrade real users. HITL doesn't make an action correct by itself. It creates a point where policy, evidence, authorization, and final-state checks can stop a bad action.

This is also one practical control for one of the top agent risks. The OWASP Top 10 for LLM Applications lists LLM06: Excessive Agency, which it breaks into excessive functionality, excessive permissions, and excessive autonomy. Requiring human approval for high-impact actions is one recommended mitigation for the excessive-autonomy facet, ideally enforced in a downstream system rather than left to the model to decide. [1]Reference 1OWASP Top 10 for Large Language Model Applicationshttps://genai.owasp.org/llm-top-10/ Treat the pattern below as one implementation of that control.

Like a release bot that can read evaluation results but must route a production traffic change to an authorized reviewer, a HITL agent runs only within a defined permission boundary. Unlike a chatbot that waits for input at every turn, this pattern supervises actions according to their possible effect.

Follow Vega, a model-release assistant for an internal AI platform. Vega can read authorized evaluation runs, inspect deployment metrics, and draft release notes. Vega can also propose a model promotion, traffic shift, rollback, or incident update, but a downstream gate controls whether any of those side effects run. The traffic thresholds below are illustrative policy choices, not universal rules.

Design Vega's risk policy, implement checkpoint/resume logic that lets Vega wait durably for review, and build an approval UI that gives reviewers enough evidence without exposing unnecessary deployment data. This lesson assumes you already know large language model (LLM) tool calling, state graphs, guardrail policy, and browser-agent risk. If those terms are fuzzy, review Function Calling & Tool Use, Agentic Architectures, Guardrails & Safety Filters, and Computer Use & Browser Agents first.

The three pillars of human-agent interaction

Not all human oversight is created equal. Production systems distinguish between three distinct interaction models based on risk tolerance and the nature of the task. Those distinctions determine where approval, supervision, and audit controls belong.

Human-in-the-Loop (HITL): Active gatekeeping

In this model, the agent can't proceed with a gated action without explicit human approval. Use it for actions whose impact isn't acceptable to discover after execution, such as money movement, production deployment, or user-record change. The agent proposes an action, pauses execution, and waits for an approve, reject, or modified proposal decision before continuing. Modified proposals still need validation.

Human-on-the-Loop (HOTL): Supervisory control

Here the agent performs tasks autonomously while a human monitors the process and retains the ability to interrupt or override. This fits actions whose effect can be stopped or repaired after detection, such as routing internal draft suggestions into a queue. It isn't a sufficient gate for shifting production traffic or publishing an incident update: the human may see the mistake only after the effect happened. HITL blocks before execution, while HOTL can only interrupt execution that's already allowed.

Human-out-of-the-Loop (HOOTL): Full autonomy

At the far end of the spectrum, some actions are automated with no real-time human oversight. This applies only to bounded actions with clear authorization and acceptable failure modes. An authorized evaluation read or a draft created in an internal queue may fit; an unrestricted database read or web action doesn't become low risk merely because it doesn't write data. Some actions stay HITL because their effect, policy, or applicable duties demand a pre-execution gate.

The trust spectrum

Not all agent actions carry the same risk. Designing a HITL system begins by classifying every tool and action along a trust spectrum. This principle drives the level of friction required before execution.

A naive implementation treats all actions equally: either everything is autonomous (dangerous) or everything requires approval (tedious). A usable HITL system defines granular policies:

Risk LevelDescriptionExamplesInteraction Model
LowBounded, authorized read or isolated draftRead permitted eval metrics, Inspect canary health, Draft release note without publishingAuto-Execute: Run within least-privilege access.
MediumInternal staged change with a recoverable effectQueue review packet, Add a permitted internal risk labelPolicy Decision: Audit and notify, or require review if policy says so.
HighExternal or production side effectPublish incident update, Promote model to canary, Shift endpoint trafficApprove: Pause and wait for an authorized reviewer.
CriticalDestructive, unusually high-impact, or legally constrainedDelete retained eval evidence, Promote despite failing gate, Disable required safety filterEscalate or Block: Require stronger authorization or deny.

Risk classification system

Start with a RiskLevel enum and a policy lookup table. That decouples the agent's logic ("I want to do X") from the governance logic ("Should I allow X?").

Human-approval policy board that routes actions from auto-run reads through audit, approval, and escalated review as side effects become stronger or more irreversible. Human-approval policy board that routes actions from auto-run reads through audit, approval, and escalated review as side effects become stronger or more irreversible.
Start with static tiers. Reads and drafts can run directly, internal changes stay auditable, writes pause for approval, and regulated or irreversible actions escalate.

To implement this, we define a policy lookup table that maps each available tool to a specific risk tier. This structure is the core policy engine: it takes a tool's name and runtime arguments as contextual input, then returns the required authorization flow before execution.

risk-classification-system.py
1from enum import Enum 2from dataclasses import dataclass 3 4class RiskLevel(Enum): 5 AUTO = "auto" # Execute immediately, no human needed 6 NOTIFY = "notify" # Execute and notify human asynchronously 7 APPROVE = "approve" # Pause and wait for human approval 8 ESCALATE = "escalate" # Route to senior reviewer or stronger authorization 9 10@dataclass 11class ToolPolicy: 12 tool_name: str 13 risk_level: RiskLevel 14 escalate_above_traffic_percent: float | None = None 15 requires_reason: bool = False 16 timeout_minutes: int = 60 # Auto-reject after timeout 17 18TOOL_POLICIES = { 19 # Safe actions run automatically 20 "read_eval_run": ToolPolicy("read_eval_run", RiskLevel.AUTO), 21 "inspect_service_metrics": ToolPolicy("inspect_service_metrics", RiskLevel.AUTO), 22 23 # External communication requires justification 24 "publish_incident_update": ToolPolicy( 25 "publish_incident_update", 26 RiskLevel.APPROVE, 27 requires_reason=True, 28 ), 29 30 # Model promotions affect production users, so they always begin at APPROVE. 31 "promote_model": ToolPolicy( 32 "promote_model", 33 RiskLevel.APPROVE, 34 escalate_above_traffic_percent=25.0, 35 timeout_minutes=30, 36 ), 37 38 # Destructive or high-impact actions require escalation 39 "delete_eval_evidence": ToolPolicy("delete_eval_evidence", RiskLevel.ESCALATE), 40 "disable_safety_filter": ToolPolicy("disable_safety_filter", RiskLevel.ESCALATE), 41} 42 43RISK_PRIORITY = { 44 RiskLevel.AUTO: 0, 45 RiskLevel.NOTIFY: 1, 46 RiskLevel.APPROVE: 2, 47 RiskLevel.ESCALATE: 3, 48} 49 50def escalate_risk(current: RiskLevel, target: RiskLevel) -> RiskLevel: 51 return target if RISK_PRIORITY[target] > RISK_PRIORITY[current] else current 52 53print("read_eval_run:", TOOL_POLICIES["read_eval_run"].risk_level.value) 54print( 55 "promotion traffic threshold:", 56 TOOL_POLICIES["promote_model"].escalate_above_traffic_percent, 57) 58print("critical wins:", escalate_risk(RiskLevel.ESCALATE, RiskLevel.APPROVE).value)
Output
1read_eval_run: auto 2promotion traffic threshold: 25.0 3critical wins: escalate

Risk isn't static. A "send email" tool might be Low Risk when emailing an internal test address, but High Risk when emailing an external domain. Production policies are dynamic, checking arguments at runtime.

Architecture: the Checkpoint/Resume pattern

The fundamental architectural challenge of HITL is the pause. When an agent pauses for approval, it can't sleep the thread (time.sleep()) or block in memory. A guarded compare-and-swap (CAS) update later lets only one current approval resolve the pending row.

Human-in-the-loop checkpoint and resume trace showing proposal, approval pause, durable checkpoint, reviewer decision, guarded resume, and final execution. Human-in-the-loop checkpoint and resume trace showing proposal, approval pause, durable checkpoint, reviewer decision, guarded resume, and final execution.
Pause before side effects, persist durable state, then resume only through guarded decision path so stale clicks can't replay execution.

Why in-memory blocking fails

  1. Durability: If the server restarts or crashes while waiting for approval (which could take hours), the agent's state is lost.
  2. Resource Usage: Holding a thread open for a human response wastes compute resources.
  3. Scalability: You can't scale to thousands of concurrent agents if each one is blocking a thread.

Use the Checkpoint/Resume pattern. When an approval is needed, the agent persists its graph state (messages, structured variables, and pending task metadata) to durable storage and returns control to the caller. After approval, the runtime reloads that state and re-enters the waiting node.

Vega's paused state might look like this:

why-in-memory-blocking-fails.json
1{ 2 "thread_id": "release_reranker_v17", 3 "request_summary": "Promote recommendation-reranker-v17 from shadow to 25% traffic.", 4 "evidence": ["offline eval pass: 0.842 nDCG@10", "canary error rate: 0.21%"], 5 "pending_action": { 6 "tool": "promote_model", 7 "args": {"model_id": "recommendation-reranker-v17", "traffic_percent": 25} 8 }, 9 "approval": { 10 "status": "pending", 11 "policy_rule": "model_release.requires_review", 12 "action_hash": "sha256:53031c1110b357d59740606f7d211b5e876014720b38f5e3ad34582a37d827eb", 13 "version": 3, 14 "expires_at": "2026-08-22T19:00:00Z" 15 } 16}

When an authorized reviewer clicks Approve, the runtime resolves this pending decision with a guarded write, reloads the checkpoint, rechecks current state and arguments, and only then attempts the action. If the reviewer clicks Reject, Vega can draft a release note without promoting the model. Checkpoint state is ordinary inspectable data, but emergency changes should still go through an authenticated, versioned, audited path rather than an ad hoc database edit.

Before building a workflow engine, you can test the persistence boundary with ordinary data. The checkpoint below stores a redacted summary and an action hash, but not the raw request message.

persist-minimized-checkpoint.py
1from hashlib import sha256 2import json 3 4raw_message = "Promote reranker-v17. Ask [email protected] to verify rollout notes." 5pending_action = { 6 "tool": "promote_model", 7 "args": {"model_id": "recommendation-reranker-v17", "traffic_percent": 25}, 8} 9action_bytes = json.dumps(pending_action, sort_keys=True).encode() 10checkpoint = { 11 "request_summary": "Release assistant proposes 25% traffic for reranker-v17.", 12 "pending_action": pending_action, 13 "action_hash": f"sha256:{sha256(action_bytes).hexdigest()}", 14 "status": "pending", 15} 16stored = json.dumps(checkpoint) 17 18print("status:", checkpoint["status"]) 19print("action hash present:", bool(checkpoint["action_hash"])) 20print("stored digest length:", len(checkpoint["action_hash"].removeprefix("sha256:"))) 21print("action hash for display:", checkpoint["action_hash"][:19] + "...") 22print("raw message stored:", raw_message in stored)
Output
1status: pending 2action hash present: True 3stored digest length: 64 4action hash for display: sha256:53031c1110b3... 5raw message stored: False

Compare two popular approaches for implementing this pattern:

FeatureLangGraph (interrupt)Temporal
Best forGraph-based agents with checkpointed interruptsWorkflow orchestration with timers, retries, and cross-service activities
State PersistenceDurable checkpointer (Postgres, Redis, etc.)Built-in durable event history
Resume MechanismAPI calls invoking Command(resume=...)Signals or Updates
Operational shapeAgent runtime plus durable checkpointerWorkflow service plus activity workers and message handlers

Using LangGraph checkpointing

LangGraph provides first-class support for this via interrupt(). [2]Reference 2LangGraph Interruptshttps://docs.langchain.com/oss/python/langgraph/interrupts When you compile the graph with a checkpointer, LangGraph persists checkpoints for the thread and pauses execution until the graph is invoked again with Command(resume=...). [3]Reference 3LangGraph Persistencehttps://docs.langchain.com/oss/python/langgraph/persistence One subtle detail matters in production: when execution resumes, LangGraph reruns the node from the top, so any code before interrupt() must be idempotent. [2]Reference 2LangGraph Interruptshttps://docs.langchain.com/oss/python/langgraph/interrupts

using-langgraph-checkpointing.py
1from langgraph.graph import StateGraph 2from langgraph.checkpoint.postgres import PostgresSaver 3from langgraph.types import interrupt, Command 4from langchain_core.messages import AIMessage 5from typing import TypedDict 6 7Message = dict[str, object] 8ToolAction = dict[str, object] 9 10class AgentState(TypedDict): 11 messages: list[Message] 12 pending_action: ToolAction | None 13 approval_status: str | None 14 15def execute_action_node(state: AgentState) -> dict: 16 """Execute a tool call, pausing for approval if needed.""" 17 action = state["pending_action"] 18 if action is None: 19 return {"messages": [], "pending_action": None, "approval_status": None} 20 21 policy = TOOL_POLICIES.get(action["tool"]) 22 if policy is None: 23 return { 24 "messages": [AIMessage(content="Action blocked: tool has no policy.")], 25 "pending_action": None, 26 "approval_status": "rejected", 27 } 28 execution_args = action["args"] 29 30 # Check if we need to pause 31 if policy and policy.risk_level in (RiskLevel.APPROVE, RiskLevel.ESCALATE): 32 # PAUSE execution here. 33 # The graph state is automatically saved to Postgres by the checkpointer. 34 # The runtime handles the pause; don't catch interrupt() in try/except. 35 human_response = interrupt({ 36 "action": action, 37 "risk_level": policy.risk_level.value, 38 "reason": f"Agent wants to {action['tool']} with args: {action['args']}", 39 "requires_reason": policy.requires_reason, 40 }) 41 42 # This code runs ONLY after the human resumes execution 43 if human_response["decision"] == "reject": 44 return { 45 "messages": [AIMessage(content="Action was rejected by human reviewer.")], 46 "pending_action": None, 47 "approval_status": "rejected", 48 } 49 50 # An approved edit is a new proposal. Validate it before execution. 51 if human_response["decision"] == "modify": 52 execution_args = validate_modified_args( 53 action["tool"], 54 human_response["modified_args"], 55 policy=policy, 56 ) 57 else: 58 execution_args = validate_current_args( 59 action["tool"], 60 action["args"], 61 policy=policy, 62 ) 63 else: 64 execution_args = validate_autonomous_args( 65 action["tool"], 66 action["args"], 67 policy=policy, 68 ) 69 70 # Bind execution to an idempotency key so retries can't repeat a side effect. 71 result = execute_tool( 72 action["tool"], 73 execution_args, 74 idempotency_key=action_idempotency_key(action), 75 ) 76 77 return { 78 "messages": [AIMessage(content=f"Action completed: {result}")], 79 "pending_action": None, 80 "approval_status": "executed", 81 } 82 83# Setup the graph with persistence 84graph = StateGraph(AgentState) 85graph.add_node("plan", plan_node) 86graph.add_node("execute", execute_action_node) 87# ... define edges ... 88 89# The checkpointer provides durable HITL state. 90# The first time you use a Postgres checkpointer, call setup() to create tables. 91with PostgresSaver.from_conn_string("postgresql://...") as checkpointer: 92 checkpointer.setup() 93 app = graph.compile(checkpointer=checkpointer)

The approval flow

The flow separates the Agent Runtime (the Python application, often powered by an LLM) from the Approval Interface (Web/Slack). In production, they coordinate through persisted state plus a thin resume endpoint, not a long-lived in-memory call stack. The checkpoint figure above is the source of truth for the sequence: pause, persist state, create a versioned approval record, notify the reviewer, validate the decision, reload the checkpoint, execute, and audit.

REST API for approval queue

To build the "Resume API" component, we need an endpoint that looks up the suspended thread and issues a command to resume it. This endpoint takes a thread ID plus a versioned approval decision payload as input, validates the thread's state, and outputs a resume command to awaken the agent with the provided instructions.

rest-api-for-approval-queue.py
1from fastapi import FastAPI, HTTPException 2from pydantic import BaseModel 3from typing import Literal 4from datetime import datetime, timezone 5 6app = FastAPI() 7JsonDict = dict[str, object] 8 9# Assume langgraph_app is compiled elsewhere 10# langgraph_app = graph.compile(...) 11 12class ApprovalDecision(BaseModel): 13 approval_id: str 14 expected_version: int 15 action_hash: str 16 decision: Literal["approve", "reject", "modify"] 17 reason: str | None = None 18 modified_args: JsonDict | None = None 19 20@app.post("/api/approvals/{thread_id}/decide") 21async def decide_approval(thread_id: str, decision: ApprovalDecision): 22 """ 23 Human approves, rejects, or modifies the pending action. 24 This wakes up the dormant agent. 25 """ 26 # 1. Verify the thread currently has an outstanding interrupt 27 config = {"configurable": {"thread_id": thread_id}} 28 state = langgraph_app.get_state(config) 29 if not any(task.interrupts for task in state.tasks): 30 raise HTTPException(400, "Agent isn't waiting for input") 31 32 # 2. Resolve the approval and write its outbox row atomically. 33 # The transaction commits both writes or rolls back both writes. 34 with approval_store.transaction() as tx: 35 approval = load_pending_approval(tx, thread_id, decision.approval_id) 36 if approval is None or approval["status"] != "pending": 37 raise HTTPException(404, "Approval request not found") 38 if approval["version"] != decision.expected_version: 39 raise HTTPException(409, "Approval request is stale") 40 if approval["action_hash"] != decision.action_hash: 41 raise HTTPException(409, "Proposed action changed; request fresh review") 42 if approval["expires_at"] <= datetime.now(timezone.utc): 43 raise HTTPException(409, "Approval request expired") 44 require_reviewer_permission(current_reviewer(), approval["required_role"]) 45 46 updated = try_resolve_approval( 47 tx, 48 approval_id=decision.approval_id, 49 expected_version=decision.expected_version, 50 expected_action_hash=decision.action_hash, 51 next_status="rejected" if decision.decision == "reject" else "authorized", 52 ) 53 if not updated: 54 raise HTTPException(409, "Approval was already resolved") 55 56 # 3. Insert, rather than publish, an idempotent resume job in the same transaction. 57 # An outbox worker publishes it after commit and retries safely if delivery fails. 58 insert_resume_outbox( 59 tx, 60 thread_id=thread_id, 61 approval_id=decision.approval_id, 62 resume_payload={ 63 "approval_id": decision.approval_id, 64 "expected_version": decision.expected_version, 65 "action_hash": decision.action_hash, 66 "decision": decision.decision, 67 "reason": decision.reason, 68 "modified_args": decision.modified_args, 69 "timestamp": datetime.now(timezone.utc).isoformat(), 70 }, 71 ) 72 return {"status": "decision_recorded"}

The compare-and-swap check records at most one reviewer decision. Store each approval request with fields such as status, version, action_hash, expires_at, required_role, and resolved_at, then resolve it with a guarded update such as WHERE id = ? AND status = 'pending' AND version = ? AND action_hash = ? AND expires_at > now(). The stored and compared action hash must contain the full digest; a shortened form belongs only in the UI. Record authorized separately from executed: an approval may be accepted and then fail during execution. The guarded approval update and outbox insert must commit in one database transaction. A reconciler that detects authorized approvals without an outbox row is an additional defense, not a substitute for atomic persistence. Otherwise, a crash after the approval update but before the outbox insert can strand an authorized action. A durable resume worker must recheck current production state and use an idempotency key so a retry can't promote the same model twice.

This small executable model shows those two separate transitions. It takes a versioned decision and an idempotency key as input, then shows that a stale click is rejected and an execution retry returns the previously recorded outcome.

resolve-once-execute-once.py
1from dataclasses import dataclass 2from hashlib import sha256 3 4ACTION_HASH = f"sha256:{sha256(b'promote:reranker-v17:25pct').hexdigest()}" 5 6@dataclass 7class Approval: 8 status: str = "pending" 9 version: int = 3 10 action_hash: str = ACTION_HASH 11 12completed_effects: dict[str, str] = {} 13 14def record_decision( 15 approval: Approval, 16 *, 17 expected_version: int, 18 action_hash: str, 19) -> str: 20 if approval.status != "pending" or approval.version != expected_version: 21 return "blocked: stale decision" 22 if approval.action_hash != action_hash: 23 return "blocked: action changed" 24 approval.status = "authorized" 25 approval.version += 1 26 return "authorized" 27 28def execute_once(approval: Approval, *, idempotency_key: str) -> str: 29 if idempotency_key in completed_effects: 30 return completed_effects[idempotency_key] 31 if approval.status != "authorized": 32 return "blocked: missing authorization" 33 result = "promotion recorded once" 34 completed_effects[idempotency_key] = result 35 approval.status = "executed" 36 return result 37 38pending = Approval() 39print("decision:", record_decision( 40 pending, 41 expected_version=3, 42 action_hash=ACTION_HASH, 43)) 44print("first execution:", execute_once(pending, idempotency_key="apr_123")) 45print("retry:", execute_once(pending, idempotency_key="apr_123")) 46print("second click:", record_decision( 47 pending, 48 expected_version=3, 49 action_hash=ACTION_HASH, 50))
Output
1decision: authorized 2first execution: promotion recorded once 3retry: promotion recorded once 4second click: blocked: stale decision

An expiry guard is separate from version matching. This example takes the current time and expiry time as inputs, then rejects a decision whose review window has already passed.

reject-expired-approval.py
1from datetime import datetime, timezone 2 3def decision_allowed(*, expires_at: str, now: datetime) -> str: 4 expiry = datetime.fromisoformat(expires_at.replace("Z", "+00:00")) 5 return "accepted" if now < expiry else "blocked: approval expired" 6 7clock = datetime(2026, 8, 22, 19, 0, tzinfo=timezone.utc) 8print("fresh:", decision_allowed( 9 expires_at="2026-08-22T19:01:00Z", 10 now=clock, 11)) 12print("stale:", decision_allowed( 13 expires_at="2026-08-22T18:59:00Z", 14 now=clock, 15))
Output
1fresh: accepted 2stale: blocked: approval expired

Designing the approval interface

A "Yes/No" button is rarely enough for production systems. The approval interface must provide context and control. When Vega asks "Can I route 25% traffic to reranker-v17?", the human needs to know what data will change and why.

The approval UI should show the authorized reviewer the smallest evidence packet needed for the decision: a redacted request summary, the policy trigger, source references the reviewer is allowed to inspect, and the exact arguments Vega proposes to execute. Dumping an entire incident thread or model-card history into every review card creates unnecessary exposure and makes the meaningful change harder to see.

The approval payload

Vega should pause with a structured payload that the UI can render for review. This payload takes authorized evidence and proposed tool arguments as input and structures them into a redacted typed object for the reviewer. For data mutations, this should ideally be a visual diff rather than raw JSON.

the-approval-payload.ts
1import { createHash } from "node:crypto"; 2 3type JsonValue = null | boolean | number | string | JsonValue[] | { [key: string]: JsonValue }; 4 5function canonicalJson(value: JsonValue): string { 6 if (Array.isArray(value)) { 7 return `[${value.map(canonicalJson).join(",")}]`; 8 } 9 if (value !== null && typeof value === "object") { 10 return `{${Object.keys(value).sort().map( 11 (key) => `${JSON.stringify(key)}:${canonicalJson(value[key])}`, 12 ).join(",")}}`; 13 } 14 return JSON.stringify(value) as string; 15} 16 17function hashAction(action: ApprovalRequest["action"]): string { 18 return `sha256:${createHash("sha256").update(canonicalJson(action)).digest("hex")}`; 19} 20 21type ApprovalRequest = { 22 id: string; 23 agentId: string; 24 requestVersion: number; 25 actionHash: string; 26 expiresAt: string; 27 action: { 28 tool: "promote_model"; 29 reason: string; 30 policyRule: string; 31 args: { 32 model_id: string; 33 from_stage: "shadow" | "canary"; 34 to_stage: "canary" | "production"; 35 traffic_percent: number; 36 }; 37 riskLevel: "HIGH" | "CRITICAL"; 38 }; 39 context: { 40 evidenceSummary: string; 41 }; 42}; 43 44const pendingApprovalAction: ApprovalRequest["action"] = { 45 tool: "promote_model", 46 reason: "Shadow and canary checks passed release threshold", 47 policyRule: "model_release.write_requires_review", 48 args: { 49 model_id: "recommendation-reranker-v17", 50 from_stage: "shadow", 51 to_stage: "canary", 52 traffic_percent: 25, // Render this as a diff in the UI 53 }, 54 riskLevel: "HIGH", 55}; 56 57const pendingApproval: ApprovalRequest = { 58 id: "apr_123", 59 agentId: "agent_42", 60 requestVersion: 4, 61 actionHash: hashAction(pendingApprovalAction), 62 expiresAt: "2026-08-22T19:00:00Z", 63 action: pendingApprovalAction, 64 context: { 65 evidenceSummary: "Eval pass, 30-minute canary clean, rollback plan attached.", 66 }, 67};
Approval packet with frozen evidence, exact mutation diff, reviewer decision, and stale-click guards such as action hash and expiry. Approval packet with frozen evidence, exact mutation diff, reviewer decision, and stale-click guards such as action hash and expiry.
Review packet must freeze intent. Reviewer sees permitted evidence, exact mutation, and guard fields that let backend reject stale clicks.

By rendering this payload, the Human-in-the-Loop UI becomes a useful decision surface. The reviewer can compare Vega's proposed effect with permitted evidence before authorizing the tool call. For complex actions, consider adding a dry-run diff that shows what would happen if the action were approved. Version, expiry, and action hash fields matter too: they let the backend reject stale clicks instead of replaying an approval long after the underlying state changed. The backend stores and compares the full digest even if the card displays an abbreviated label.

The packet builder is also a data-minimization boundary. This executable example takes a deployment note and exact action as input, redacts the email address, and emits a stable hash that binds the review card to the proposed mutation.

build-redacted-review-packet.py
1from hashlib import sha256 2import json 3import re 4 5def build_packet(note: str, action: dict[str, object]) -> dict[str, object]: 6 redacted_note = re.sub(r"[\w.+-]+@[\w.-]+", "[email redacted]", note) 7 canonical_action = json.dumps(action, sort_keys=True, separators=(",", ":")) 8 action_hash = f"sha256:{sha256(canonical_action.encode()).hexdigest()}" 9 return {"evidence": redacted_note, "action": action, "action_hash": action_hash} 10 11pending_approval_action = { 12 "tool": "promote_model", 13 "reason": "Shadow and canary checks passed release threshold", 14 "policyRule": "model_release.write_requires_review", 15 "args": { 16 "model_id": "recommendation-reranker-v17", 17 "from_stage": "shadow", 18 "to_stage": "canary", 19 "traffic_percent": 25, 20 }, 21 "riskLevel": "HIGH", 22} 23packet = build_packet( 24 "Page [email protected] only if canary error budget burns.", 25 pending_approval_action, 26) 27print("evidence:", packet["evidence"]) 28print("has action hash:", bool(packet["action_hash"])) 29print("stored digest length:", len(packet["action_hash"].removeprefix("sha256:"))) 30print("action hash for display:", packet["action_hash"][:19] + "...")
Output
1evidence: Page [email redacted] only if canary error budget burns. 2has action hash: True 3stored digest length: 64 4action hash for display: sha256:04f7578c5e22...

Give the reviewer a concise rationale, the exact tool arguments, a diff, and the policy rule that triggered review. Don't make raw chain-of-thought your approval primitive. Explanations can be unfaithful, and they may reveal internal reasoning or sensitive context you didn't intend to surface. [4]Reference 4Language Models Don't Always Say What They Think: Unfaithful Explanations in Chain-of-Thought Promptinghttps://arxiv.org/abs/2305.04388

Also keep approval state compact. Persist a data-minimized audit record with access controls and retention policy, then inject only structured fields such as decision, reason, modified_args, or a short reviewer summary back into the runtime. Otherwise long-lived threads accumulate reviewer chatter that burns context window on approval metadata instead of task state.

Using Temporal for long-running workflows

For workflows that coordinate several durable activities, timers, retries, and approval messages, Temporal can be a better fit than a hand-rolled checkpoint table. LangGraph checkpoints can also wait durably; the choice isn't only "short wait versus long wait." Temporal's workflow execution records event history and communicates with callers through activities plus message-passing primitives such as Signals and Updates. [5]Reference 5Temporal Workflow Execution Overviewhttps://docs.temporal.io/workflow-execution

Signals are a good default when the approval service can fire-and-forget. If the UI needs synchronous confirmation that the workflow accepted the decision, or you want the runtime to reject a stale approval before it lands in history, an Update is usually cleaner. [6]Reference 6Temporal Python SDK: Workflow message passinghttps://docs.temporal.io/develop/python/workflows/message-passing

Temporal also gives you durable state, timers, retries, and event history out of the box. That means you don't have to build the orchestration layer for "pause, notify, wait, resume" yourself.

This approach is useful for multi-step approval chains (for example, waiting for both a support lead and finance sign-off). The workflow can wait without keeping a worker blocked, evaluate partial approvals, and continue waiting for the remaining required decision.

The Temporal workflow below uses an Update for the approval decision because the UI needs to learn whether the action ID and version still match the pending request. The workflow then calls a validation activity before any external side effect.

using-temporal-for-long-running-workflows.py
1from temporalio import workflow 2 3ApprovalPayload = dict[str, object] 4 5# Assume activities are defined elsewhere 6# plan_actions, notify_human, validate_for_execution, execute_action, is_risky = ... 7 8@workflow.defn 9class AgentWorkflow: 10 def __init__(self) -> None: 11 self._pending_action_id: str | None = None 12 self._pending_version: int | None = None 13 self._human_decision: ApprovalPayload | None = None 14 15 @workflow.update 16 def decide(self, decision: ApprovalPayload) -> str: 17 self._human_decision = decision 18 return "accepted" 19 20 @decide.validator 21 def validate_decision(self, decision: ApprovalPayload) -> None: 22 """Reject stale decisions before the Update is accepted.""" 23 if decision["action_id"] != self._pending_action_id: 24 raise ValueError("stale action id") 25 if decision["expected_version"] != self._pending_version: 26 raise ValueError("stale action version") 27 if self._human_decision is not None: 28 raise ValueError("decision already recorded") 29 30 @workflow.run 31 async def run(self, task: str): 32 # Step 1: Plan 33 plan = await workflow.execute_activity(plan_actions, task, ...) 34 35 for action in plan.actions: 36 if is_risky(action): 37 self._pending_action_id = action["id"] 38 self._pending_version = action["version"] 39 self._human_decision = None 40 41 # Send notification (Slack/Email) 42 await workflow.execute_activity( 43 notify_human, 44 {"action_id": action["id"], "action": action}, 45 ..., 46 ) 47 48 # Wait durably until a validated decision Update arrives. 49 await workflow.wait_condition( 50 lambda: self._human_decision is not None 51 ) 52 53 decision = self._human_decision 54 self._pending_action_id = None 55 self._pending_version = None 56 self._human_decision = None 57 58 if decision["decision"] == "reject": 59 continue 60 61 # The reviewer decision isn't enough: validate any modified 62 # arguments and current business state in an Activity. 63 action = await workflow.execute_activity( 64 validate_for_execution, 65 {"action": action, "decision": decision}, 66 ..., 67 ) 68 69 # Step 2: Execute with an idempotency key carried by the action. 70 await workflow.execute_activity(execute_action, action, ...)

The validator rejects stale versions and a second decision while the first accepted Update is waiting to be consumed, so two fast clicks can't overwrite one another.

Keep policy code inside the workflow deterministic. If is_risky() depends on live balances, anomaly services, or a policy database, fetch that data in an Activity first and pass the result into workflow state. Temporal replays workflow code against event history, so non-deterministic logic inside the workflow will break replay. [5]Reference 5Temporal Workflow Execution Overviewhttps://docs.temporal.io/workflow-execution

If a notification doesn't need a response, a Signal may still be appropriate. Approval buttons need synchronous accept/reject feedback, so an Update with validation is a stronger fit for this decision path. [6]Reference 6Temporal Python SDK: Workflow message passinghttps://docs.temporal.io/develop/python/workflows/message-passing

For approval chains that can run for months or accumulate a large event history, periodically use Continue-As-New to cap history size while carrying forward unresolved approval state. [5]Reference 5Temporal Workflow Execution Overviewhttps://docs.temporal.io/workflow-execution

Advanced patterns

Pause and resume is the core mechanism, but production queues also need dynamic escalation, safe modifications, bounded batches, and reviewer-load metrics. These controls reduce needless review work without weakening policy floors.

Dynamic risk escalation

Static policies need runtime context. Publishing one external incident update is already a side effect that may require approval; attempting hundreds of updates in a short interval should move into a stricter queue or be blocked outright. This visual starts from a static policy floor and conditionally escalates the required authorization layer.

Dynamic risk escalation board where each case combines a static policy floor and runtime-required minimum, then resolves to stricter final tier. Dynamic risk escalation board where each case combines a static policy floor and runtime-required minimum, then resolves to stricter final tier.
Static policy sets floor. Runtime signals can raise friction, never lower it, so final tier is always stricter of two inputs.

The numbered cells match the examples below. A 50% traffic shift moves from APPROVE to ESCALATE; an otherwise autonomous eval read moves to APPROVE after an anomaly spike; and disabling a safety filter stays at ESCALATE even when runtime conditions look quiet. This monotonic rule is the visual form of escalate_risk().

Dynamic policies evaluate context by taking the proposed action and operational metadata as input, applying rule-based checks, and outputting an escalated risk tier if anomalies are detected. This function combines the static base policy with runtime conditions. It checks traffic percentage, recent failure count, and time window, elevating the required authorization level when thresholds are crossed:

dynamic-risk-escalation.py
1from dataclasses import dataclass 2from enum import Enum 3from typing import Literal, cast 4 5class RiskLevel(Enum): 6 AUTO = "auto" 7 NOTIFY = "notify" 8 APPROVE = "approve" 9 ESCALATE = "escalate" 10 11@dataclass 12class ToolPolicy: 13 tool_name: str 14 risk_level: RiskLevel 15 16TOOL_POLICIES = { 17 "read_eval_run": ToolPolicy("read_eval_run", RiskLevel.AUTO), 18 "promote_model": ToolPolicy("promote_model", RiskLevel.APPROVE), 19 "disable_safety_filter": ToolPolicy("disable_safety_filter", RiskLevel.ESCALATE), 20} 21 22RISK_PRIORITY = { 23 RiskLevel.AUTO: 0, 24 RiskLevel.NOTIFY: 1, 25 RiskLevel.APPROVE: 2, 26 RiskLevel.ESCALATE: 3, 27} 28 29def escalate_risk(current: RiskLevel, target: RiskLevel) -> RiskLevel: 30 return target if RISK_PRIORITY[target] > RISK_PRIORITY[current] else current 31 32ToolArgs = dict[str, float | str | bool] 33ToolAction = dict[str, object] 34RuntimeContext = dict[str, object] 35 36def is_business_hours(context: RuntimeContext) -> bool: 37 hour = int(context.get("local_hour", 12)) 38 return 8 <= hour < 18 39 40def calculate_dynamic_risk(action: ToolAction, context: RuntimeContext) -> RiskLevel: 41 """Vega's risk increases with traffic size, recent anomalies, and time of day.""" 42 tool_name = str(action["tool"]) 43 args = cast(ToolArgs, action.get("args", {})) 44 base_risk = TOOL_POLICIES[tool_name].risk_level 45 46 # 1. Blast-radius escalation: large traffic shifts are critical 47 if float(args.get("traffic_percent", 0)) > 25: 48 base_risk = escalate_risk(base_risk, RiskLevel.ESCALATE) 49 50 # 2. Anomaly-based escalation: many recent failures suggest something is wrong 51 if context.get("recent_failures", 0) > 3: 52 base_risk = escalate_risk(base_risk, RiskLevel.APPROVE) 53 54 # 3. Temporal escalation: off-hours promotions need senior review 55 if tool_name == "promote_model" and not is_business_hours(context): 56 base_risk = escalate_risk(base_risk, RiskLevel.ESCALATE) 57 58 return base_risk 59 60print("promote 50%:", calculate_dynamic_risk( 61 {"tool": "promote_model", "args": {"traffic_percent": 50}}, 62 {"recent_failures": 0, "local_hour": 14}, 63).value) 64print("read after failures:", calculate_dynamic_risk( 65 {"tool": "read_eval_run", "args": {}}, 66 {"recent_failures": 4, "local_hour": 14}, 67).value) 68print("disable filter:", calculate_dynamic_risk( 69 {"tool": "disable_safety_filter", "args": {}}, 70 {"recent_failures": 0, "local_hour": 10}, 71).value) 72print("off-hours promote:", calculate_dynamic_risk( 73 {"tool": "promote_model", "args": {"traffic_percent": 5}}, 74 {"recent_failures": 0, "local_hour": 2}, 75).value)
Output
1promote 50%: escalate 2read after failures: approve 3disable filter: escalate 4off-hours promote: escalate

Approval with modification

A useful HITL pattern lets the human modify a proposed action rather than merely reject it. Instead of a binary yes/no choice, the human can correct arguments, such as reducing a traffic percentage or changing a rollout stage. That edit is a new proposal, not a guarantee of safety.

Build the approval UI as a form, not a button. Populate the form with the agent's proposed arguments (e.g., email body, SQL query) and allow the human to edit them before hitting "Approve".

For example, if Vega proposes promote_model(model_id="reranker-v17", traffic_percent=50), a human reviewer can modify it to promote_model(model_id="reranker-v17", traffic_percent=10) for a smaller canary. Before execution, the host must validate schema, reviewer permission, applicable policy, action version, and current deployment state. The edit doesn't train Vega and doesn't bypass a stronger approval tier.

Batch approvals

If Vega needs to propose 50 low-risk release-note publications after a migration freeze, asking for separate decisions for each one creates reviewer fatigue. Reviewers who face repetitive requests may stop inspecting individual effects. A batch review can group related proposals without hiding the service, audience, destination, or policy status of each item.

Instead of creating one alert per proposed operation, the orchestrator collects pending actions over a bounded window or groups them by a shared task identifier. The UI can then present: "Vega proposes 50 release-note publishes (review all items)." The approval record must bind to the exact item list or item hashes, so an item inserted after approval can't ride along in the batch.

Implementing batch approvals requires per-item state and idempotency. If a batch contains 50 actions and the reviewer authorizes the fixed set, the orchestrator must track each action separately. If action 42 fails, already-completed publishes aren't automatically undone unless the operation provides a compensation path. Retry only the failed authorized item with its idempotency key, and re-request review if its inputs or destination changed.

Bind a batch approval to the exact items shown to the reviewer. Here a later publish inserted into the queue changes the digest, so it can't reuse the earlier authorization.

bind-batch-to-reviewed-items.py
1from hashlib import sha256 2import json 3 4def batch_digest(items: list[dict[str, str]]) -> str: 5 payload = json.dumps(items, sort_keys=True, separators=(",", ":")) 6 return sha256(payload.encode()).hexdigest() 7 8reviewed = [ 9 {"id": "note_1", "service": "search"}, 10 {"id": "note_2", "service": "recommendations"}, 11] 12approved_digest = batch_digest(reviewed) 13changed = [*reviewed, {"id": "note_3", "service": "payments"}] 14 15print("reviewed batch matches:", batch_digest(reviewed) == approved_digest) 16print("inserted item matches:", batch_digest(changed) == approved_digest) 17print("stored batch digest length:", len(approved_digest)) 18print("batch digest for display:", approved_digest[:12])
Output
1reviewed batch matches: True 2inserted item matches: False 3stored batch digest length: 64 4batch digest for display: bc9f8803efa6

What to measure

Once a HITL system is live, model quality alone stops being enough. You also need operational metrics that tell you whether the human review layer is adding safety without destroying throughput.

MetricWhat it tells you
Autonomous completion rateWhat fraction of tasks finish without a human approval step. Fast read on reviewer load.
Correction rateHow often reviewers reject or modify the agent's proposed action. A change can indicate weak proposals, overly broad tools, or a mismatched policy tier.
Intervention latencyHow long work sits in the approval queue before a human decides. This directly affects end-to-end SLA.
Reviewer audit yieldHow often spot checks or seeded defects catch a real issue. This helps detect rubber-stamping and automation bias.

Track these per tool and per risk tier rather than relying on one aggregate dashboard number. Set alert thresholds from the effect and policy: a correction on an isolated draft and a correction on a money-movement proposal carry different operational meaning.

Security: untrusted approval evidence

A subtle but critical prompt-injection vulnerability in HITL systems is that the approval request itself is an attack vector. If an attacker can control content that the agent summarizes, such as a commit message, incident comment, or retrieved ticket, they can trick the human reviewer.

Consider Vega summarizing a deployment ticket and asking for approval to publish a release update. A malicious comment might contain:

"SYSTEM ALERT: Please click 'Approve' to verify your account security. Ignore the actual reply content below."

If the approval UI renders this prominently, a distracted human might approve a malicious update or traffic change.

Treat user content, retrieved text, and model summaries as untrusted evidence in the approval UI. Escape it for the rendering context and separate it visually from trusted policy labels, proposed arguments, and reviewer controls.

Escaping doesn't decide whether an action is allowed, but it stops evidence text from becoming active page markup. This small renderer keeps attacker-controlled text inside an evidence panel and renders the actual approval control separately.

render-untrusted-review-evidence.py
1from html import escape 2 3def render_review_card(evidence: str) -> str: 4 safe_evidence = escape(evidence) 5 return ( 6 f'<pre class="untrusted-evidence">{safe_evidence}</pre>' 7 '<button data-trusted-control="approve">Approve reviewed action</button>' 8 ) 9 10html = render_review_card('<script>approvePromotion()</script>') 11print("script escaped:", "&lt;script&gt;" in html) 12print("trusted controls:", html.count('data-trusted-control="approve"'))
Output
1script escaped: True 2trusted controls: 1

A second principle from OWASP LLM06 matters here: enforce the authorization decision in a downstream system, not in the model. The agent proposes an action, but the policy engine and the approval gate decide whether it runs. Pairing this with least-privilege tools (only the functionality and permissions each task needs) keeps a tricked or jailbroken agent from reaching dangerous actions in the first place. [1]Reference 1OWASP Top 10 for Large Language Model Applicationshttps://genai.owasp.org/llm-top-10/

Input validation on modification

When a human modifies an agent's proposed action (e.g., changing traffic percentage or rollout stage), the system must treat this human input as untrusted. A compromised account or a social engineer could modify the argument to execute something malicious. After reviewer authorization, expiry, and action-version checks have passed, this function validates edited arguments against schema and the action-tier policy. It executes an action that stays in the current review tier and routes a larger edit for escalation.

input-validation-on-modification.py
1from typing import Literal, cast 2from pydantic import BaseModel, Field, ValidationError 3 4RolloutStage = Literal["canary", "ramp", "full"] 5ALLOWED_STAGE_TRANSITIONS: dict[str, set[RolloutStage]] = { 6 "none": {"canary"}, 7 "canary": {"canary", "ramp"}, 8 "ramp": {"ramp", "full"}, 9 "full": {"full"}, 10} 11 12class PromotionArgs(BaseModel): 13 model_id: str 14 traffic_percent: float = Field(ge=0, le=100) 15 rollout_stage: RolloutStage 16 17class ToolSpec: 18 def __init__(self, args_schema): 19 self.args_schema = args_schema 20 21ToolRegistry = { 22 "promote_model": ToolSpec(PromotionArgs), 23} 24 25class SafetyGuardrails: 26 def required_tier(self, tool_name: str, validated_args: BaseModel) -> str: 27 if tool_name == "promote_model": 28 promotion = cast(PromotionArgs, validated_args) 29 if promotion.rollout_stage == "full": 30 return "escalate" 31 traffic = promotion.traffic_percent 32 return "approve" if traffic <= 25 else "escalate" 33 return "reject" 34 35safety_guardrails = SafetyGuardrails() 36 37JsonDict = dict[str, object] 38ActionState = dict[str, object] 39Modification = dict[str, JsonDict] 40 41def reject_action(state: ActionState, reason: str) -> JsonDict: 42 return {"status": "rejected", "reason": reason, "state": state} 43 44def execute_tool(tool_name: str, validated_args: BaseModel) -> JsonDict: 45 return { 46 "status": "executed", 47 "tool": tool_name, 48 "args": validated_args.model_dump(), 49 } 50 51def resume_with_modification( 52 state: ActionState, 53 modification: Modification, 54) -> JsonDict: 55 """ 56 Resume agent after human modified the tool arguments. 57 CRITICAL: Re-validate the new arguments against safety policies. 58 """ 59 new_args = modification["new_args"] 60 pending_action = cast(JsonDict, state["pending_action"]) 61 tool_name = cast(str, pending_action["tool"]) 62 63 # 1. Syntax Validation (Pydantic) 64 try: 65 validated_args = ToolRegistry[tool_name].args_schema(**new_args) 66 except ValidationError as e: 67 return reject_action(state, reason=f"Invalid modification: {e}") 68 69 # 2. Check the requested stage against current deployment state. 70 current_stage = cast(str, state["current_rollout_stage"]) 71 requested_stage = cast(PromotionArgs, validated_args).rollout_stage 72 if requested_stage not in ALLOWED_STAGE_TRANSITIONS.get(current_stage, set()): 73 return reject_action( 74 state, 75 reason=f"Invalid rollout transition: {current_stage} -> {requested_stage}", 76 ) 77 78 # 3. The same policy is applied to the edited arguments. 79 required_tier = safety_guardrails.required_tier(tool_name, validated_args) 80 if required_tier == "escalate": 81 return {"status": "needs_escalation", "args": validated_args.model_dump()} 82 if required_tier == "reject": 83 return reject_action(state, reason="Modification violated safety policy") 84 85 # 4. Resume execution with safe arguments 86 return execute_tool(tool_name, validated_args) 87 88state = { 89 "pending_action": {"tool": "promote_model"}, 90 "current_rollout_stage": "canary", 91} 92 93approved = resume_with_modification( 94 state, 95 { 96 "new_args": { 97 "model_id": "reranker-v17", 98 "traffic_percent": 10, 99 "rollout_stage": "canary", 100 } 101 }, 102) 103 104escalated = resume_with_modification( 105 state, 106 { 107 "new_args": { 108 "model_id": "reranker-v17", 109 "traffic_percent": 50, 110 "rollout_stage": "canary", 111 } 112 }, 113) 114 115print("approved:", approved["status"], approved["args"]["traffic_percent"]) 116print("larger edit:", escalated["status"])
Output
1approved: executed 10.0 2larger edit: needs_escalation

Scaling oversight with AI triage

As agent volume grows, human review can become the primary operational bottleneck. A rules engine or reviewer model can help prioritize the queue and reject proposals that violate policy, provided it never turns an approval-required effect into autonomous execution.

This introduces an "AI-in-the-loop" filter that can automatically reject obvious policy violations and fast-path actions already classified as autonomous by explicit policy. The human reviewer is still required for actions at or above the approval floor.

A practical design combines three layers: hard policy rules, anomaly features, and an evaluator that emits a queue score plus rationale. The evaluator shouldn't silently override an approval-marked or critical action. Its job is triage, not final authority. Reviewer decisions may later support evaluation or training, but only after access control, purpose limitation, redaction, label-quality review, and leakage-safe dataset splitting.

Treat the policy tier as a floor. This router takes an explicit policy and a model suggestion as inputs; it lets a draft stay autonomous, but refuses to downgrade a model-promotion proposal or destructive action.

preserve-policy-floor-during-triage.py
1RANK = {"auto": 0, "approve": 1, "escalate": 2} 2POLICY_FLOOR = { 3 "draft_release_note": "auto", 4 "promote_model": "approve", 5 "delete_eval_evidence": "escalate", 6} 7 8def routed_tier(tool: str, evaluator_suggestion: str) -> str: 9 floor = POLICY_FLOOR[tool] 10 if RANK[evaluator_suggestion] < RANK[floor]: 11 return floor 12 return evaluator_suggestion 13 14print("draft:", routed_tier("draft_release_note", "auto")) 15print("promotion suggested auto:", routed_tier("promote_model", "auto")) 16print("delete suggested approve:", routed_tier("delete_eval_evidence", "approve"))
Output
1draft: auto 2promotion suggested auto: approve 3delete suggested approve: escalate

That human approval record is also part of your governance story. NIST AI RMF frames governance, measurement, and operational controls as lifecycle responsibilities across design, deployment, and management. [7]Reference 7Artificial Intelligence Risk Management Framework (AI RMF 1.0)https://www.nist.gov/itl/ai-risk-management-framework If a deployment is classified as a high-risk AI system under the EU AI Act, Article 14 requires effective human oversight proportionate to its risk, including abilities related to understanding limitations, automation bias, interpretation of output, and intervention or stopping the system. [8]Reference 8EU AI Act: Regulation laying down harmonised rules on artificial intelligencehttps://eur-lex.europa.eu/legal-content/EN/TXT/?uri=CELEX:32024R1689 For any consequential workflow, retain authorized, data-minimized evidence of who decided, which version and effect were reviewed, what executed, and why the outcome was recorded.

Practice: design Vega's weekend policy

Suppose Vega runs unattended on Saturday night during a migration freeze. Design a dynamic risk policy with these rules:

  • Automatically perform authorized evaluation reads and deployment-health checks
  • Require approval for every model-promotion proposal
  • Escalate any promotion over 10% traffic to a senior reviewer
  • Escalate any proposal that disables a required safety filter
  • Auto-reject any request after 2:00 AM if the same service already had two failed rollouts in the past hour

Write the ToolPolicy table and the calculate_dynamic_risk function. Then identify the edge case: what happens if Vega proposes three 9% traffic shifts in one hour to bypass the senior-review threshold?

This is a split-transaction attack. Dynamic policies must look at aggregate service behavior, rather than single action size. A useful fix is a rolling-sum check: if one service accumulates more than 10% new traffic within 24 hours, escalate regardless of individual step size.

HITL controls to keep

  • Categorize every tool into Auto, Notify, Approve, or Escalate.

  • Use the Checkpoint/Resume pattern (LangGraph, Temporal) so approvals can be asynchronous and survive restarts.

  • Allow reviewers to edit proposals, then validate the edited action again before execution.

  • Escalate risk based on velocity, time of day, and dollar amounts.

  • Group repetitive actions to prevent alert fatigue.

The point of HITL isn't to keep humans clicking "Approve" forever. It's to put human judgment where failure is expensive or irreversible, while keeping policy-required gates in place even as lower-risk automation improves.

A safe HITL design has four moving pieces: risk classification, durable checkpoints, compare-and-swap resume semantics, and approval UIs that give reviewers real context. The common traps are in-memory blocking, stale approvals, alert fatigue, and prompt injection through the approval message itself.

The next step is applying those approval patterns to AI-assisted software work. Coding agents can create useful patches, but they need the same risk tiers, review checks, data-minimized audit evidence, and human ownership before their changes reach a repository or deployment pipeline.

Mastery check

Key concepts

  • HITL, HOTL, and HOOTL oversight modes
  • Risk tiers: Auto, Notify, Approve, and Escalate
  • Durable checkpoint and resume instead of in-memory waiting
  • Approval packets, compare-and-swap guards, and stale-click protection
  • Dynamic escalation, safe modification, and reviewer throughput metrics

Evaluation rubric

  • Picks the right oversight mode for each tool action instead of blocking everything
  • Explains why durable execution and guarded resume writes are required
  • Designs reviewer UX with context, diffs, expiry, and validation
  • Balances safety and throughput with escalation logic, batching, and metrics

Follow-up questions

Common pitfalls

  • Symptom: Review queue grows and reviewers start blindly approving requests.

  • Cause: Everything is routed through the same approval path, including safe or repetitive work.

  • Fix: Add risk tiers, bounded batches, and triage while keeping pre-execution review for effects that policy requires humans to authorize.

  • Symptom: Approved actions disappear after deploy or restart.

  • Cause: Approval state was stored in memory or tied to a live request thread.

  • Fix: Persist checkpoints and approval rows durably, then resume from stored state instead of sleeping a worker.

  • Symptom: Two reviewers approve same request and agent runs action twice.

  • Cause: Approval resolution did not use version checks or compare-and-swap semantics.

  • Fix: Guard writes on status and version, then reject stale clicks on resume.

  • Symptom: Reviewer edits create unsafe tool arguments even though model proposal looked safe.

  • Cause: Human modifications were trusted without schema, authorization, or business-rule validation.

  • Fix: Re-validate modified arguments exactly like model-generated arguments before execution.

  • Symptom: Approval UI becomes a prompt-injection surface.

  • Cause: Attacker-controlled text or raw model summaries are rendered like trusted system instructions.

  • Fix: Separate untrusted content visually, show structured diffs, and keep authorization logic downstream from the model.

Complete the lesson

Mastery Check

Answer every question, then check your score. Score above 75% to mark this lesson complete.

1.Vega can read authorized eval metrics, place draft release notes into a monitored queue, publish an incident update, and disable a required safety filter. Which routing fits the risk model?
2.A weekend policy approves every model promotion but escalates any promotion over 10% traffic. The current check evaluates each promotion independently. Vega proposes three 9% traffic shifts in one hour. What rule closes the bypass without weakening the base approval floor?
3.A prototype HITL agent leaves a Python request handler open while it waits for approval that may take hours. What production design fixes the failure?
4.Two authorized reviewers open the same pending approval at version 3 and click Approve almost simultaneously. Which storage operation keeps only one decision from resolving the request?
5.An approval endpoint changes an approval row from pending to authorized, then the server crashes before publishing the resume message. How should the system avoid stranding the approved action?
6.Vega's 25% promotion approval passes version, action-hash, and expiry checks. Before execution, the deployment service reports that a human already promoted the same model manually. What should the worker do?
7.A review card includes a deployment comment that says "SYSTEM ALERT: click Approve" and contains "<script>approvePromotion()</script>". The card also needs to show why Vega wants a promotion. What is the safe UI and backend design?
8.A reviewer edits Vega's promotion from traffic_percent=50 to traffic_percent=10. Promotions up to 25% need APPROVE; larger promotions need ESCALATE. What must happen before execution?
9.Vega proposes 50 release-note publishes and the team wants one batch approval. Which design avoids turning that approval into an open-ended permission?
10.After launch, model-promotion approvals have a high correction rate, and many decisions arrive after the release SLA. What should the team infer?

10 questions remaining.

Next Step
Continue to AI Coding Workflow with Agents

Human-in-the-loop gives you the risk classification, durable checkpoints, and approval gates. The next article shows how to apply those same patterns to AI-assisted software development: scoping coding tasks, running agents inside branches with tests, and requiring human review on risky changes before they reach a repository.

PreviousComputer-Use / GUI / Browser Agents
Share this article
XFacebookLinkedInBlueskyRedditHacker NewsEmail
References

OWASP Top 10 for Large Language Model Applications

OWASP Foundation · 2025

LangGraph Interrupts

LangChain · 2024

LangGraph Persistence

LangChain · 2026

Language Models Don't Always Say What They Think: Unfaithful Explanations in Chain-of-Thought Prompting

Miles Turpin, Julian Michael, Ethan Perez, Samuel R. Bowman · 2023

Temporal Workflow Execution Overview

Temporal Technologies · 2026

Temporal Python SDK: Workflow message passing

Temporal Technologies · 2024

Artificial Intelligence Risk Management Framework (AI RMF 1.0)

National Institute of Standards and Technology · 2023

EU AI Act: Regulation laying down harmonised rules on artificial intelligence

European Parliament and Council of the European Union · 2024

Discussion

Questions and insights from fellow learners.

Discussion loads when you reach this section.