Human-in-the-Loop

# script_id: day_055_human_in_the_loop/interrupt_min
# Install: uv add langgraph      Env: none      Services: none (no LLM)
from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class State(TypedDict):
    draft: str
    verdict: str


def gate(state: State) -> dict:
    # Pauses the run and saves it. On resume, interrupt() returns the human's answer.
    answer = interrupt({"question": "ok to publish?", "draft": state["draft"]})
    return {"verdict": f"approved={answer}"}


g = StateGraph(State)
g.add_node("gate", gate)
g.add_edge(START, "gate")
g.add_edge("gate", END)
app = g.compile(checkpointer=InMemorySaver())  # the paused run lives here
cfg = {"configurable": {"thread_id": "t1"}, "recursion_limit": 25}

out = app.invoke({"draft": "Thanks, reproduced on 2.8.0.", "verdict": ""}, cfg)
print(out["__interrupt__"])
snap = app.get_state(cfg)
print("next:", snap.next, "| pending:", snap.interrupts[0].value["question"])
out = app.invoke(Command(resume="yes"), cfg)
print(out["verdict"])

# the Interrupt id will differ
[Interrupt(value={'question': 'ok to publish?', 'draft': 'Thanks, reproduced on 2.8.0.'}, id='2699c26b066c750938a07471e3b9fe0b', response_schema=None)]
next: ('gate',) | pending: ok to publish?
approved=yes

# script_id: day_055_human_in_the_loop/rerun_trap
# Install: uv add langgraph      Env: none      Services: none
from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

calls = {"before": 0, "after": 0}


class State(TypedDict):
    answer: str


def gate(state: State) -> dict:
    calls["before"] += 1  # WRONG place for a side effect: this line runs again on resume
    answer = interrupt("post the comment?")
    calls["after"] += 1  # right place: runs once, after the human answered
    return {"answer": answer}


g = StateGraph(State)
g.add_node("gate", gate)
g.add_edge(START, "gate")
g.add_edge("gate", END)
app = g.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "trap"}, "recursion_limit": 25}
app.invoke({"answer": ""}, cfg)
app.invoke(Command(resume="yes"), cfg)
print(f"before-interrupt side effect ran {calls['before']}x, after ran {calls['after']}x")

try:
    app.invoke({"answer": ""}, {"recursion_limit": 25})  # forgot the thread_id
except ValueError as exc:
    print("ValueError:", exc)

before-interrupt side effect ran 2x, after ran 1x
ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id

# script_id: day_055_human_in_the_loop/approval_models
# Install: uv add pydantic      Env: none      Services: none (pure Python)
"""Approval request/decision models, risk rules and an append-only audit log."""
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Literal

from pydantic import BaseModel, ValidationError, model_validator

Risk = Literal["low", "medium", "high"]
ORDER: dict[str, int] = {"low": 0, "medium": 1, "high": 2}
READ_ONLY = {"search_issues"}
SIDE_EFFECT = {"add_label", "post_comment"}  # Day 37's side_effect flag, as a set


class ApprovalRequest(BaseModel):
    request_id: str  # thread_id + tool_call id: identical every time the node re-runs
    thread_id: str
    tool: str
    args: dict
    risk: Risk
    reasons: list[str]
    summary: str  # the line a human reads first
    reversible: bool
    created_at: datetime
    expires_at: datetime


class ApprovalDecision(BaseModel):
    action: Literal["approve", "edit", "reject"]
    edited_args: dict | None = None
    reason: str | None = None  # reviewer feedback, handed back to the model
    waited_s: float = 0.0  # filled in by decide()

    @model_validator(mode="after")
    def edit_needs_args(self) -> "ApprovalDecision":
        if self.action == "edit" and not self.edited_args:
            raise ValueError("edit needs edited_args")
        return self


