승인 없이는 못 지나가는 그래프

draft → validate → prepare
    ├─ 저위험 → auto_approve → execute → END
    └─ 고위험 → approval          (여기서 interrupt 로 멈춘다)
                  ├─ 반려 → END
                  ├─ 수정 → revise → approval
                  └─ 승인 → execute → END

from sec_graph import build_graph

graph = build_graph()
print(graph.get_graph().draw_mermaid())

graph TD;
	__start__([__start__]):::first
	draft(draft)
	validate(validate)
	prepare(prepare)
	auto_approve(auto_approve)
	approval(approval)
	revise(revise)
	execute(execute)
	__end__([__end__]):::last
	__start__ --> draft;
	approval -.-> __end__;
	approval -.-> execute;
	approval -.-> revise;
	auto_approve --> execute;
	draft --> validate;
	prepare -.-> approval;
	prepare -.-> auto_approve;
	revise --> approval;
	validate --> prepare;
	execute --> __end__;

"""보안 특화 Agent 그래프 골격.

raw/07_CODE_ARCHITECTURE/03_EXAMPLES/03_HITL_확장예시.md의 실무형 흐름을
LangGraph 코드로 구현한 것이다.

    draft -> validate -> prepare_action
        |- 저위험 -> auto_approve -> execute
        `- 고위험 -> interrupt
                        |- reject  -> END
                        |- edit    -> revise -> interrupt
                        `- approve -> execute -> END

두 가지 원칙을 구조로 지킨다.

  1) 승인 화면에 보여 준 값(pending_action)만 실행한다.
     draft 는 사람이 읽는 문장이고, pending_action 은 실행 대상이다.
  2) 멱등성 키는 입력에서 파생한다. 실행 시점에 새로 만들지 않는다.
"""
import hashlib
from typing import Literal

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

from sec_state import SecureAgentState

MAX_RETRY = 2

# 이미 실행한 action_id를 기록한다. 운영에서는 durable 저장소로 교체한다.
# 주의: 이 집합은 확인과 추가가 원자적이지 않아 동시 요청을 막지 못한다.
_EXECUTED_ACTION_IDS: set = set()


def make_action_id(request, requester_id, pending_action):
    """입력에서 파생한 멱등성 키.

    같은 요청·요청자·실행 대상이면 같은 키가 나온다.
    실행 시점에 무작위로 만들면 같은 작업이 매번 다른 키를 갖게 되어
    중복 실행을 막지 못한다.
    """
    material = "|".join([request, requester_id, pending_action])
    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]


def make_draft(state: SecureAgentState) -> dict:
    """LLM 호출 위치. 여기서는 side effect를 만들지 않는다."""
    return {
        "draft": f"'{state['request']}'에 대한 초안",
        "audit_log": [f"draft 생성 (requester={state['requester_id']})"],
    }


def validate(state: SecureAgentState) -> dict:
    """규칙 기반 사전 검사. LLM 판단보다 먼저 결정적 규칙을 적용한다."""
    high_risk_markers = ["삭제", "발송", "결제", "권한"]
    risk = "high" if any(m in state["request"] for m in high_risk_markers) else "low"
    return {"risk_level": risk, "audit_log": [f"validate 완료 (risk={risk})"]}


def prepare_action(state: SecureAgentState) -> dict:
    """실행할 내용을 확정하고 멱등성 키를 만든다. 아직 실행하지 않는다.

    여기서 확정한 pending_action 이 승인 화면에 그대로 실리고,
    실행도 이 값을 대상으로 한다. 셋이 어긋나지 않게 하는 것이 목적이다.
    """
    pending = f"[{state['risk_level']}] {state['draft']}"
    action_id = make_action_id(state["request"], state["requester_id"], pending)
    return {
        "pending_action": pending,
        "executed_action_id": action_id,
        "audit_log": [
            f"prepare_action 완료 (action_id={action_id}). 아직 실행하지 않음"
        ],
    }


