Human-in-the-Loop for AI Agents: Approval Gates, Escalation, Audit Trails
Where an agent needs human sign-off, four patterns (gates, escalation, collaborative edits, sampling), pause and resume with LangGraph, timeouts, audit logs and pitfalls.
This article is published in English.
Sources: Agentic Architectures — Article 13: Human-in-the-Loop Patterns
Every agent system eventually forces one decision, and most teams reach it by accident: what may the agent do without asking a person first?
On one early production agent, the unspoken answer was "anything the tools permit." It read documents, queried databases, opened tickets and sent notifications, all on its own. That held up until a loop nobody had anticipated made it create forty-three duplicate tickets in roughly ninety seconds. Nobody was positioned to stop it, and by the time a human noticed, cleanup ate most of an afternoon.
After an incident like that, human-in-the-loop stops being a fallback for when the agent breaks and becomes a first-class architectural concern. Some decisions deserve human approval not because the agent is flaky but because a mistake costs enough that a person should own the call.
The rest covers how to place those checkpoints deliberately, how to implement them without wrecking the agent's usefulness, and how to build the approval plumbing on AWS: choosing which actions need approval, the four main patterns (approval gates, escalation, collaborative editing and oversight sampling), pausing and resuming execution, the approval interface and audit trail, and timeout handling.
Where to draw the line
Autonomy and control pull against each other. A fully autonomous agent is fast and scalable but can do damage at machine speed. A fully supervised one is safe yet so slow that automation loses its point. Two dimensions locate the right line: how reversible the action is and how expensive an error would be.
+--------------------+------------------------+------------------------+
| | Low Error Cost | High Error Cost |
+--------------------+------------------------+------------------------+
| Reversible | AUTONOMOUS | OVERSIGHT SAMPLING |
| | (agent acts freely) | (act, log, review some)|
+--------------------+------------------------+------------------------+
| Irreversible | APPROVAL FOR NOVEL | APPROVAL GATE |
| | (approve first time, | (always require human |
| | then autonomous) | approval before act) |
+--------------------+------------------------+------------------------+
Reading a document is reversible and cheap, so it runs autonomously. Drafting an email is reversible too, so autonomy works, perhaps with sampling. Emailing a customer can't be undone and may cost a lot, which calls for an approval gate. Deleting production data is irreversible and potentially catastrophic, which calls for a gate with a second approver.
A frequent error is applying one policy to the entire agent. The policy belongs at the level of the tool or even the single action. One agent can read freely, draft under sampling and send behind a gate.
# harness/hitl/action_policy.py
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable
class ApprovalPolicy(Enum):
AUTONOMOUS = "autonomous" # act freely
OVERSIGHT_SAMPLING = "sampling" # act, log, review a sample
APPROVAL_FOR_NOVEL = "approval_novel" # approve first occurrence, then auto
APPROVAL_GATE = "approval_gate" # always require approval
DUAL_APPROVAL = "dual_approval" # require two approvers
@dataclass
class ActionPolicy:
action_name: str
policy: ApprovalPolicy
reason: str
approver_role: Optional[str] = None # required role to approve
timeout_seconds: int = 3600
timeout_action: str = "reject" # "reject" | "escalate" | "proceed"
# Example policy configuration for a customer support agent
SUPPORT_AGENT_POLICIES = {
"read_customer_record": ActionPolicy(
action_name="read_customer_record",
policy=ApprovalPolicy.AUTONOMOUS,
reason="Read-only, reversible, low risk",
),
"draft_response": ActionPolicy(
action_name="draft_response",
policy=ApprovalPolicy.OVERSIGHT_SAMPLING,
reason="Reversible but customer-facing quality matters",
),
"send_customer_email": ActionPolicy(
action_name="send_customer_email",
policy=ApprovalPolicy.APPROVAL_GATE,
reason="Irreversible, customer-facing, reputation risk",
approver_role="support_agent",
timeout_seconds=1800,
timeout_action="reject",
),
"issue_refund": ActionPolicy(
action_name="issue_refund",
policy=ApprovalPolicy.DUAL_APPROVAL,
reason="Financial impact, irreversible",
approver_role="support_manager",
timeout_seconds=7200,
timeout_action="escalate",
),
"delete_account": ActionPolicy(
action_name="delete_account",
policy=ApprovalPolicy.DUAL_APPROVAL,
reason="Catastrophic and irreversible",
approver_role="senior_manager",
timeout_seconds=86400,
timeout_action="reject",
),
}
class ActionPolicyEngine:
"""
Determines whether an action requires human approval and how.
Consulted at the tool execution layer before any action runs.
"""
def __init__(self, policies: dict):
self.policies = policies
def get_policy(self, action_name: str) -> ActionPolicy:
return self.policies.get(
action_name,
# Default to approval gate for unknown actions: fail safe
ActionPolicy(
action_name=action_name,
policy=ApprovalPolicy.APPROVAL_GATE,
reason="Unknown action, defaulting to approval required",
)
)
def requires_approval(self, action_name: str, is_novel: bool = False) -> bool:
policy = self.get_policy(action_name)
if policy.policy == ApprovalPolicy.AUTONOMOUS:
return False
if policy.policy == ApprovalPolicy.OVERSIGHT_SAMPLING:
return False # acts first, reviewed after
if policy.policy == ApprovalPolicy.APPROVAL_FOR_NOVEL:
return is_novel
return True # APPROVAL_GATE and DUAL_APPROVAL always require approval
Pattern 1: approval gates
An approval gate halts the agent before an action and waits for a person to accept or reject it. It is the most common pattern and the most important. The catch is that an agent can't keep a process alive while waiting, since a reviewer may need minutes or hours.
LangGraph handles this through interrupts: execution stops at a defined point, state is persisted, and the run resumes when the decision arrives. The LangGraph documentation on human-in-the-loop with interrupts covers the mechanics at langchain-ai.github.io/langgraph/concepts/human_in_the_loop/.
# harness/hitl/approval_gate.py
import boto3
import time
import uuid
import json
from typing import Optional, Literal
from dataclasses import dataclass
from langgraph.types import interrupt, Command
@dataclass
class ApprovalRequest:
request_id: str
run_id: str
action_name: str
action_args: dict
context_summary: str
requested_at: float
approver_role: str
status: str # PENDING | APPROVED | REJECTED | TIMEOUT
decided_by: Optional[str] = None
decided_at: Optional[float] = None
decision_note: Optional[str] = None
class ApprovalGateManager:
"""
Manages approval requests using LangGraph interrupts for pause/resume
and DynamoDB for durable request state.
"""
def __init__(
self,
table_name: str = "agent-approval-requests",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
self.sns = boto3.client("sns", region_name=region)
def create_approval_request(
self,
run_id: str,
action_name: str,
action_args: dict,
context_summary: str,
approver_role: str,
notification_topic_arn: Optional[str] = None,
) -> ApprovalRequest:
"""
Creates a pending approval request and notifies approvers.
"""
request = ApprovalRequest(
request_id=str(uuid.uuid4()),
run_id=run_id,
action_name=action_name,
action_args=action_args,
context_summary=context_summary,
requested_at=time.time(),
approver_role=approver_role,
status="PENDING",
)
self.table.put_item(Item={
"request_id": request.request_id,
"run_id": request.run_id,
"action_name": request.action_name,
"action_args": json.dumps(request.action_args),
"context_summary": request.context_summary,
"requested_at": int(request.requested_at),
"approver_role": request.approver_role,
"status": "PENDING",
})
# Notify approvers
if notification_topic_arn:
self.sns.publish(
TopicArn=notification_topic_arn,
Subject=f"Approval needed: {action_name}",
Message=json.dumps({
"request_id": request.request_id,
"action": action_name,
"context": context_summary,
"approver_role": approver_role,
}),
)
return request
def record_decision(
self,
request_id: str,
decision: Literal["APPROVED", "REJECTED"],
decided_by: str,
decision_note: Optional[str] = None,
) -> bool:
"""
Records a human's approval decision.
Called by the approval interface when a human responds.
"""
self.table.update_item(
Key={"request_id": request_id},
UpdateExpression=(
"SET #s = :status, decided_by = :by, "
"decided_at = :at, decision_note = :note"
),
ExpressionAttributeNames={"#s": "status"},
ExpressionAttributeValues={
":status": decision,
":by": decided_by,
":at": int(time.time()),
":note": decision_note or "",
},
# Only allow decision on PENDING requests: prevents double-decision
ConditionExpression="#s = :pending",
ExpressionAttributeValues2={":pending": "PENDING"} if False else None,
)
return True
def get_decision(self, request_id: str) -> Optional[str]:
"""Polls the current decision status for a request."""
response = self.table.get_item(Key={"request_id": request_id})
item = response.get("Item")
return item.get("status") if item else None
# The LangGraph node that implements the approval gate
def approval_gate_node(state: dict) -> dict:
"""
A LangGraph node that pauses execution and waits for human approval.
Uses interrupt() to suspend the graph. The graph state is persisted
and can be resumed when the approval decision arrives.
"""
pending_action = state.get("pending_action")
if not pending_action:
return state
# interrupt() pauses the graph and surfaces this data to the caller.
# The caller (your application) presents it to a human and resumes
# the graph with the decision.
decision = interrupt({
"type": "approval_required",
"action": pending_action["name"],
"args": pending_action["args"],
"context": state.get("context_summary", ""),
"run_id": state.get("agent_run_id"),
})
# When resumed, decision contains the human's response
if decision.get("approved"):
return {
**state,
"action_approved": True,
"approved_by": decision.get("approver"),
}
else:
return {
**state,
"action_approved": False,
"rejection_reason": decision.get("reason", "Rejected by human reviewer"),
}
Pattern 2: escalation
Escalation differs from a gate. A gate is planned, because it is known beforehand that the action needs sign-off. Escalation is reactive: the agent runs into something it can't or shouldn't handle and passes it to a person.
That requires the agent to notice when it is out of its depth, which is harder than it sounds because agents lean toward confident action over admitting uncertainty.
# harness/hitl/escalation.py
import boto3
import time
import json
from typing import Optional
from dataclasses import dataclass
from langchain_aws import ChatBedrock
from langchain_core.messages import SystemMessage, HumanMessage
ESCALATION_TRIGGERS = {
"low_confidence": "Agent confidence in its solution is below threshold",
"conflicting_information": "Retrieved information contradicts itself",
"policy_ambiguity": "The correct action is genuinely ambiguous under policy",
"high_stakes_uncertainty": "High-stakes decision with insufficient certainty",
"repeated_failure": "Agent has failed the same task multiple times",
"explicit_user_request": "User asked to speak with a human",
}
@dataclass
class EscalationEvent:
escalation_id: str
run_id: str
trigger: str
agent_context: str
agent_attempted_solution: Optional[str]
confidence: float
escalated_at: float
assigned_to: Optional[str] = None
resolution: Optional[str] = None
class EscalationManager:
"""
Handles cases where the agent should hand off to a human.
Distinct from approval gates: escalation is triggered by the agent
recognizing its own limitations.
"""
def __init__(
self,
table_name: str = "agent-escalations",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
bedrock = boto3.client("bedrock-runtime", region_name=region)
self.confidence_model = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-haiku-4-5",
model_kwargs={"temperature": 0, "max_tokens": 256},
)
def should_escalate(
self,
task: str,
proposed_solution: str,
attempts: int,
) -> tuple:
"""
Assesses whether the agent should escalate to a human.
Returns (should_escalate, trigger, confidence).
"""
# Repeated failure is a deterministic trigger
if attempts >= 3:
return True, "repeated_failure", 0.0
# Ask the model to self-assess confidence
response = self.confidence_model.invoke([
SystemMessage(content="""
Assess your confidence in a proposed solution. Be honest about uncertainty.
Consider: Is the information sufficient? Is the answer unambiguous?
Are there conflicting considerations? Is this high-stakes?
Return JSON:
{
"confidence": 0.0 to 1.0,
"should_escalate": true | false,
"trigger": "low_confidence | conflicting_information | policy_ambiguity | high_stakes_uncertainty | none"
}
"""),
HumanMessage(content=f"Task: {task}\n\nProposed solution: {proposed_solution}")
])
try:
assessment = json.loads(response.content)
return (
assessment.get("should_escalate", False),
assessment.get("trigger", "low_confidence"),
assessment.get("confidence", 0.5),
)
except json.JSONDecodeError:
# If we cannot assess confidence, escalate to be safe
return True, "low_confidence", 0.0
def create_escalation(
self,
run_id: str,
trigger: str,
agent_context: str,
attempted_solution: Optional[str],
confidence: float,
notification_topic_arn: Optional[str] = None,
) -> EscalationEvent:
import uuid
event = EscalationEvent(
escalation_id=str(uuid.uuid4()),
run_id=run_id,
trigger=trigger,
agent_context=agent_context,
agent_attempted_solution=attempted_solution,
confidence=confidence,
escalated_at=time.time(),
)
self.table.put_item(Item={
"escalation_id": event.escalation_id,
"run_id": event.run_id,
"trigger": event.trigger,
"agent_context": event.agent_context[:2000],
"attempted_solution": (attempted_solution or "")[:2000],
"confidence": str(confidence),
"escalated_at": int(event.escalated_at),
"status": "OPEN",
})
if notification_topic_arn:
boto3.client("sns").publish(
TopicArn=notification_topic_arn,
Subject=f"Agent escalation: {trigger}",
Message=event.agent_context[:1000],
)
return event
Pattern 3: collaborative editing
Occasionally the right model is neither approve nor reject but edit. The agent writes a draft, a person polishes it, and the polished version goes forward. This fits content generation, document drafting and code that a human will own once the agent has written it.
The design principle is that the human's edits should flow back as a learning signal. If reviewers keep changing the output in the same way, the agent ought to learn from that.
# harness/hitl/collaborative.py
import boto3
import time
import json
from typing import Optional
from dataclasses import dataclass
@dataclass
class CollaborativeEdit:
edit_id: str
run_id: str
original_draft: str
human_edited: str
edit_distance: float # how much was changed
edit_categories: list # "tone", "factual", "structure", "detail"
edited_by: str
edited_at: float
class CollaborativeEditManager:
"""
Manages the draft-edit-finalize flow where a human refines agent output.
Captures edits as learning signals for future improvement.
"""
def __init__(
self,
table_name: str = "agent-collaborative-edits",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def record_edit(
self,
run_id: str,
original_draft: str,
human_edited: str,
edited_by: str,
edit_categories: list = None,
) -> CollaborativeEdit:
"""
Records a human edit to agent output.
The edit distance and categories become learning signals.
"""
import uuid
edit_distance = self._compute_edit_distance(original_draft, human_edited)
edit = CollaborativeEdit(
edit_id=str(uuid.uuid4()),
run_id=run_id,
original_draft=original_draft,
human_edited=human_edited,
edit_distance=edit_distance,
edit_categories=edit_categories or [],
edited_by=edited_by,
edited_at=time.time(),
)
self.table.put_item(Item={
"edit_id": edit.edit_id,
"run_id": edit.run_id,
"original_draft": original_draft[:5000],
"human_edited": human_edited[:5000],
"edit_distance": str(edit_distance),
"edit_categories": edit.edit_categories,
"edited_by": edited_by,
"edited_at": int(edit.edited_at),
})
return edit
def analyze_edit_patterns(
self,
run_ids: list = None,
min_edits: int = 20,
) -> dict:
"""
Analyzes accumulated edits to find systematic patterns.
If humans consistently edit for the same reason, the agent's
system prompt or procedure should be updated.
"""
response = self.table.scan(Limit=200)
edits = response.get("Items", [])
if len(edits) < min_edits:
return {"sufficient_data": False, "edit_count": len(edits)}
# Aggregate edit categories
category_counts = {}
total_distance = 0.0
for edit in edits:
for cat in edit.get("edit_categories", []):
category_counts[cat] = category_counts.get(cat, 0) + 1
total_distance += float(edit.get("edit_distance", 0))
avg_distance = total_distance / len(edits)
dominant_category = max(category_counts, key=category_counts.get) if category_counts else None
return {
"sufficient_data": True,
"edit_count": len(edits),
"average_edit_distance": round(avg_distance, 3),
"category_distribution": category_counts,
"dominant_edit_category": dominant_category,
"recommendation": self._recommendation(dominant_category, avg_distance),
}
def _recommendation(self, dominant_category: Optional[str], avg_distance: float) -> str:
if avg_distance < 0.1:
return "Edits are minor. Agent output quality is high."
if dominant_category == "tone":
return "Humans frequently adjust tone. Update system prompt with tone guidance."
if dominant_category == "factual":
return "Humans frequently correct facts. Review agent's information sources."
if dominant_category == "structure":
return "Humans frequently restructure. Add output format guidance to system prompt."
return "Review edit patterns manually for systematic improvements."
def _compute_edit_distance(self, a: str, b: str) -> float:
"""Normalized Levenshtein distance, 0 (identical) to 1 (completely different)."""
if not a and not b:
return 0.0
# Simple word-level distance for efficiency on long text
a_words, b_words = a.split(), b.split()
max_len = max(len(a_words), len(b_words))
if max_len == 0:
return 0.0
# Count matching words in order (simplified)
common = len(set(a_words) & set(b_words))
return 1.0 - (common / max_len)
Pattern 4: oversight sampling
For reversible actions with a high error cost, demanding approval each time is too slow. Sampling is the compromise. The agent acts on its own while a sample of its actions is sent to people for review afterward. Review doesn't block anything, yet it catches systematic problems and feeds the evaluation pipeline.
# harness/hitl/oversight_sampling.py
import boto3
import random
import time
import json
from typing import Optional
class OversightSampler:
"""
Routes a sample of autonomous actions to human review after execution.
Does not block the action. Catches quality drift and systematic errors.
Feeds into the Article 8 evaluation pipeline.
"""
def __init__(
self,
base_sample_rate: float = 0.05,
table_name: str = "agent-oversight-samples",
region: str = "us-east-1",
):
self.base_sample_rate = base_sample_rate
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def should_sample(
self,
action_name: str,
agent_confidence: float = 1.0,
is_novel: bool = False,
) -> bool:
"""
Decides whether this action should be sampled for review.
Samples more aggressively for low-confidence and novel actions.
"""
rate = self.base_sample_rate
# Increase sampling for low confidence
if agent_confidence < 0.7:
rate = min(rate * 3, 1.0)
# Always sample novel actions
if is_novel:
return True
return random.random() < rate
def record_for_review(
self,
run_id: str,
action_name: str,
action_args: dict,
action_result: str,
agent_confidence: float,
):
"""Queues an executed action for human review."""
import uuid
self.table.put_item(Item={
"sample_id": str(uuid.uuid4()),
"run_id": run_id,
"action_name": action_name,
"action_args": json.dumps(action_args)[:2000],
"action_result": action_result[:2000],
"agent_confidence": str(agent_confidence),
"sampled_at": int(time.time()),
"review_status": "PENDING",
})
def record_review_outcome(
self,
sample_id: str,
was_correct: bool,
reviewer: str,
notes: Optional[str] = None,
):
"""
Records the human's assessment of a sampled action.
A pattern of incorrect actions should trigger a policy review.
"""
self.table.update_item(
Key={"sample_id": sample_id},
UpdateExpression=(
"SET review_status = :s, was_correct = :c, "
"reviewed_by = :r, review_notes = :n, reviewed_at = :at"
),
ExpressionAttributeValues={
":s": "REVIEWED",
":c": was_correct,
":r": reviewer,
":n": notes or "",
":at": int(time.time()),
}
)
def get_error_rate(self, action_name: str, days: int = 7) -> dict:
"""
Computes the error rate for a sampled action over a window.
If this exceeds threshold, the action's policy should be tightened.
"""
cutoff = int(time.time()) - (days * 86400)
response = self.table.scan(
FilterExpression="action_name = :a AND sampled_at > :c AND review_status = :s",
ExpressionAttributeValues={
":a": action_name,
":c": cutoff,
":s": "REVIEWED",
}
)
reviews = response.get("Items", [])
if not reviews:
return {"sufficient_data": False}
incorrect = sum(1 for r in reviews if not r.get("was_correct", True))
error_rate = incorrect / len(reviews)
return {
"sufficient_data": len(reviews) >= 10,
"sample_size": len(reviews),
"error_rate": round(error_rate, 3),
"recommendation": (
"TIGHTEN_POLICY" if error_rate > 0.1 else "MAINTAIN"
),
}
Timeouts and fallbacks
Every gate needs a timeout policy, because the question of what happens when nobody answers has no safe default. Proceeding automatically defeats the gate. Rejecting automatically can stall legitimate work indefinitely. The proper response depends on how risky the action is, which is why it belongs in the ActionPolicy defined earlier.
# harness/hitl/timeout_handler.py
import boto3
import time
from typing import Literal
class ApprovalTimeoutHandler:
"""
Handles approval requests that exceed their timeout.
The timeout action is policy-defined per action type.
Runs as a scheduled Lambda, checking for expired pending requests.
"""
def __init__(
self,
approval_table: str = "agent-approval-requests",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(approval_table)
self.sns = boto3.client("sns", region_name=region)
def process_expired_requests(self, policies: dict) -> dict:
"""
Scans for pending requests past their timeout and applies
the policy-defined timeout action.
"""
now = int(time.time())
response = self.table.scan(
FilterExpression="#s = :pending",
ExpressionAttributeNames={"#s": "status"},
ExpressionAttributeValues={":pending": "PENDING"},
)
results = {"rejected": 0, "escalated": 0, "proceeded": 0}
for item in response.get("Items", []):
requested_at = int(item.get("requested_at", now))
action_name = item.get("action_name", "")
policy = policies.get(action_name)
if not policy:
continue
age = now - requested_at
if age < policy.timeout_seconds:
continue # not expired yet
timeout_action = policy.timeout_action
if timeout_action == "reject":
self._apply_timeout(item["request_id"], "REJECTED",
"Timed out without approval")
results["rejected"] += 1
elif timeout_action == "escalate":
self._escalate_expired(item)
results["escalated"] += 1
elif timeout_action == "proceed":
# Only for low-risk actions where the gate is advisory
self._apply_timeout(item["request_id"], "APPROVED",
"Auto-approved after timeout per policy")
results["proceeded"] += 1
return results
def _apply_timeout(self, request_id: str, status: str, note: str):
self.table.update_item(
Key={"request_id": request_id},
UpdateExpression="SET #s = :status, decision_note = :note, decided_at = :at",
ExpressionAttributeNames={"#s": "status"},
ExpressionAttributeValues={
":status": status,
":note": note,
":at": int(time.time()),
}
)
def _escalate_expired(self, item: dict):
# Notify a higher tier and extend the deadline
self.table.update_item(
Key={"request_id": item["request_id"]},
UpdateExpression="SET escalated = :t, escalated_at = :at",
ExpressionAttributeValues={":t": True, ":at": int(time.time())},
)
Audit trail
Each human decision needs a tamper-proof record: who approved what, when, with what context and with what outcome. That serves compliance and debugging, and it supports the accountability that makes people comfortable sitting in the loop at all.
# harness/hitl/audit.py
import boto3
import time
import json
import hashlib
from typing import Optional
class HITLAuditLog:
"""
Immutable audit trail for all human-in-the-loop decisions.
Uses a hash chain so tampering is detectable.
Writes to DynamoDB with a separate append-only access pattern.
"""
def __init__(
self,
table_name: str = "agent-hitl-audit",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def record_decision(
self,
run_id: str,
decision_type: str, # "approval" | "rejection" | "escalation" | "edit"
action_name: str,
decided_by: str,
decision_details: dict,
previous_hash: Optional[str] = None,
) -> str:
"""
Records a decision in the audit log with a hash chain.
Returns the hash of this entry for chaining the next one.
"""
timestamp = int(time.time())
entry = {
"run_id": run_id,
"decision_type": decision_type,
"action_name": action_name,
"decided_by": decided_by,
"decision_details": json.dumps(decision_details),
"timestamp": timestamp,
"previous_hash": previous_hash or "genesis",
}
# Compute hash of this entry chained to the previous
entry_content = json.dumps(entry, sort_keys=True)
entry_hash = hashlib.sha256(entry_content.encode()).hexdigest()
self.table.put_item(Item={
"audit_id": f"{run_id}#{timestamp}#{entry_hash[:8]}",
**entry,
"entry_hash": entry_hash,
})
return entry_hash
def verify_chain(self, run_id: str) -> bool:
"""
Verifies the hash chain for a run's audit entries.
Returns True if the chain is intact, False if tampering is detected.
"""
response = self.table.query(
KeyConditionExpression="run_id = :rid",
ExpressionAttributeValues={":rid": run_id},
ScanIndexForward=True,
)
entries = sorted(response.get("Items", []), key=lambda x: x["timestamp"])
previous_hash = "genesis"
for entry in entries:
if entry.get("previous_hash") != previous_hash:
return False
# Recompute and verify
check_entry = {
k: entry[k] for k in
["run_id", "decision_type", "action_name", "decided_by",
"decision_details", "timestamp", "previous_hash"]
}
recomputed = hashlib.sha256(
json.dumps(check_entry, sort_keys=True).encode()
).hexdigest()
if recomputed != entry.get("entry_hash"):
return False
previous_hash = entry["entry_hash"]
return True
Reality check from production
Human-in-the-loop is where the ambition of full automation meets accountability. Several lessons come out of practice.
Approval fatigue is dangerous. Send people too many decisions and they stop reading and start stamping. A gate that always says yes is worse than none, since it gives the look of oversight without the substance. Track approval rates. A gate approved 99% of the time is either unnecessary or no longer being reviewed.
Context decides quality. Someone asked to approve "send_email" with nothing else will decide worse than someone who sees the message text, the customer's history and the agent's reason for choosing it. Oversight quality is capped by the information provided, so the approval interface deserves real investment.
Timeouts need owners. A request that times out and is auto-rejected can silently block important work. Someone must own the queue of pending and expired requests. Build a dashboard, alert on aging requests and define a clear escalation path for when the main approver is away.
Edits are the best training signal and usually go unused. Each time a person edits agent output, they show exactly where the agent falls short. An analyze_edit_patterns function can turn those edits into improvements. If reviewers keep fixing the same thing, fix it in the system prompt. That closes the loop between oversight and improvement.
Regulated industries require the audit trail. In finance, healthcare or law, a hash-chained audit log is mandatory rather than pleasant to have. Regulators will ask who approved a given action, and the answer has to come with confidence that the record wasn't altered. It should be built in from the start.
Latency changes completely. An agent that finishes in four seconds on its own may need four hours once a gate is in the path, because those hours are human response time. That suits asynchronous workflows and ruins real-time ones, so product design should follow human latency, not agent latency.
Reference architecture
The flow runs from the agent reaching an action, through the policy check, into either direct execution, an approval gate, an escalation or a sampled path, and back out through timeout handling and the audit log.
Agent reaches an action
|
v
+---------------------------+
| ActionPolicyEngine |
| Look up action policy |
+---------------------------+
|
+-----+-----+-----------+-----------+
| | | | |
v v v v v
AUTONO SAMP APPROVAL APPROVAL DUAL
MOUS LING FOR NOVEL GATE APPROVAL
| | | | |
| | v v v
| | +--------------------------------+
| | | ApprovalGateManager |
| | | - create request |
| | | - LangGraph interrupt() |
| | | - notify approvers (SNS) |
| | | - persist state (DynamoDB) |
| | +--------------------------------+
| | |
| | Human decides via interface
| | |
| | +-----+-----+
| | | |
| | APPROVED REJECTED / TIMEOUT
| | | |
v v v v
+--------------------------------+
| Execute or Abort Action |
+--------------------------------+
|
v
+--------------------------------+
| HITLAuditLog (hash chain) |
| Immutable decision record |
+--------------------------------+
|
v
+--------------------------------+
| Feed to Article 8 eval + |
| collaborative edit learning |
+--------------------------------+
Reference infrastructure stack
The components, the technology behind each and the role it plays are laid out in the reference table below, covering approval state, notification, durable resume, audit storage and review dashboards on AWS.
+-----------------------------+---------------------+------------------------------+
| Component | Technology | Role |
+-----------------------------+---------------------+------------------------------+
| Action Policy Engine | Custom | Per-action approval routing |
| | | based on risk profile |
+-----------------------------+---------------------+------------------------------+
| Pause / Resume | LangGraph interrupt | Durable suspend while |
| | + checkpointing | waiting for human |
+-----------------------------+---------------------+------------------------------+
| Approval Requests | DynamoDB | Pending request state, |
| | | decision records |
+-----------------------------+---------------------+------------------------------+
| Notifications | SNS | Alert approvers when a |
| | | decision is needed |
+-----------------------------+---------------------+------------------------------+
| Escalation | Custom + Haiku | Agent self-assessment of |
| | confidence model | when to hand off to human |
+-----------------------------+---------------------+------------------------------+
| Collaborative Edits | DynamoDB | Draft-edit-finalize flow, |
| | | edit pattern learning |
+-----------------------------+---------------------+------------------------------+
| Oversight Sampling | Custom + DynamoDB | Post-hoc review of a sample |
| | | of autonomous actions |
+-----------------------------+---------------------+------------------------------+
| Timeout Handling | Scheduled Lambda | Policy-defined action on |
| | | expired approval requests |
+-----------------------------+---------------------+------------------------------+
| Audit Trail | DynamoDB hash chain | Immutable, tamper-evident |
| | | record of all decisions |
+-----------------------------+---------------------+------------------------------+
| Auth for Approvers | CredentialManager | Role-based approval |
| | (Article 5) | authorization |
+-----------------------------+---------------------+------------------------------+
| Learning Loop | Article 8 eval | Edits and reviews feed the |
| | pipeline | improvement pipeline |
+-----------------------------+---------------------+------------------------------+
| Local Dev Alternative | Docker Compose + | Full HITL flow testable |
| | mock approver UI | without AWS |
+-----------------------------+---------------------+------------------------------+