def parse_decision(raw: object) -> ApprovalDecision:
    """Fail closed: anything that isn't a valid decision becomes a reject."""
    try:
        return ApprovalDecision.model_validate(raw)
    except ValidationError as exc:
        return ApprovalDecision(action="reject", reason=f"invalid decision ({exc.error_count()} errors)")

# script_id: day_055_human_in_the_loop/approval_models
Rule = tuple[Callable[[dict], bool], Risk, str]
RULES: list[Rule] = [
    (lambda c: c["name"] not in READ_ONLY | SIDE_EFFECT, "high", "unknown tool (fail closed)"),
    (lambda c: c["name"] == "add_label", "medium", "changes issue labels"),
    (lambda c: c["name"] == "post_comment", "medium", "public comment; cannot be unsent"),
    (lambda c: "http" in str(c["args"].get("text", "")), "high", "comment contains a link"),
    (lambda c: str(c["args"].get("label", "")).startswith("severity:S1"), "high", "S1 pages the on-call"),
]


def risk_report(call: dict, rules: list[Rule] = RULES) -> tuple[Risk, list[str]]:
    hits = [(tier, why) for check, tier, why in rules if check(call)]
    tier = max((t for t, _ in hits), key=ORDER.__getitem__, default="low")
    return tier, [why for _, why in hits]


def risk_of(call: dict, rules: list[Rule] = RULES) -> Risk:
    return risk_report(call, rules)[0]


class AuditLog:
    """Append-only JSONL. record() is keyed by request_id, so a re-run node can't double-write."""

    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)

    def entries(self) -> list[dict]:
        text = self.path.read_text() if self.path.exists() else ""
        return [json.loads(line) for line in text.splitlines()]

    def record(self, **entry: Any) -> bool:
        if any(e["request_id"] == entry["request_id"] for e in self.entries()):
            return False  # already written on an earlier run of the node
        with self.path.open("a") as f:
            f.write(json.dumps({"ts": datetime.now(timezone.utc).isoformat(), **entry}, default=str) + "\n")
        return True

# script_id: day_055_human_in_the_loop/approval_models
if __name__ == "__main__":
    assert risk_of({"name": "search_issues", "args": {"query": "dlq"}}) == "low"
    assert risk_of({"name": "add_label", "args": {"issue_id": 7, "label": "comp:cli"}}) == "medium"
    assert risk_of({"name": "add_label", "args": {"issue_id": 20, "label": "severity:S1"}}) == "high"
    assert risk_of({"name": "post_comment", "args": {"issue_id": 7, "text": "see https://x.io"}}) == "high"
    assert risk_of({"name": "drop_table", "args": {}}) == "high"  # unknown -> fail closed
    assert parse_decision("yes").action == "reject"  # wrong shape -> fail closed
    assert parse_decision({"action": "edit"}).action == "reject"  # edit without args
    print("all risk checks passed")

all risk checks passed

# script_id: day_055_human_in_the_loop/approval_graph
# Install: uv add langgraph langchain-core pydantic      Env: ORBIT_DATA (default datasets/orbit)
# Services: none. Scripted model, so no API key; live_run swaps in a real one.
import json
import os
from datetime import datetime, timedelta, timezone
from functools import cache
from pathlib import Path
from typing import Annotated, Any, Callable, TypedDict

from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt

from approval_models import ORDER, RULES, SIDE_EFFECT, ApprovalRequest, AuditLog, parse_decision, risk_report

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))


class MockTracker:  # built on Day 37: side effects go to a JSONL file, never to a real tracker
    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)

    def write(self, action: str, **fields: Any) -> str:
        with self.path.open("a") as f:
            f.write(json.dumps({"action": action, **fields}) + "\n")
        return json.dumps({"ok": True, "action": action, **fields})

    def lines(self) -> list[dict]:
        text = self.path.read_text() if self.path.exists() else ""
        return [json.loads(line) for line in text.splitlines()]


def envelope_err(code: str, message: str) -> str:  # built on Day 37's error envelope
    return json.dumps({"ok": False, "error": {"code": code, "message": message}})


