Case 3 — HITL Financial Approval & State Modification

Executive Briefing: A corporate treasury agent processes customer refund requests and vendor disbursements. Low-risk actions (checking invoices, reading order history) execute autonomously. High-risk disbursements ($> $500$) must pause graph execution using LangGraph's functional interrupt(), persist thread state in PostgreSQL, await human operator review via an admin dashboard or CLI, and resume execution with either approval, rejection, or modified dollar amounts via Command(resume=...).

1. Enterprise Scenario & Requirements

Modern LangGraph v0.2+ replaces legacy exception-based interruptions with the functional interrupt() primitive:

1. Tool-Level Interrupt: Build a process_disbursement tool that checks transfer amount. If amount $> $500$, trigger interrupt({"action": "disbursement", "amount": amount, "recipient": recipient}).
2. Persistent Checkpointing: Compile the graph with PostgresSaver (or MemorySaver for local tests) scoped by thread_id.
3. Suspended State Inspection: Demonstrate capturing the suspended state, reading the pending approval payload from outside the graph, and simulating human review.
4. Resumption with Command: Resume execution by passing Command(resume={"approved": True, "edited_amount": 450}) to graph.stream().

2. Explicit Deliverables

1. disbursement_agent.py: Compiled LangGraph workflow with functional interrupt() trigger and checkpointer.
2. admin_review_cli.py: Interactive CLI tool that lists paused threads, displays pending transaction details, and allows the operator to Approve, Reject, or Edit.
3. test_hitl.py: Automated test suite verifying: (a) sub-$500 transactions execute without pausing, (b) large transactions pause cleanly at the interrupt node, (c) approval resumes and completes the transfer, and (d) rejection halts and notifies the customer.
4. session_replay.log: Complete log of an interrupted session being resumed and completed.

3. Starter Code & Scaffold

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from typing import TypedDict, Annotated
import operator

class TreasuryState(TypedDict):
    messages: Annotated[list, operator.add]
    approved: bool

def disbursement_node(state: TreasuryState):
    amount = 1250.0  # High stakes transaction
    # Functional interrupt halts execution and passes payload to external caller
    approval_response = interrupt({
        "task": "authorize_disbursement",
        "amount": amount,
        "recipient": "Vendor-AcmeCorp"
    })
    
    if approval_response.get("approved"):
        final_amt = approval_response.get("edited_amount", amount)
        return {"approved": True, "messages": [f"Disbursed ${final_amt} successfully."]}
    return {"approved": False, "messages": ["Disbursement rejected by compliance officer."]}

4. 100-Point Grading Rubric