When a LangGraph approval expires before the graph resumes

A reviewer approves a refund, gets pulled into another incident, and the graph resumes forty minutes later. The tool still runs even though your policy allows that approval for only ten minutes. Nothing is wrong with the checkpoint; the missing control is an expiry check at the moment of execution.

This bug is easy to hide in a happy-path demo. The interrupt appears, the reviewer clicks Approve, and Command(resume=...) continues the saved run. A useful test must stretch the time between those events and prove that stale authority cannot cross the tool boundary.

Put the expiry check beside the side effect

LangGraph interrupts solve a workflow problem. A call to interrupt() pauses execution, the configured checkpointer saves graph state, and a later invocation with Command(resume=...) supplies the value returned by that interrupt. The same thread_id points the runtime back to the persisted state. Official LangGraph documentation also warns that the interrupted node starts again from its beginning when resumed, so code before the interrupt can run more than once.

None of those mechanics define how long a human decision remains valid. A durable checkpoint can outlive a browser session, an on-call shift, a release window, or the facts shown to the reviewer. That durability is useful, but it is not authorization. Your application has to decide whether the approval is still acceptable.

Place that decision as close as possible to the irreversible operation. Checking only when the reviewer clicks Approve leaves a time-of-check to time-of-use gap. A worker may sit in a queue after the click. A process may crash after saving the resume value. A deployment may restore an old checkpoint hours later. If the executor does not check the deadline, every delay after the UI response silently extends the approval.

The record also needs to bind more than a Boolean. Store which request was approved, a stable fingerprint of the normalized action, who reviewed it, when it expires, and whether it has already been consumed. The resume payload should carry an opaque ticket ID. Do not accept expires_at, action details, or reviewer identity from the browser as authoritative values. A client can be stale, buggy, or hostile.

The following application boundary is deliberately independent of LangGraph. That makes it possible to test the security decision without a checkpoint, a model call, or a real tool. The mutation that matters is straightforward: remove the deadline comparison and the expired test will execute the fake side effect.

from dataclasses import dataclass
from datetime import datetime
from hashlib import sha256
import json
from typing import Any, Callable, Literal


class AuthorizationDenied(Exception):
    pass


def action_fingerprint(action: dict[str, Any]) -> str:
    canonical = json.dumps(action, sort_keys=True, separators=(",", ":"))
    return sha256(canonical.encode("utf-8")).hexdigest()


@dataclass
class Approval:
    ticket_id: str
    request_id: str
    action_hash: str
    reviewer_id: str
    decision: Literal["approved", "rejected"]
    expires_at: datetime
    used_at: datetime | None = None


class ApprovalStore:
    def __init__(self) -> None:
        self.records: dict[str, Approval] = {}

    def add(self, approval: Approval) -> None:
        self.records[approval.ticket_id] = approval

    def consume(
        self,
        ticket_id: str,
        request_id: str,
        action: dict[str, Any],
        now: datetime,
    ) -> Approval:
        approval = self.records.get(ticket_id)
        if approval is None:
            raise AuthorizationDenied("approval_not_found")
        if approval.request_id != request_id:
            raise AuthorizationDenied("approval_request_mismatch")
        if approval.action_hash != action_fingerprint(action):
            raise AuthorizationDenied("approval_action_mismatch")
        if approval.decision != "approved":
            raise AuthorizationDenied("approval_rejected")
        if approval.used_at is not None:
            raise AuthorizationDenied("approval_already_used")
        if now >= approval.expires_at:
            raise AuthorizationDenied("approval_expired")

        approval.used_at = now
        return approval


def execute_authorized(
    *,
    store: ApprovalStore,
    ticket_id: str,
    request_id: str,
    action: dict[str, Any],
    now: Callable[[], datetime],
    executor: Callable[[dict[str, Any]], None],
) -> None:
    store.consume(ticket_id, request_id, action, now())
    executor(action)

The boundary uses a half-open validity interval: an approval is valid before expires_at and invalid at that instant or later. Writing the comparison explicitly prevents teams in different services from choosing opposite boundary behavior. Use timezone-aware UTC datetimes in production and reject naive values when records are deserialized. The sample keeps that validation outside the listing so the actual authorization branches stay visible.

Consumption protects against replay, but it introduces a transaction question. If the process marks a ticket used and crashes before the external API accepts the operation, a simple retry will be denied. If it calls the external API first and crashes before marking the ticket used, a retry may repeat the operation. A production design usually combines the ticket with an idempotency key understood by the target service, or performs the consume and an outbox write in one database transaction. Expiry testing does not replace retry-safety testing.