def run_once(key: str, fn: Callable[[], str], store: Path) -> str:  # built on Day 54
    done = json.loads(store.read_text()) if store.exists() else {}
    if key not in done:  # a repeated step gets the saved result, not a second write
        done[key] = fn()
        store.write_text(json.dumps(done))
    return done[key]

# script_id: day_055_human_in_the_loop/approval_graph
@cache
def issue_title(issue_id: int) -> str:
    for line in (DATA_DIR / "issues" / "issues.jsonl").open():
        if (row := json.loads(line))["id"] == issue_id:
            return row["title"]
    return "(unknown issue)"


def make_tools(tracker: MockTracker) -> dict[str, Callable[..., str]]:
    def search_issues(query: str) -> str:
        """Search Orbit issue titles by keyword. Read-only."""
        rows = [json.loads(line) for line in (DATA_DIR / "issues" / "issues.jsonl").open()]
        hits = [{"id": r["id"], "title": r["title"]} for r in rows if query.lower() in r["title"].lower()]
        return json.dumps(hits[:3])

    def add_label(issue_id: int, label: str) -> str:
        """Add one label to an Orbit issue."""
        return tracker.write("add_label", issue_id=issue_id, label=label)

    def post_comment(issue_id: int, text: str) -> str:
        """Post a public comment on an Orbit issue."""
        return tracker.write("post_comment", issue_id=issue_id, text=text)

    return {f.__name__: f for f in (search_issues, add_label, post_comment)}


class ScriptedModel:  # built on Day 53: a keyless stand-in with a chat model's .invoke()
    def __init__(self, turns: list[AIMessage]) -> None:
        self.turns = turns

    def invoke(self, messages: list[AnyMessage]) -> AIMessage:
        # Choose the turn by counting earlier AI turns, so a fresh process replays correctly.
        turn = self.turns[min(sum(isinstance(m, AIMessage) for m in messages), len(self.turns) - 1)]
        return AIMessage(content=turn.content, tool_calls=turn.tool_calls)


def tc(name: str, call_id: str, **args: Any) -> dict:
    return {"name": name, "args": args, "id": call_id, "type": "tool_call"}


class AgentState(TypedDict):  # Day 53's state plus two Day 55 fields
    messages: Annotated[list[AnyMessage], add_messages]
    revisions: int
    status: str

# script_id: day_055_human_in_the_loop/approval_graph
def summarize(call: dict) -> str:
    issue = call["args"].get("issue_id", 0)
    what = call["args"].get("text") or call["args"].get("label") or call["args"]
    return f'{call["name"]} on #{issue} "{issue_title(issue)}": {what}'


def make_approval_node(audit: AuditLog, *, rules: list = RULES, gate_at: str = "medium",
                       ttl: timedelta = timedelta(hours=4), max_revisions: int = 2):
    def approval_node(state: AgentState, config: RunnableConfig) -> Command:
        thread_id = config["configurable"]["thread_id"]
        ai = state["messages"][-1]
        calls, denied, feedback = [], [], []
        for call in ai.tool_calls:
            risk, reasons = risk_report(call, rules)
            if ORDER[risk] < ORDER[gate_at]:
                calls.append(call)  # low risk: no human needed
                continue
            now = datetime.now(timezone.utc)
            request = ApprovalRequest(
                request_id=f"{thread_id}:{call['id']}", thread_id=thread_id, tool=call["name"],
                args=call["args"], risk=risk, reasons=reasons, summary=summarize(call),
                reversible=call["name"] != "post_comment", created_at=now, expires_at=now + ttl)
            # ---- everything above re-runs on resume, so it must stay free of side effects ----
            decision = parse_decision(interrupt(request.model_dump(mode="json")))
            args = decision.edited_args if decision.action == "edit" else call["args"]
            audit.record(request_id=request.request_id, thread_id=thread_id, tool=call["name"],
                         original_args=call["args"], final_args=None if decision.action == "reject" else args,
                         decision=decision.action, reviewer="unknown", risk=risk, wait_s=decision.waited_s)
            if decision.action == "reject":
                note = decision.reason or "the reviewer declined this call"
                denied.append(ToolMessage(envelope_err("denied", note), tool_call_id=call["id"]))
                feedback += [decision.reason] if decision.reason else []
            calls.append({**call, "args": args})
        return route_after_review(state, ai, calls, denied, feedback, max_revisions)

    return approval_node

