Build approval gates, durable checkpoints, and guarded resumes for agent actions that change external state.
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] 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.
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.
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.
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.
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.
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 Level | Description | Examples | Interaction Model |
|---|---|---|---|
| Low | Bounded, authorized read or isolated draft | Read permitted eval metrics, Inspect canary health, Draft release note without publishing | Auto-Execute: Run within least-privilege access. |
| Medium | Internal staged change with a recoverable effect | Queue review packet, Add a permitted internal risk label | Policy Decision: Audit and notify, or require review if policy says so. |
| High | External or production side effect | Publish incident update, Promote model to canary, Shift endpoint traffic | Approve: Pause and wait for an authorized reviewer. |
| Critical | Destructive, unusually high-impact, or legally constrained | Delete retained eval evidence, Promote despite failing gate, Disable required safety filter | Escalate or Block: Require stronger authorization or deny. |
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?").
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.
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)1read_eval_run: auto
2promotion traffic threshold: 25.0
3critical wins: escalateRisk 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.
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.
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:
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.
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)1status: pending
2action hash present: True
3stored digest length: 64
4action hash for display: sha256:53031c1110b3...
5raw message stored: FalseCompare two popular approaches for implementing this pattern:
| Feature | LangGraph (interrupt) | Temporal |
|---|---|---|
| Best for | Graph-based agents with checkpointed interrupts | Workflow orchestration with timers, retries, and cross-service activities |
| State Persistence | Durable checkpointer (Postgres, Redis, etc.) | Built-in durable event history |
| Resume Mechanism | API calls invoking Command(resume=...) | Signals or Updates |
| Operational shape | Agent runtime plus durable checkpointer | Workflow service plus activity workers and message handlers |
LangGraph provides first-class support for this via interrupt(). [2] 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] 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]
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 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.
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.
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.
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))1decision: authorized
2first execution: promotion recorded once
3retry: promotion recorded once
4second click: blocked: stale decisionAn 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.
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))1fresh: accepted
2stale: blocked: approval expiredA "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.
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.
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};
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.
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] + "...")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]
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.
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]
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]
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.
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]
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]
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]
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.
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.
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:
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)1promote 50%: escalate
2read after failures: approve
3disable filter: escalate
4off-hours promote: escalateA 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.
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.
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])1reviewed batch matches: True
2inserted item matches: False
3stored batch digest length: 64
4batch digest for display: bc9f8803efa6Once 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.
| Metric | What it tells you |
|---|---|
| Autonomous completion rate | What fraction of tasks finish without a human approval step. Fast read on reviewer load. |
| Correction rate | How 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 latency | How long work sits in the approval queue before a human decides. This directly affects end-to-end SLA. |
| Reviewer audit yield | How 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.
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.
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:", "<script>" in html)
12print("trusted controls:", html.count('data-trusted-control="approve"'))1script escaped: True
2trusted controls: 1A 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]
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.
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"])1approved: executed 10.0
2larger edit: needs_escalationAs 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.
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"))1draft: auto
2promotion suggested auto: approve
3delete suggested approve: escalateThat 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] 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] 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.
Suppose Vega runs unattended on Saturday night during a migration freeze. Design a dynamic risk policy with these rules:
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.
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.
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.
Answer every question, then check your score. Score above 75% to mark this lesson complete.
10 questions remaining.
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
Questions and insights from fellow learners.