Build an oracle that can catch a bypass

A secure-looking fixture is not evidence. The test has to observe something that changes when the guard is broken. For this scenario, the strongest oracle is the fake executor's call list. An expired resume must leave it empty. The denial code is a second oracle because it distinguishes expiry from a missing ticket or an action mismatch. Checking only status == "denied" would allow the wrong branch to pass.

Assertion order carries real information here, so put the strongest oracle physically first. Python stops the test at the first failing assertion, which means whichever assertion runs first is the one that names the defect in the report. If the status check runs first, removing the deadline comparison produces AssertionError: assert 'executed' == 'denied', a message about a state field. Put assert calls == [] first and the same mutation produces a message about a refund that actually left the system. The two failures describe the same bug at very different levels of alarm, and only one of them tells an on-call engineer that money moved.

Wire the boundary into a small graph with one interrupt node and one execution node. The interrupt payload contains display data for the reviewer. The resume value contributes only the ticket ID. The execution node reads the action and request ID from checkpointed state, then asks the server-side store to consume the ticket.

from datetime import datetime
from typing import Any, Callable, NotRequired, TypedDict

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

from approval import ApprovalStore, AuthorizationDenied, execute_authorized


class ReviewState(TypedDict):
    request_id: str
    action: dict[str, Any]
    status: str
    approval_ticket_id: NotRequired[str]
    denial_reason: NotRequired[str]


def build_graph(
    *,
    store: ApprovalStore,
    now: Callable[[], datetime],
    executor: Callable[[dict[str, Any]], None],
):
    def review(state: ReviewState) -> dict[str, str]:
        response = interrupt(
            {
                "request_id": state["request_id"],
                "action": state["action"],
                "question": "Approve this action?",
            }
        )
        return {"approval_ticket_id": str(response["ticket_id"])}

    def execute(state: ReviewState) -> dict[str, str]:
        try:
            execute_authorized(
                store=store,
                ticket_id=state["approval_ticket_id"],
                request_id=state["request_id"],
                action=state["action"],
                now=now,
                executor=executor,
            )
        except AuthorizationDenied as error:
            return {"status": "denied", "denial_reason": str(error)}
        return {"status": "executed"}

    builder = StateGraph(ReviewState)
    builder.add_node("review", review)
    builder.add_node("execute", execute)
    builder.add_edge(START, "review")
    builder.add_edge("review", "execute")
    builder.add_edge("execute", END)
    return builder.compile(checkpointer=InMemorySaver())

Notice what the graph does not do. It does not calculate a deadline from the time of resume. That would grant a fresh window to an old decision. It does not compare a deadline copied into graph state, because a migration or state-editing path could alter that value. It does not treat possession of the thread_id as proof of review. The authoritative record remains in the store owned by the enforcement service.

Now exercise three materially different failures. The first expires a correct ticket. The second keeps time valid but changes the action after review. The third reuses a ticket that already authorized one execution. These cases share a denial result but prove different invariants. If a refactor accidentally changes only the expiry branch, the mismatch and replay cases still tell you that the rest of the gate works.

from datetime import datetime, timedelta, timezone

from langgraph.types import Command

from approval import Approval, ApprovalStore, action_fingerprint
from review_graph import build_graph


def issue(store, *, ticket_id, request_id, action, expires_at):
    store.add(
        Approval(
            ticket_id=ticket_id,
            request_id=request_id,
            action_hash=action_fingerprint(action),
            reviewer_id="reviewer-17",
            decision="approved",
            expires_at=expires_at,
        )
    )


def test_expired_approval_never_reaches_executor():
    clock = [datetime(2026, 8, 4, 9, 0, tzinfo=timezone.utc)]
    calls = []
    store = ApprovalStore()
    action = {"tool": "refund", "order_id": "O-19", "amount_cents": 2500}
    graph = build_graph(store=store, now=lambda: clock[0], executor=calls.append)
    config = {"configurable": {"thread_id": "review-O-19"}}

    paused = graph.invoke(
        {"request_id": "req-19", "action": action, "status": "pending"},
        config=config,
    )
    assert paused["__interrupt__"][0].value["request_id"] == "req-19"

    issue(
        store,
        ticket_id="ticket-19",
        request_id="req-19",
        action=action,
        expires_at=clock[0] + timedelta(minutes=10),
    )
    clock[0] += timedelta(minutes=11)

    result = graph.invoke(Command(resume={"ticket_id": "ticket-19"}), config=config)

    assert calls == []
    assert result["status"] == "denied"
    assert result["denial_reason"] == "approval_expired"