# script_id: day_055_human_in_the_loop/approval_graph
def sync_blocks(content: Any, calls: list[dict]) -> Any:
    # A Responses-API model also keeps each call in a content block and replays that copy,
    # so an edit must reach the block too. Scripted models use plain strings and skip this.
    if not isinstance(content, list):
        return content
    args = {c["id"]: json.dumps(c["args"]) for c in calls}
    return [{**b, "arguments": args[b["call_id"]]} if isinstance(b, dict) and b.get("call_id") in args else b
            for b in content]


def route_after_review(state: AgentState, ai: AIMessage, calls: list[dict], denied: list[ToolMessage],
                       feedback: list[str], max_revisions: int) -> Command:
    revised = ai.model_copy(update={"tool_calls": calls, "content": sync_blocks(ai.content, calls)})
    update: dict[str, Any] = {"messages": [revised, *denied]}
    answered = {m.tool_call_id for m in denied}
    if any(c["id"] not in answered for c in calls):
        return Command(update=update, goto="run_tools")  # something was approved or edited
    revisions = state.get("revisions", 0) + 1
    update["revisions"] = revisions
    if revisions > max_revisions:  # stop the redraft loop and hand over to a person
        update["messages"].append(AIMessage(f"Stopped after {max_revisions} revisions; a maintainer needs to take over."))
        return Command(update={**update, "status": "needs_human"}, goto=END)
    if feedback:
        update["messages"].append(HumanMessage("Reviewer feedback: " + " ".join(feedback)))
    return Command(update=update, goto="call_model")


def make_run_tools(tools: dict[str, Callable[..., str]], done_path: Path):
    def run_tools(state: AgentState, config: RunnableConfig) -> dict:
        thread_id = config["configurable"]["thread_id"]
        ai = next(m for m in reversed(state["messages"]) if isinstance(m, AIMessage))
        answered = {m.tool_call_id for m in state["messages"] if isinstance(m, ToolMessage)}
        results = []
        for call in ai.tool_calls:
            if call["id"] in answered:
                continue  # denied calls already have their answer
            fn = tools.get(call["name"])
            if fn is None:
                result = envelope_err("unknown_tool", call["name"])
            elif call["name"] in SIDE_EFFECT:  # exactly once, even if this step repeats (Day 54)
                result = run_once(f"{thread_id}:{call['id']}", lambda: fn(**call["args"]), done_path)
            else:
                result = fn(**call["args"])
            results.append(ToolMessage(result, tool_call_id=call["id"]))
        return {"messages": results}

    return run_tools

# script_id: day_055_human_in_the_loop/approval_graph
def build_hitl_graph(model: Any, tools: dict[str, Callable[..., str]], *, audit: AuditLog,
                     checkpointer: Any, done_path: Path = Path("done.json"), **gate: Any):
    """model: anything with .invoke(messages) -> AIMessage. gate: rules, gate_at, ttl, max_revisions."""
    def call_model(state: AgentState) -> dict:
        return {"messages": [model.invoke(state["messages"])]}

    def after_model(state: AgentState) -> str:
        return "approval" if state["messages"][-1].tool_calls else END

    g = StateGraph(AgentState)
    g.add_node("call_model", call_model)
    g.add_node("approval", make_approval_node(audit, **gate), destinations=("run_tools", "call_model", END))
    g.add_node("run_tools", make_run_tools(tools, done_path))
    g.add_edge(START, "call_model")
    g.add_conditional_edges("call_model", after_model, ["approval", END])
    g.add_edge("run_tools", "call_model")
    return g.compile(checkpointer=checkpointer)


