A tool runs twiceafter resume.

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

sent = []  # stands in for the mail provider

def send_email(to: str, body: str) -> None:
    sent.append((to, body))
    print(f"  email sent to {to}")

class State(TypedDict):
    decision: str

def review(state: State):
    send_email("dana@example.com", "Your refund is being reviewed")  # side effect BEFORE the interrupt
    answer = interrupt("approve the refund?")
    return {"decision": answer}

g = StateGraph(State)
g.add_node("review", review)
g.add_edge(START, "review")
g.add_edge("review", END)
app = g.compile(checkpointer=InMemorySaver())

cfg = {"configurable": {"thread_id": "ticket-77"}}
print("run:")
app.invoke({"decision": ""}, cfg)
print("resume with approval:")
app.invoke(Command(resume="yes"), cfg)
print("emails sent:", len(sent))

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command, interrupt
from langgraph.func import entrypoint, task

# A stand-in mail provider. Like many real ones, it keeps the first message sent under a key
# and answers a repeat with that same message.
inbox: dict[str, str] = {}
def send_email(to: str, body: str, key: str | None = None) -> str:
    k = key or f"nokey-{len(inbox)}"
    if key and k in inbox:
        return k
    inbox[k] = f"{to}: {body}"
    return k

class State(TypedDict):
    decision: str

def run_graph(name, node):
    inbox.clear()
    g = StateGraph(State)
    g.add_node("review", node)
    g.add_edge(START, "review")
    g.add_edge("review", END)
    app = g.compile(checkpointer=InMemorySaver())
    cfg = {"configurable": {"thread_id": name}}
    app.invoke({"decision": ""}, cfg)
    app.invoke(Command(resume="yes"), cfg)
    print(f"{name:<34} emails on file: {len(inbox)}")

# 0. the bug: effect, then interrupt
def effect_first(state):
    send_email("dana@example.com", "refund under review")
    return {"decision": interrupt("approve?")}
run_graph("effect before interrupt", effect_first)

# 1. effect after the interrupt: the code above the interrupt has no effects to repeat
def effect_after(state):
    answer = interrupt("approve?")
    send_email("dana@example.com", f"refund {answer}")
    return {"decision": answer}
run_graph("effect after interrupt", effect_after)

# 2. a key that is the same on every replay of this step
def keyed(state):
    send_email("dana@example.com", "refund under review", key="ticket-77:review-notice")
    return {"decision": interrupt("approve?")}
run_graph("effect before interrupt, with a key", keyed)

# 3. the effect in its own node: a finished node is not run again on resume
def run_split():
    inbox.clear()
    def notify(state):
        send_email("dana@example.com", "refund under review")
        return {}
    def approve(state):
        return {"decision": interrupt("approve?")}
    g = StateGraph(State)
    g.add_node("notify", notify); g.add_node("approve", approve)
    g.add_edge(START, "notify"); g.add_edge("notify", "approve"); g.add_edge("approve", END)
    app = g.compile(checkpointer=InMemorySaver())
    cfg = {"configurable": {"thread_id": "split"}}
    app.invoke({"decision": ""}, cfg)
    app.invoke(Command(resume="yes"), cfg)
    print(f"{'effect in its own node':<34} emails on file: {len(inbox)}")
run_split()

# 4. Functional API: the effect inside a @task. A finished task is replayed from the checkpoint.
def run_task(crash_after_send: bool, name: str):
    inbox.clear()
    state = {"first": True}
    @task
    def notify():
        send_email("dana@example.com", "refund under review")
        if crash_after_send and state["first"]:
            state["first"] = False
            raise RuntimeError("process died before the result was recorded")
        return "sent"
    @entrypoint(checkpointer=InMemorySaver())
    def flow(_: str):
        notify().result()
        return interrupt("approve?")
    cfg = {"configurable": {"thread_id": name}}
    try:
        flow.invoke("go", cfg)
    except RuntimeError:
        pass
    flow.invoke(None, cfg) if crash_after_send else None
    if not crash_after_send:
        flow.invoke(Command(resume="yes"), cfg)
    print(f"{name:<34} emails on file: {len(inbox)}")
run_task(False, "effect in a @task")
run_task(True, "@task, died before its result")