def test_ticket_cannot_authorize_changed_action():
    now = datetime(2026, 8, 4, 9, 0, tzinfo=timezone.utc)
    store = ApprovalStore()
    approved = {"tool": "refund", "order_id": "O-20", "amount_cents": 2500}
    changed = {"tool": "refund", "order_id": "O-20", "amount_cents": 250000}
    issue(
        store,
        ticket_id="ticket-20",
        request_id="req-20",
        action=approved,
        expires_at=now + timedelta(minutes=10),
    )

    try:
        store.consume("ticket-20", "req-20", changed, now)
        raise AssertionError("changed action was authorized")
    except Exception as error:
        assert str(error) == "approval_action_mismatch"


def test_ticket_is_single_use():
    now = datetime(2026, 8, 4, 9, 0, tzinfo=timezone.utc)
    store = ApprovalStore()
    action = {"tool": "close_account", "account_id": "A-7"}
    issue(
        store,
        ticket_id="ticket-21",
        request_id="req-21",
        action=action,
        expires_at=now + timedelta(minutes=10),
    )

    store.consume("ticket-21", "req-21", action, now)

    try:
        store.consume("ticket-21", "req-21", action, now)
        raise AssertionError("ticket replay was authorized")
    except Exception as error:
        assert str(error) == "approval_already_used"

The second and third examples call the authorization boundary directly. That is intentional. They do not need LangGraph to establish whether a hash mismatch or replay is denied, and keeping them at unit level makes the failure fast to localize. The first case crosses the interrupt and resume path because expiry bugs often live in the wiring between the checkpoint and the executor.

One improvement for a production suite is to use pytest.raises(AuthorizationDenied, match="...") instead of the explicit try blocks. The listing uses only ordinary Python control flow so the relationship between the oracle and the call is obvious. Either form can fail when enforcement is removed. What matters is that the test reaches the code under test with a mutated input rather than proving facts about its own fixture.

Add a concurrency case when more than one worker can receive the same resumed job. The in-memory store above is suitable for explaining the contract, but its read and write are not a production concurrency control. Two processes could both observe used_at is None before either writes it. Exercise the real repository with two consumers released from a barrier, then assert that exactly one receives authorization and exactly one target request is accepted. The database operation should use a conditional update, transaction, or equivalent atomic primitive supported by that store. A Python lock in the test would only prove the test serialized its own calls.

Cancellation is a separate mutation worth keeping. Suppose the reviewer approves a refund, then support resolves the complaint and revokes the request before the worker resumes. The ticket may still be inside its time window, yet execution should fail with approval_revoked. Model revocation as server-side state and check it before expiry so the diagnostic names the stronger reason. That test catches systems that reduce authorization to a timestamp comparison and ignore later business events.

Read the evidence in the right order

Start at the external effect, not at the trace label. Check the refund provider, message broker, database outbox, or fake executor and answer one question: did the protected action happen? A graph status can say denied while an earlier node already called the tool. Conversely, a missing success span does not prove safety if tracing was sampled or interrupted. The side-effect system is the first source of truth.

Next inspect the authorization event produced immediately before execution. A useful event has a stable denial code and correlation fields, not the raw resume object. For an expiry failure, record evaluated_at and expires_at in UTC, the ticket ID, request ID, action fingerprint, and thread ID. Do not log account secrets, full prompts, or the complete action payload merely to make debugging convenient.

Then inspect the saved graph state. Confirm that the resume used the same thread_id as the interrupted invocation and that the request ID and action fingerprint match what the reviewer saw. LangGraph uses the thread ID to locate checkpointed state. Using another ID is a checkpoint selection problem, not an expired approval. The distinction matters because renewing a ticket will not repair a missing checkpoint.

The interrupt itself should contain the request that was displayed. In the default invocation style documented by LangGraph, a paused result exposes __interrupt__. Verify that there is one expected interrupt and that its value names the request being tested. If there are multiple pending interrupts, match the right one by its returned interrupt identifier rather than assuming list position across parallel tasks.

When the expiry branch is accidentally removed from consume, the first test above reaches the executor and stops on its opening assertion, which is why that assertion is the call list. A concise pytest run then points straight at the violated effect oracle. File paths and timing text vary by environment, and pytest abbreviates a long left-hand repr before expanding it in the explanation, but the important diagnostic is the nonempty call list:

$ pytest -q tests/test_review_graph.py::test_expired_approval_never_reaches_executor
>       assert calls == []
E       AssertionError: assert [{'amount_cen...l': 'refund'}] == []
E
E         Left contains one more item: {'amount_cents': 2500, 'order_id': 'O-19', 'tool': 'refund'}
E         Use -v to get more diff

1 failed

That message is more useful than “authorization test failed.” It proves the stale ticket crossed the boundary and shows the exact normalized action observed by the fake. If the denial reason assertion fails instead while calls stays empty, execution remained safe but classification or branch ordering is wrong. If the test never reaches resume, inspect the thread ID and checkpointer setup before changing the policy.

Capture the clock source in failure artifacts. A mixed use of datetime.now(), database server time, and a test fixture clock can make expiry cases appear random. The event should say which service evaluated the deadline. Do not “fix” this by adding a grace period until you have proved clock skew. A grace period changes the policy and can mask a worker that is minutes out of sync.

Separate expiry bugs from checkpoint bugs

Several failures look like “resume did not work” in a UI. They require different fixes. An expired approval reaches the intended checkpoint, evaluates a valid stored ticket, and returns approval_expired without calling the tool. A wrong-thread failure does not load the paused state at all. Depending on the graph and supplied input, it may start a new thread, fail because required state is absent, or pause at a fresh interrupt. Do not label that as expiry merely because the original approval is old.

A deployment mismatch is another near-miss. The worker may restore a checkpoint written by code with a different state shape or node layout. Evidence includes deserialization errors, missing state fields, or a graph that stops at an unexpected node before the authorization boundary runs. The approval record can remain perfectly valid. Renewing it changes no part of the broken state.

Time-zone parsing produces a third lookalike. If one service writes a local timestamp without an offset and another interprets it as UTC, a fresh ticket can appear expired by several hours. The tell is arithmetic: evaluated_at and expires_at are far apart in a pattern matching an offset, while the reviewer click and queue timestamps are close. Rejecting naive timestamps on ingestion is safer than teaching every consumer to guess the producer's zone.

Queue delay is not itself a bug. If policy says approval lasts ten minutes and the worker starts at minute eleven, denial is correct even if infrastructure caused the delay. The product decision is whether to ask for review again, improve queue priority, or lengthen the validity window. The test should preserve the existing policy and make the delay visible. It should not quietly extend the deadline to keep an availability metric green.

Approval renewal can create its own misleading trace. A user may submit a fresh decision while an older resume message is already in flight. The worker must bind the new ticket to the same immutable request and action, and the event should identify which ticket it evaluated. Do not update the old record's deadline in place. Keeping separate records preserves the fact that one decision expired and another was made later. Test both delivery orders: old resume followed by new resume, and new resume followed by delayed old resume. Only the fresh, unused ticket may authorize the action.

Also check what the interface does after denial. A front end that displays “completed” because it received an HTTP success envelope can hide a correctly blocked tool. Assert on the domain status returned by the graph and on the absence of the target effect. If your API uses HTTP status codes, document whether an expired business approval is returned as a client error or as a successful workflow response with a denied state. Consistency matters more than forcing one transport convention onto every application.

Clock skew deserves a controlled test of its own. Inject the evaluator's clock and exercise times immediately before, exactly at, and immediately after the deadline. Those are boundary cases, not measurements of production skew. Compare service clocks through operational telemetry before claiming that skew caused an incident. If the database is the authority for both ticket creation and consumption, tests should model database time rather than a web server's clock.

Finally, distinguish an expired approval from a consumed one. A retry after a successful side effect should report approval_already_used, and the target should show the original operation. An expiry denial should show no earlier consume event and no target operation. Collapsing both into “invalid approval” makes support simpler but removes the evidence QA needs to find replay defects.

Roll the guard into an existing graph

Begin in observation mode if your current system has no stored expiry. Add the authorization event and calculate what the decision would have been, but do not block execution yet. Mark it clearly as would_deny so nobody mistakes telemetry for enforcement. Compare those events with queue age, reviewer behavior, and business-critical workflows. This stage reveals old checkpoints and background jobs that were never designed to request a fresh review.

Next issue new approval records with request binding, action fingerprints, explicit UTC deadlines, and single-use state. Keep accepting legacy approvals through a narrow compatibility path with a fixed retirement date. A legacy record should be identifiable in logs. Never synthesize an expiry far in the future, because that converts missing data into permanent authority.