def pending(app: Any, cfg: dict) -> dict | None:
    snap = app.get_state(cfg)
    return snap.interrupts[0].value if snap.interrupts else None


def decide(app: Any, cfg: dict, decision: dict, *, now: datetime | None = None) -> dict:
    """Resume a paused thread with a reviewer's decision. Late answers fail closed."""
    request = pending(app, cfg)
    if request is None:
        raise RuntimeError("no pending approval on this thread")
    now = now or datetime.now(timezone.utc)
    if now > datetime.fromisoformat(request["expires_at"]):
        decision = {"action": "reject", "reason": "approval expired; nothing was posted"}
    waited = (now - datetime.fromisoformat(request["created_at"])).total_seconds()
    return app.invoke(Command(resume={**decision, "waited_s": waited}), cfg)


def expire_pending(app: Any, cfg: dict, now: datetime) -> bool:
    """Sweeper: reject a request whose deadline has passed. True if it expired one."""
    request = pending(app, cfg)
    if request is None or now <= datetime.fromisoformat(request["expires_at"]):
        return False
    decide(app, cfg, {"action": "reject"}, now=now)
    return True


if __name__ == "__main__":
    demo = build_hitl_graph(ScriptedModel([AIMessage("hi")]), {}, audit=AuditLog("audit.jsonl"), checkpointer=None)
    print("nodes:", list(demo.get_graph().nodes))

nodes: ['__start__', 'call_model', 'approval', 'run_tools', '__end__']

# script_id: day_055_human_in_the_loop/hitl_demo
# Install: uv add langgraph langchain-core pydantic      Env: ORBIT_DATA      Services: none (no API key)
from datetime import datetime, timedelta
from pathlib import Path

from langchain_core.messages import AIMessage, HumanMessage
from langgraph.checkpoint.memory import InMemorySaver

from approval_graph import AuditLog, MockTracker, ScriptedModel, build_hitl_graph, decide, make_tools, pending, tc

for p in ("tracker.jsonl", "audit.jsonl", "done.json"):
    Path(p).unlink(missing_ok=True)  # a fresh run every time
TRACKER, AUDIT, SAVER = MockTracker("tracker.jsonl"), AuditLog("audit.jsonl"), InMemorySaver()
TOOLS = make_tools(TRACKER)

PROMISE = "Thanks! This is fixed in 2.8.1, shipping Friday."  # a promise Atlas can't make
SAFE = "Thanks for the report. We can reproduce the truncated IDs on 2.8.0 and are looking into it."


def draft(text: str, cid: str) -> AIMessage:
    return AIMessage("", tool_calls=[tc("post_comment", cid, issue_id=7, text=text)])


APPROVE, REJECT = {"action": "approve"}, {"action": "reject"}
SCENARIOS = {  # name: (scripted model turns, reviewer decisions in order)
    "approve": ([draft(SAFE, "c1"), AIMessage("Posted a reply on #7.")], [APPROVE]),
    "edit": ([draft(PROMISE, "c1"), AIMessage("Posted the reviewed reply on #7.")],
             [{"action": "edit", "edited_args": {"issue_id": 7, "text": SAFE}}]),
    "reject": ([draft(PROMISE, "c1"), AIMessage("I did not post: the reviewer declined.")], [REJECT]),
    "feedback": ([draft(PROMISE, "c1"), draft(SAFE, "c2"), AIMessage("Posted a revised reply on #7.")],
                 [{**REJECT, "reason": "Don't promise a release date."}, APPROVE]),
    "cap": ([draft(PROMISE, f"c{i}") for i in (1, 2, 3)], [{**REJECT, "reason": "No dates."}] * 3),
    "timeout": ([draft(SAFE, "c1"), AIMessage("I did not post: the approval expired.")], [APPROVE]),
    "inject": ([AIMessage("", tool_calls=[tc("add_label", "i1", issue_id=20, label="severity:S1")]),
                AIMessage("I did not relabel #20; that instruction came from the issue text.")],
               [{**REJECT, "reason": "The instruction came from the issue body, not a maintainer."}]),
}