def auto_approve(state: SecureAgentState) -> dict:
    """저위험 작업은 사람 승인 없이 통과시키되 그 사실을 기록한다.

    '승인 없이 실행'이 아니라 '자동 승인'으로 다룬다.
    누가 통과시켰는지가 감사 로그에 남아야 하기 때문이다.
    """
    return {
        "approved": True,
        "audit_log": ["저위험으로 자동 승인 (사람 개입 없음)"],
    }


def approval(state: SecureAgentState) -> dict:
    """사람에게 승인을 요청하고 멈춘다.

    payload에는 직렬화 가능한 값만 넣는다.
    실행 대상인 pending_action 을 그대로 보여 준다.
    """
    decision = interrupt(
        {
            "question": "이 작업을 실행할까요?",
            "risk_level": state["risk_level"],
            "pending_action": state["pending_action"],
            "options": ["approve", "edit", "reject"],
        }
    )
    return {
        "approved": decision == "approve",
        "audit_log": [f"사람 결정: {decision}"],
    }


def revise(state: SecureAgentState) -> dict:
    """초안을 고친다. 내용이 바뀌었으므로 실행 대상과 키를 다시 확정한다."""
    draft = state["draft"] + " (수정본)"
    pending = f"[{state['risk_level']}] {draft}"
    return {
        "retry_count": state["retry_count"] + 1,
        "draft": draft,
        "pending_action": pending,
        "executed_action_id": make_action_id(
            state["request"], state["requester_id"], pending
        ),
        "audit_log": ["revise 수행. 실행 대상과 키를 다시 확정"],
    }


def execute_side_effect(state: SecureAgentState) -> dict:
    """승인된 경우에만 실제 행동을 수행한다.

    실행 대상은 pending_action 이다. 승인 화면에 보인 값과 같다.
    키는 prepare_action 이 확정한 것을 쓰고 여기서 새로 만들지 않는다.
    """
    if not state["approved"]:
        return {"result": "실행하지 않음", "audit_log": ["미승인으로 종료"]}

    action_id = state["executed_action_id"]
    if action_id in _EXECUTED_ACTION_IDS:
        return {
            "result": "이미 실행됨",
            "audit_log": [f"중복 실행 차단 (action_id={action_id})"],
        }

    _EXECUTED_ACTION_IDS.add(action_id)
    # 실제 발송/삭제/결제 코드는 이 위치에 둔다.
    # 반드시 state["pending_action"] 을 대상으로 해야 승인한 것과 같아진다.
    return {
        "result": f"실행 완료: {state['pending_action']}",
        "executed_action_id": action_id,
        "audit_log": [f"side effect 실행 (action_id={action_id})"],
    }


def route_risk(state: SecureAgentState) -> Literal["approval", "auto_approve"]:
    """고위험이면 사람 승인을 거치고, 저위험이면 자동 승인으로 간다."""
    return "approval" if state["risk_level"] == "high" else "auto_approve"


def route_after_approval(state: SecureAgentState) -> Literal["execute", "revise", "__end__"]:
    if state["approved"]:
        return "execute"
    if state["retry_count"] >= MAX_RETRY:
        return "__end__"
    return "revise"


def build_graph():
    builder = StateGraph(SecureAgentState)

    builder.add_node("draft", make_draft)
    builder.add_node("validate", validate)
    builder.add_node("prepare", prepare_action)
    builder.add_node("auto_approve", auto_approve)
    builder.add_node("approval", approval)
    builder.add_node("revise", revise)
    builder.add_node("execute", execute_side_effect)

    builder.add_edge(START, "draft")
    builder.add_edge("draft", "validate")
    builder.add_edge("validate", "prepare")
    builder.add_conditional_edges("prepare", route_risk)
    builder.add_edge("auto_approve", "execute")
    builder.add_conditional_edges("approval", route_after_approval)
    builder.add_edge("revise", "approval")
    builder.add_edge("execute", END)

    # HITL을 쓰므로 checkpointer는 필수다. 운영에서는 durable backend로 교체한다.
    return builder.compile(checkpointer=InMemorySaver())

    if not state["approved"]:
        return {"result": "실행하지 않음", "audit_log": ["미승인으로 종료"]}

    action_id = state["executed_action_id"]
    ...
    # 반드시 state["pending_action"] 을 대상으로 해야 승인한 것과 같아진다.