{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Human-in-the-Loop: Pre-Action Gates, Risk Tiering, and Audit Logging\n",
    "\n",
    "The README for this lesson introduces Human-in-the-Loop with a short snippet that asks the user `APPROVE` or `REJECT` after the agent has already produced a response. That pattern is a fine starting point, but production HITL implementations commonly need three additional pieces:\n",
    "\n",
    "1. A **pre-action gate** that runs **before** the agent executes a risky step, so cost, irreversibility, and latency stay under control.\n",
    "2. **Risk tiering**, so low-risk actions auto-execute, medium-risk actions are batch-approved, and only high-risk actions block on a human.\n",
    "3. An **audit log plus revision loop**, so every gate decision is recorded as JSONL, and a rejection re-prompts the agent with a structured reason instead of just printing `Revising...`.\n",
    "\n",
    "This notebook builds each of these on top of the same primitives as `06-system-message-framework.ipynb`. It runs end-to-end in `DEMO_MODE = True` (no interactive input needed) or with real `input()` prompts when `DEMO_MODE = False`. Note: in DEMO_MODE the third goal's retry is scripted so the loop mechanics are visible end-to-end. Real revision-driven re-classification requires `DEMO_MODE = False` and an operator.\n",
    "\n",
    "**Out of scope (handled in other lessons):** authentication and access control (Lesson 06 README threat 2), tool-call middleware (Lesson 14 MAF deep dive), multi-agent debate patterns.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import os\n",
    "from datetime import datetime, timezone\n",
    "from pathlib import Path\n",
    "\n",
    "from dotenv import load_dotenv\n",
    "from azure.identity import DefaultAzureCredential, get_bearer_token_provider\n",
    "from openai import OpenAI\n",
    "\n",
    "load_dotenv()\n",
    "\n",
    "DEMO_MODE = True  # set False to use real input() prompts\n",
    "\n",
    "# Per-run unique log filename so demo runs don't overwrite each other and\n",
    "# the notebook doesn't touch any pre-existing gate_log.jsonl in the working\n",
    "# directory.\n",
    "GATE_LOG_PATH = Path(\n",
    "    f\"gate_log_{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}.jsonl\"\n",
    ")\n",
    "\n",
    "# This notebook uses the Azure OpenAI Responses API via the stable /openai/v1/ endpoint.\n",
    "# GitHub Models is deprecated (retiring July 2026) and does not support the Responses API.\n",
    "endpoint = os.environ.get(\"AZURE_OPENAI_ENDPOINT\", \"\")\n",
    "if not endpoint:\n",
    "    raise RuntimeError(\n",
    "        \"AZURE_OPENAI_ENDPOINT environment variable is not set. This notebook needs \"\n",
    "        \"an Azure OpenAI resource with a model deployment that supports the Responses \"\n",
    "        \"API. Set AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_DEPLOYMENT in \"\n",
    "        \"your environment or a local .env file, then run `az login`.\"\n",
    "    )\n",
    "\n",
    "deployment = os.environ[\"AZURE_OPENAI_DEPLOYMENT\"]\n",
    "\n",
    "# Authenticate with Entra ID (run `az login` first). No api_version is needed.\n",
    "token_provider = get_bearer_token_provider(\n",
    "    DefaultAzureCredential(),\n",
    "    \"https://cognitiveservices.azure.com/.default\",\n",
    ")\n",
    "\n",
    "client = OpenAI(\n",
    "    base_url=f\"{endpoint.rstrip('/')}/openai/v1/\",\n",
    "    api_key=token_provider,\n",
    ")\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pattern 1: Pre-action gate\n",
    "\n",
    "The README's HITL snippet calls the agent first, then asks the user to approve the output. That is a **post-action** flow. The agent has already executed, so the LLM call cost is already paid, and any side effect (sent email, written database row, posted comment) has already happened.\n",
    "\n",
    "A **pre-action** flow inserts the gate before the agent runs the risky step. The agent proposes the action, the gate decides whether to execute, and only on approval does the side effect occur.\n",
    "\n",
    "| Aspect | Post-action approval (README snippet) | Pre-action gate (this notebook) |\n",
    "|---|---|---|\n",
    "| When does approval run? | After the agent has produced output | Before any side-effect executes |\n",
    "| LLM cost on rejection | Already paid | Paid only for the proposal, not the action |\n",
    "| Irreversible side effects | Possible (the action already happened) | Prevented |\n",
    "| Audit clarity | Approval is a print statement | Approval is a JSON record with timestamp, action, reason |\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def gate_action(action_description: str, risk_tier: str, attempt: int = 0) -> dict:\n",
    "    \"\"\"Run a single pre-action gate.\n",
    "\n",
    "    Returns a decision dict with keys: decision, reason, ts.\n",
    "    Decision is one of: approve, deny, escalate.\n",
    "    Safe default on EOF or unexpected input is deny.\n",
    "\n",
    "    DEMO_MODE behavior: high-risk actions are denied on attempt 0 and\n",
    "    auto-approved on attempt >= 1. This is scripted approval to show the\n",
    "    loop mechanics (deny -> retry -> approve). It is NOT revision-driven\n",
    "    re-classification. Real revision-driven re-classification requires\n",
    "    DEMO_MODE=False and a human operator who evaluates the revised\n",
    "    proposal on its own merits.\n",
    "    \"\"\"\n",
    "    print(f\"[gate] proposed action ({risk_tier}, attempt={attempt}): {action_description}\")\n",
    "\n",
    "    if DEMO_MODE:\n",
    "        if risk_tier == \"high\":\n",
    "            decision = \"approve\" if attempt >= 1 else \"deny\"\n",
    "            reason = (\n",
    "                \"DEMO_MODE: scripted approval on retry to show loop mechanics\"\n",
    "                if attempt >= 1\n",
    "                else \"DEMO_MODE: high risk denied on first attempt\"\n",
    "            )\n",
    "        else:\n",
    "            decision = \"approve\"\n",
    "            reason = f\"DEMO_MODE canned response for tier={risk_tier}\"\n",
    "    else:\n",
    "        try:\n",
    "            raw = input(\"[gate] approve / deny / escalate? \").strip().lower()\n",
    "        except EOFError:\n",
    "            raw = \"\"\n",
    "        if raw in {\"approve\", \"deny\", \"escalate\"}:\n",
    "            decision, reason = raw, \"operator input\"\n",
    "        elif raw == \"\":\n",
    "            decision, reason = \"deny\", \"no input received, defaulted to deny\"\n",
    "        else:\n",
    "            decision, reason = \"deny\", f\"invalid input {raw!r}, defaulted to deny\"\n",
    "\n",
    "    return {\n",
    "        \"decision\": decision,\n",
    "        \"reason\": reason,\n",
    "        \"action\": action_description,\n",
    "        \"risk_tier\": risk_tier,\n",
    "        \"ts\": datetime.now(timezone.utc).isoformat(),\n",
    "    }\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pattern 2: Risk tiering\n",
    "\n",
    "Not every action needs human approval. A read-only lookup against a public API has different stakes than sending a customer email. Treating both the same wastes operator attention and slows the agent.\n",
    "\n",
    "A simple 3-tier model:\n",
    "\n",
    "| Tier | Examples | Approval flow |\n",
    "|---|---|---|\n",
    "| `low` (read-only) | Search a knowledge base, look up flight options, fetch a public web page | Auto-execute, logged for audit |\n",
    "| `medium` (cheap mutation) | Cache a result, increment a counter, schedule a reminder | Auto-execute, but batched daily review |\n",
    "| `high` (external-facing or irreversible) | Send an email, charge a card, post to a public channel | Block on human approval |\n",
    "\n",
    "This is one tiering. Production systems often use more granular tiers (e.g., AWS IAM permission levels, role-based access tiers). The 3-tier version below is the smallest useful version for an agent that mixes read-only and side-effecting actions.\n",
    "\n",
    "The classifier below uses keyword heuristics so the demo stays deterministic and cheap. In a production system you would swap this for a learned classifier or a policy engine.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "LOW_RISK_KEYWORDS = {\n",
    "    \"look\", \"lookup\", \"search\", \"fetch\", \"read\", \"query\", \"view\",\n",
    "    \"get\", \"list\", \"weather\", \"summarize\",\n",
    "}\n",
    "HIGH_RISK_KEYWORDS = {\n",
    "    \"send\", \"email\", \"post\", \"publish\", \"charge\", \"pay\", \"transfer\",\n",
    "    \"delete\", \"drop\", \"cancel\", \"refund\",\n",
    "}\n",
    "MEDIUM_RISK_KEYWORDS = {\n",
    "    \"cache\", \"schedule\", \"reminder\", \"book\", \"reserve\", \"update\",\n",
    "    \"increment\", \"log\",\n",
    "}\n",
    "\n",
    "AUTO_APPROVE_REASONS = {\n",
    "    \"low\": \"auto-approved (low risk)\",\n",
    "    \"medium\": \"auto-approved (medium risk, queued for batched review)\",\n",
    "}\n",
    "\n",
    "\n",
    "def classify_risk(action: str) -> str:\n",
    "    \"\"\"Classify an action string into one of: low, medium, high.\n",
    "\n",
    "    Keyword-based heuristic. Checks high-risk first (most severe), then\n",
    "    low-risk explicit reads, then medium-risk mutations. Unrecognized\n",
    "    actions default to medium, not low.\n",
    "\n",
    "    Default for unrecognized actions is 'medium', not 'low'. A read-only\n",
    "    keyword set will always have blind spots, and the parent README's\n",
    "    threat list (critical-system access, knowledge-base poisoning,\n",
    "    cascading errors) all involve cases an action-name alone cannot rule\n",
    "    out. Routing unknown actions through batched review is the safer\n",
    "    default than auto-executing them.\n",
    "    \"\"\"\n",
    "    text = action.lower()\n",
    "    if any(kw in text for kw in HIGH_RISK_KEYWORDS):\n",
    "        return \"high\"\n",
    "    if any(kw in text for kw in LOW_RISK_KEYWORDS):\n",
    "        return \"low\"\n",
    "    if any(kw in text for kw in MEDIUM_RISK_KEYWORDS):\n",
    "        return \"medium\"\n",
    "    # Explicit fail-safe default: unrecognized actions route to batched review.\n",
    "    return \"medium\"\n",
    "\n",
    "\n",
    "def tiered_gate(action: str, attempt: int = 0) -> dict:\n",
    "    \"\"\"Classify then gate. Low and medium tiers auto-approve; high blocks.\"\"\"\n",
    "    tier = classify_risk(action)\n",
    "    if tier in AUTO_APPROVE_REASONS:\n",
    "        return {\n",
    "            \"decision\": \"approve\",\n",
    "            \"reason\": AUTO_APPROVE_REASONS[tier],\n",
    "            \"action\": action,\n",
    "            \"risk_tier\": tier,\n",
    "            \"ts\": datetime.now(timezone.utc).isoformat(),\n",
    "        }\n",
    "    return gate_action(action, tier, attempt=attempt)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pattern 3: Audit log and revision loop\n",
    "\n",
    "A `print(\"Response approved.\")` is not an audit log. For trust, every gate decision should be recorded as a structured event that you can later query, replay, or attach to an incident review.\n",
    "\n",
    "Two pieces:\n",
    "\n",
    "1. **Append-only JSONL.** One line per decision, with timestamp, action, tier, decision, reason. Easy to grep, easy to ship to a real log store later.\n",
    "2. **Revision loop on rejection.** When the gate returns `deny`, the agent re-prompts itself with the rejection reason in context, so the next proposal can avoid the problem.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def log_decision(decision: dict) -> None:\n",
    "    \"\"\"Append a gate decision to the JSONL audit log.\"\"\"\n",
    "    with GATE_LOG_PATH.open(\"a\", encoding=\"utf-8\") as f:\n",
    "        f.write(json.dumps(decision) + \"\\n\")\n",
    "\n",
    "\n",
    "def propose_action(goal: str, prior_rejection: str | None = None) -> str:\n",
    "    \"\"\"Ask the LLM to propose a concrete next action for a goal.\n",
    "\n",
    "    If prior_rejection is provided, it is fed back so the LLM can avoid\n",
    "    the same failure mode in the next proposal.\n",
    "    \"\"\"\n",
    "    system = (\n",
    "        \"You are an action planner for an agent. Propose ONE concrete next\\n\"\n",
    "        \"action (a single sentence) toward the user's goal. If a prior\\n\"\n",
    "        \"rejection reason is given, propose a different action that addresses\\n\"\n",
    "        \"the rejection.\"\n",
    "    )\n",
    "    user_text = f\"Goal: {goal}\"\n",
    "    if prior_rejection:\n",
    "        user_text += f\"\\n\\nPrior proposal was denied. Reason: {prior_rejection}\"\n",
    "\n",
    "    response = client.responses.create(\n",
    "        model=deployment,\n",
    "        input=[\n",
    "            {\"role\": \"system\", \"content\": system},\n",
    "            {\"role\": \"user\", \"content\": user_text},\n",
    "        ],\n",
    "        store=False,\n",
    "    )\n",
    "    return response.output_text.strip()\n",
    "\n",
    "\n",
    "def run_with_revision(goal: str, max_revisions: int = 2) -> dict:\n",
    "    \"\"\"Propose, gate, and on rejection revise up to max_revisions times.\"\"\"\n",
    "    prior_reason: str | None = None\n",
    "    for attempt in range(max_revisions + 1):\n",
    "        action = propose_action(goal, prior_rejection=prior_reason)\n",
    "        decision = tiered_gate(action, attempt=attempt)\n",
    "        decision[\"attempt\"] = attempt\n",
    "        log_decision(decision)\n",
    "        if decision[\"decision\"] == \"approve\":\n",
    "            return decision\n",
    "        prior_reason = decision[\"reason\"]\n",
    "    return {**decision, \"final\": \"max_revisions_reached\"}\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# End-to-end demo: three goals at three different risk profiles.\n",
    "# GATE_LOG_PATH is per-run (timestamped) so no prior log is touched.\n",
    "\n",
    "goals = [\n",
    "    \"Look up the weather in Seattle for the customer's trip planning.\",\n",
    "    \"Schedule a reminder for the customer to check in 24 hours before their flight.\",\n",
    "    \"Send a marketing email to the customer about premium upgrade options.\",\n",
    "]\n",
    "\n",
    "for goal in goals:\n",
    "    print(f\"\\n=== Goal: {goal} ===\")\n",
    "    outcome = run_with_revision(goal, max_revisions=1)\n",
    "    print(f\"[final] {outcome['decision']} ({outcome['reason']})\")\n",
    "\n",
    "print(f\"\\n=== Audit log ({GATE_LOG_PATH.name}) ===\")\n",
    "for line in GATE_LOG_PATH.read_text(encoding=\"utf-8\").splitlines():\n",
    "    record = json.loads(line)\n",
    "    print(f\"  [{record['risk_tier']:6s}] {record['decision']:8s} \"\n",
    "          f\"attempt={record.get('attempt', '?')} action={record['action'][:140]}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Additional resources\n",
    "\n",
    "Several other public projects implement variations of these HITL patterns. Compare approaches to find what fits your stack:\n",
    "\n",
    "- **LangChain** human-in-the-loop tool wrappers ([docs](https://python.langchain.com/docs/integrations/tools/human_tools)): drop-in tool wrappers that pause execution for human input.\n",
    "- **AutoGen** `UserProxyAgent` ([v0.2 docs](https://microsoft.github.io/autogen/0.2/docs/topics/human-in-the-loop); AutoGen v0.4+ restructured this): uses an agent role specifically to represent the human in multi-agent conversations.\n",
    "- **Microsoft Agent Framework (MAF)** function-invocation middleware ([docs](https://learn.microsoft.com/agent-framework/)): middleware that runs around every tool/function call, suitable for gating logic and approval flows.\n",
    "\n",
    "Each project handles the three sub-patterns differently: LangChain wraps them as tools, AutoGen uses an agent role, and Microsoft Agent Framework uses function-invocation middleware. Read one or two implementations end-to-end before picking a design for your own agent.\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python",
   "version": "3.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}