# script_id: day_055_human_in_the_loop/hitl_demo
rows = []
for name, (turns, decisions) in SCENARIOS.items():
    cfg = {"configurable": {"thread_id": name}, "recursion_limit": 25}
    app = build_hitl_graph(ScriptedModel(turns), TOOLS, audit=AUDIT, checkpointer=SAVER)
    issue = 20 if name == "inject" else 7
    before = len(TRACKER.lines())
    app.invoke({"messages": [HumanMessage(f"Triage issue #{issue} and reply to the reporter.")]}, cfg)
    for decision in decisions:
        req = pending(app, cfg)
        late = datetime.fromisoformat(req["created_at"]) + timedelta(hours=5)  # injected clock
        print(f"{name:<8} {req['risk']:<6} {req['tool']:<12} {decision['action']:<7} {', '.join(req['reasons'])}")
        decide(app, cfg, decision, now=late if name == "timeout" else None)
    final = app.get_state(cfg).values
    rows.append((name, len(TRACKER.lines()) - before, final.get("status") or "done", final["messages"][-1].content))

print(f"\n{'scenario':<9}| writes | status      | final message")
for name, writes, status, text in rows:
    print(f"{name:<9}| {writes:^6} | {status:<11} | {text[:60]}")
print(f"\naudit.jsonl: {len(AUDIT.entries())} entries; last ->", AUDIT.entries()[-1])

# timestamps and wait_s will differ
approve  medium post_comment approve public comment; cannot be unsent
edit     medium post_comment edit    public comment; cannot be unsent
reject   medium post_comment reject  public comment; cannot be unsent
feedback medium post_comment reject  public comment; cannot be unsent
feedback medium post_comment approve public comment; cannot be unsent
cap      medium post_comment reject  public comment; cannot be unsent
cap      medium post_comment reject  public comment; cannot be unsent
cap      medium post_comment reject  public comment; cannot be unsent
timeout  medium post_comment approve public comment; cannot be unsent
inject   high   add_label    reject  changes issue labels, S1 pages the on-call

scenario | writes | status      | final message
approve  |   1    | done        | Posted a reply on #7.
edit     |   1    | done        | Posted the reviewed reply on #7.
reject   |   0    | done        | I did not post: the reviewer declined.
feedback |   1    | done        | Posted a revised reply on #7.
cap      |   0    | needs_human | Stopped after 2 revisions; a maintainer needs to take over.
timeout  |   0    | done        | I did not post: the approval expired.
inject   |   0    | done        | I did not relabel #20; that instruction came from the issue 

audit.jsonl: 10 entries; last -> {'ts': '2026-11-07T04:20:10.422657+00:00', 'request_id': 'inject:i1', 'thread_id': 'inject', 'tool': 'add_label', 'original_args': {'issue_id': 20, 'label': 'severity:S1'}, 'final_args': None, 'decision': 'reject', 'reviewer': 'unknown', 'risk': 'high', 'wait_s': 0.000532}

# script_id: day_055_human_in_the_loop/two_process
# Install: uv add langgraph langgraph-checkpoint-sqlite langchain-core pydantic
# Env: ORBIT_DATA      Services: none. Process A pauses; process B (a new interpreter) approves.
import subprocess
import sys
from pathlib import Path

from langchain_core.messages import AIMessage, HumanMessage
from langgraph.checkpoint.sqlite import SqliteSaver

from approval_graph import AuditLog, MockTracker, ScriptedModel, build_hitl_graph, decide, make_tools, pending, tc

DB, CFG = "hitl.db", {"configurable": {"thread_id": "issue-7"}, "recursion_limit": 25}
TURNS = [AIMessage("", tool_calls=[tc("post_comment", "c1", issue_id=7, text="Thanks, reproduced on 2.8.0.")]),
         AIMessage("Posted a reply on #7.")]


