Human-in-the-Loop for AI Agents: Cryptographic Dual-Key Approvals

import time
import hmac
import hashlib
import secrets
from typing import TypedDict, Optional, Dict, Any
from langgraph.graph import StateGraph, END
from langgraph.types import interrupt

# Shared server-side authorization secret
AUTH_SECRET_KEY = b"production-crypto-key-dual-approval-2026"
TOKEN_TTL_SECONDS = 900  # 15 minutes

class AgentApprovalState(TypedDict):
    thread_id: str
    proposed_action: str
    target_resource: str
    estimated_cost: float
    action_tier: int          # Tier 1 (Auto), Tier 2 (Sandbox), Tier 3 (HITL)
    approval_token: Optional[str]
    timestamp: float
    nonce: str
    execution_status: str

# Helper: Generate cryptographically signed token
def generate_approval_token(thread_id: str, action: str, ts: float, nonce: str) -> str:
    payload = f"{thread_id}:{action}:{ts}:{nonce}".encode()
    return hmac.new(AUTH_SECRET_KEY, payload, hashlib.sha256).hexdigest()

# Helper: Verify signature and TTL decay
def verify_token(token: str, thread_id: str, action: str, ts: float, nonce: str) -> bool:
    if time.time() - ts > TOKEN_TTL_SECONDS:
        return False  # Expired token
    expected = generate_approval_token(thread_id, action, ts, nonce)
    return hmac.compare_digest(token, expected)

# Node 1: Action Proposal
def propose_mutation_node(state: AgentApprovalState) -> Dict[str, Any]:
    return {
        "proposed_action": "MIGRATE_SCHEMA_DROP_COLUMN",
        "target_resource": "users_v2.legacy_credentials",
        "action_tier": 3,
        "timestamp": time.time(),
        "nonce": secrets.token_hex(8)
    }

# Node 2: Cryptographic HITL Gate Node
def dual_key_gate_node(state: AgentApprovalState) -> Dict[str, Any]:
    if state["action_tier"] >= 3:
        # Suspend execution and generate human review payload
        human_decision = interrupt({
            "prompt": "Tier 3 Destructive Mutation Requires Cryptographic Signoff",
            "action": state["proposed_action"],
            "target": state["target_resource"],
            "nonce": state["nonce"],
            "expires_in": f"{TOKEN_TTL_SECONDS}s"
        })
        
        token = human_decision.get("signature_token", "")
        is_valid = verify_token(
            token=token,
            thread_id=state["thread_id"],
            action=state["proposed_action"],
            ts=state["timestamp"],
            nonce=state["nonce"]
        )
        
        if is_valid and human_decision.get("authorized"):
            return {"approval_token": token, "execution_status": "AUTHORIZED"}
        return {"approval_token": None, "execution_status": "REJECTED_BY_POLICY"}
    
    return {"execution_status": "AUTO_APPROVED"}

# Node 3: Safe Commit Node
def commit_action_node(state: AgentApprovalState) -> Dict[str, Any]:
    if state["execution_status"] == "AUTHORIZED":
        return {"execution_status": f"Committed: {state['proposed_action']} on {state['target_resource']}"}
    return {"execution_status": "Halted: Action was not authorized by human operator."}

# Assemble State Graph
builder = StateGraph(AgentApprovalState)
builder.add_node("propose", propose_mutation_node)
builder.add_node("hitl_gate", dual_key_gate_node)
builder.add_node("commit", commit_action_node)

builder.set_entry_point("propose")
builder.add_edge("propose", "hitl_gate")
builder.add_edge("hitl_gate", "commit")
builder.add_edge("commit", END)

hitl_graph = builder.compile()