Turn enforcement on for one low-risk tool before applying it to every interrupt. That may sound backward, but a failed control rollout can block urgent remediation jobs. Choose a tool with reversible effects and clear ownership. Run valid, expired, mismatched, rejected, and replayed cases in a staging environment. Verify the target system, not only graph output.

During the migration, checkpoints created by the old graph may resume under the new graph. Decide whether to cancel them, route them to a migration node, or require a new review. The safest general rule is to re-review when the stored action cannot be reconstructed exactly. A data migration that guesses the approved action defeats the binding that the new control is supposed to provide.

Add the fast boundary tests to every change and keep a smaller interrupt integration suite for graph wiring. The CI job below installs the project, runs authorization unit cases first, then the checkpoint cases. It does not call a model or external tool, so a failure points to application logic rather than provider variability.

name: approval-expiry

on:
  pull_request:
    paths:
      - "src/approval.py"
      - "src/review_graph.py"
      - "tests/test_approval.py"
      - "tests/test_review_graph.py"

jobs:
  authorization:
    runs-on: ubuntu-latest
    timeout-minutes: 10
    permissions:
      contents: read
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"
      - run: python -m pip install -r requirements.txt
      - run: python -m pytest -q tests/test_approval.py
      - run: python -m pytest -q tests/test_review_graph.py

Make denial metrics actionable before enforcing broadly. An alert should separate expired, rejected, replayed, missing, and mismatched approvals. Owners need the age of the checkpoint and queue, but they do not need private prompt content. Review a sample of denials manually during rollout. A spike in action mismatches may expose unstable canonicalization rather than an attack.

Define rollback behavior in advance. Disabling all authorization checks is not a safe rollback. Prefer pausing the affected tool, routing it to a manual operator, or restoring the prior executor while continuing to reject expired records. The control and the graph deployment should have separate feature switches with restricted operators and auditable changes.

Know what the safer design costs

Every execution now needs a strongly consistent read, and usually a write, against the approval store. That adds latency and a dependency on the store's availability. Caching positive decisions at the worker weakens expiry and replay guarantees unless the cache participates in the same consistency model. For high-risk tools, a failed authorization store should normally fail closed. That choice can reduce availability during an outage.

Short windows create more review prompts. Reviewers may approve the same action twice after queue delays, and users may abandon a flow that keeps asking. Long windows reduce friction but increase the period in which changed circumstances can make an old decision unsafe. Choose the window by action risk and volatility, not by one global default. Deleting a draft email and transferring funds should not inherit the same lifetime.

Binding to a canonical action adds versioning work. If one release serializes amount_cents as an integer and another serializes a formatted currency string, their hashes differ even when the business action is equal. Define a versioned canonical schema and include its version in the approval record. Test cross-version behavior before deploying a new producer or consumer.

Single-use tickets complicate crash recovery. Idempotency keys, transactional outboxes, and reconciliation jobs are extra engineering, not free safety. Your tests need to cover the point where authorization is consumed but delivery is uncertain. Operators also need a way to see whether the target accepted the operation before asking for another approval.

Detailed audit events consume storage and create privacy risk. An action fingerprint is safer than a raw action, but even identifiers and timestamps can reveal sensitive activity patterns. Set retention deliberately and restrict access. Sampling is a poor fit for security decisions because the missing event is often the one needed during an incident; use complete, minimal events instead.

When an expiring approval is the wrong control

Do not add a human timeout to read-only actions merely because every graph has an interrupt. A low-risk search or formatting tool may need authentication and resource scoping, but repeated approval adds delay without changing the decision. Use the lightest control that matches the consequence.

Avoid expiry as a substitute for action binding. A fresh approval for “manage account” is still too broad if the agent can turn it into “close account.” Narrow the approved operation first. Time limits reduce exposure duration; they do not repair an ambiguous grant.

Do not use wall-clock expiry for a workflow whose authority is tied to a business state transition. A payment may remain approved until the invoice changes, a release may remain approved until the artifact digest changes, and a legal review may remain valid for a named document version. In those cases, invalidate on state or version changes, possibly with a time limit as a second condition.

An offline emergency procedure may not be able to reach the approval store. Do not hide that constraint behind a generous cache. Design a separate break-glass path with stronger authentication, narrow scope, explicit operator intent, and post-event review. Test that ordinary agent traffic cannot select that path.

Finally, do not infer safety from a green interrupt test when another code path can call the tool directly. Inventory every executor entry point, including retries, admin jobs, webhooks, and legacy endpoints. The authorization boundary should be unavoidable. A graph-level test proves the resume route is wired correctly; only boundary tests across all callers prove that stale approval cannot reach the side effect.