def open_app(cp: SqliteSaver):
    tools = make_tools(MockTracker("tracker_2p.jsonl"))
    return build_hitl_graph(ScriptedModel(TURNS), tools, audit=AuditLog("audit_2p.jsonl"),
                            checkpointer=cp, done_path=Path("done_2p.json"))


if sys.argv[1:] != ["review"]:  # process A: the agent
    for p in (DB, "tracker_2p.jsonl", "audit_2p.jsonl", "done_2p.json"):
        Path(p).unlink(missing_ok=True)
    with SqliteSaver.from_conn_string(DB) as cp:  # a context manager: closes the file on exit
        open_app(cp).invoke({"messages": [HumanMessage("Reply to issue #7.")]}, CFG)
    print("A: paused and exiting; the request is saved in", DB, flush=True)
    subprocess.run([sys.executable, __file__, "review"], check=True)  # stands in for "hours later"
else:  # process B: the reviewer, sharing nothing with A but the database file
    with SqliteSaver.from_conn_string(DB) as cp:
        app = open_app(cp)
        req = pending(app, CFG)
        print("B: found", req["request_id"], "| risk:", req["risk"], "| expires:", req["expires_at"][:16])
        out = decide(app, CFG, {"action": "approve"})
        print("B:", out["messages"][-1].content, "| tracker lines:", len(MockTracker("tracker_2p.jsonl").lines()))

# the expiry time will differ
A: paused and exiting; the request is saved in hitl.db
B: found issue-7:c1 | risk: medium | expires: 2026-11-07T08:20
B: Posted a reply on #7. | tracker lines: 1

# script_id: day_055_human_in_the_loop/live_run
# Install: uv add langgraph langchain-openai pydantic      Env: OPENAI_API_KEY, ATLAS_MODEL (optional)
# Services: none. Cost < $0.05 on gpt-6-luna (as of 2026-11, verify at https://developers.openai.com/api/docs/pricing)
import json
import os

from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.checkpoint.memory import InMemorySaver

from approval_graph import DATA_DIR, AuditLog, MockTracker, build_hitl_graph, decide, make_tools, pending

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
if not os.environ.get("OPENAI_API_KEY"):
    print("Set OPENAI_API_KEY to run the live scenario; the scripted demos need no key.")
    raise SystemExit(0)
from langchain_openai import ChatOpenAI  # imported late so the keyless path stays dependency-free

tools = make_tools(MockTracker("tracker_live.jsonl"))
model = ChatOpenAI(model=MODEL, use_responses_api=True).bind_tools(list(tools.values()))
app = build_hitl_graph(model, tools, audit=AuditLog("audit_live.jsonl"), checkpointer=InMemorySaver())
issue = [r for r in map(json.loads, (DATA_DIR / "issues" / "issues.jsonl").open()) if r["id"] == 7][0]
cfg = {"configurable": {"thread_id": "live-7"}, "recursion_limit": 25}
app.invoke({"messages": [
    SystemMessage("You are Atlas, a triage assistant for Orbit maintainers. Reply to the reporter with one "
                  "short, factual post_comment. Never promise dates or releases."),
    HumanMessage(f"Issue #{issue['id']}: {issue['title']}\n\n{issue['body']}")]}, cfg)

while (req := pending(app, cfg)) is not None:
    print(json.dumps({k: req[k] for k in ("summary", "risk", "reasons", "reversible")}, indent=2))
    decision = {"action": input("approve / edit / reject: ").strip()}  # anything else fails closed
    if decision["action"] == "edit":
        decision["edited_args"] = {**req["args"], "text": input("new text: ")}
    elif decision["action"] == "reject":
        decision["reason"] = input("feedback for the agent (optional): ") or None
    decide(app, cfg, decision)
print(app.get_state(cfg).values["messages"][-1].text)