Human review, pauses & checkpoints

Pause a traversal for a human decision, decide it from another application, and continue after server or worker restarts. Use checkpoints to reuse completed preparation when the handler starts again. The approval example uses invented fixture data and needs only the SDK and a graph server.

What a pause is

app.request_approval(prompt, operation_key=..., checkpoint=...) sends the request to POST /api/v2/approvals under the task lease. The server stores a pending approval, closes the lease and marks the traverser waiting. The worker stops the handler through its internal _TraversalPaused exception, releases the permit and increments stats.paused. No handler remains blocked while waiting for the reviewer.

A decision queues a continuation for the same traverser ID. The handler restarts from its beginning under a fresh lease, with an increased attempt number. Calling request_approval again with the same operation key returns the stored decision. app.pause is a retained alias for this durable mechanism.

Write replayable handlers around these server rules, covered by the tests below:

• Approval identity is (traverser, operation_key). Reusing the key with a different prompt returns 409 approval_conflict; reconstruct the same prompt on continuation.
• Decisions are approved or rejected. Repeating the same decision is idempotent. Changing it later returns 409 approval_decided.
• Approval continuation adds one to max_attempts, so the pause does not consume the configured failure-retry budget.

Checkpoints make the replay safe

Code before the pause executes again unless you guard it. Save completed preparation with app.save_checkpoint(name, value, expected_revision=None) and retrieve it with app.load_checkpoint(name, default=None). Checkpoints hold JSON on the current traverser, and writes return a revision. expected_revision=0 creates only when absent and detects a competing write. It does not make an earlier external side effect atomic with saving the checkpoint.

The refund example loads a checkpoint, prepares the case only when none exists, saves it, builds a stable prompt from that saved value and requests approval:

from __future__ import annotations

from typing import Dict

from atfgraph import Agent, HandlerError, current_context

app = Agent(node_id="refunds", version="1.0.0", description="Refund desk with a durable approval pause", capacity=2)

AUTO_LIMIT_AUD = 200.0
CASE = "refund case"

FIXTURE_TICKETS: Dict[str, dict] = {
    "QF-4417": {"route": "BNE-SYD", "fare_aud": 612.40, "fare_basis": "flex", "flown": False},
    "NZ-0882": {"route": "AKL-QUN", "fare_aud": 189.00, "fare_basis": "saver", "flown": False},
}


@app.skill()
async def assess(ticket: str) -> dict:
    record = FIXTURE_TICKETS.get(ticket.upper())
    if record is None:
        raise HandlerError(f"unknown ticket {ticket!r}", code="ticket_not_found")
    return {"ticket": ticket.upper(), **record}


@app.traversal(timeout_seconds=3600)
async def request_refund(ticket: str, amount_aud: float, reason: str = "schedule change") -> dict:
    attempt = current_context().attempt
    case = await app.load_checkpoint(CASE)
    if case is None:
        ticket_record = await app.traverse("refunds.assess", ticket=ticket)
        case = {
            "ticket": ticket_record["ticket"],
            "route": ticket_record["route"],
            "fare_aud": ticket_record["fare_aud"],
            "amount_aud": round(float(amount_aud), 2),
            "reason": reason,
            "prepared_on_attempt": attempt,
        }
        await app.save_checkpoint(CASE, case, expected_revision=0)

    if case["amount_aud"] <= AUTO_LIMIT_AUD:
        return {"ticket": case["ticket"], "amount_aud": case["amount_aud"], "status": "approved", "decided_by": "desk-policy", "attempts": attempt}

    decision = await app.request_approval(
        {"question": f"Refund {case['amount_aud']} AUD on {case['ticket']} ({case['route']})?", "case": case, "auto_limit_aud": AUTO_LIMIT_AUD},
        operation_key=f"refund:{case['ticket']}",
        checkpoint=case,
    )
    return {
        "ticket": case["ticket"],
        "amount_aud": case["amount_aud"],
        "status": decision["status"],
        "decided_by": decision.get("decided_by"),
        "response": decision.get("response"),
        "attempts": current_context().attempt,
        "prepared_on_attempt": case["prepared_on_attempt"],
    }

The real-daemon restart test checks a result with two attempts, preparation in attempt 1 and one refunds.assess step. The checkpoint lets continuation reuse that preparation rather than rerun the handler's preparatory branch.

The approval's checkpoint= field is separate from save_checkpoint. It attaches review context to the approval and is returned by list_approvals(). The example uses the same object for both purposes.

The traversal's timeout_seconds=3600 applies to each active attempt. Pausing closes that attempt's lease, so time spent awaiting the reviewer is not charged to its handler timeout.

What the decision looks like

On continuation, request_approval returns an approval record with decision status, reviewer, response, operation key, prompt, checkpoint and IDs. Both approval and rejection resume the handler. Your business logic decides how to handle rejection; the example returns status: "rejected" as a successful execution result.

Not verified: the complete returned field list beyond the tested id, status, traverser_id, operation_key, prompt, checkpoint, decided_by and response. Inspect list_approvals() against a real daemon when relying on additional fields.

Deciding out of band

A separate reviewer program can use Client.list_approvals() to find pending requests, get_approval(id) to inspect one, and decide_approval(id, decision, response=None) to record its answer. The example reviewer uses:

from __future__ import annotations

import asyncio
import json
import os

from atfgraph import Client


async def wait_for_status(client: Client, traverser_id: str, status: str, seconds: float = 30.0) -> None:
    deadline = asyncio.get_running_loop().time() + seconds
    while asyncio.get_running_loop().time() < deadline:
        current = await client.get(traverser_id)
        if current.status == status:
            return
        if current.is_terminal:
            raise SystemExit(f"traverser {traverser_id} reached {current.status} before {status}: {current.error}")
        await asyncio.sleep(0.05)
    raise SystemExit(f"traverser {traverser_id} never reached {status}")


async def main() -> None:
    async with Client(os.getenv("ATFGRAPH_SERVER")) as client:
        handle = await client.submit("refunds.request_refund", {"ticket": "QF-4417", "amount_aud": 480.0, "reason": "cancelled leg"})
        await wait_for_status(client, handle.id, "waiting")
        pending = [a for a in await client.list_approvals() if a["traverser_id"] == handle.id]
        approval = pending[0]
        print("prompt    ", json.dumps(approval["prompt"]["question"]))
        print("checkpoint", json.dumps(approval["checkpoint"], sort_keys=True))
        decided = await client.decide_approval(approval["id"], "approved", {"note": "verified against the disruption log"})
        print("decided   ", decided["status"], "by", decided["decided_by"])
        result = await handle.result(timeout=60)
        print("result", json.dumps(result, sort_keys=True))
        for step in (await handle.path()).items:
            print(f"  {step.target:<26} {step.status}")


if __name__ == "__main__":
    asyncio.run(main())

waiting is not terminal. handle.result(), Client.wait and Client.traverse continue waiting through the pause until the resumed handler finishes. Their client timeout includes human review time.

The CLI reference has no approval-decision command. atfg ps shows waiting runs, while decisions use Client.decide_approval, POST /api/v2/approvals/{id}/decide or the Console's Policies page. The owner or an administrator can decide; a worker-only credential cannot.

With the server running, start the example from the repository root:

python -m atfgraph.worker --module examples.python.features.approvals.agent:app
python -m examples.python.features.approvals.review --ticket QF-4417 --amount 480
python -m examples.python.features.approvals.review --ticket QF-4417 --amount 480 --decision rejected
python -m examples.python.features.approvals.review --ticket NZ-0882 --amount 150

The third invocation is below the desk's review limit and creates no approval. The settlement/rejection test checks that the approval list stays empty for that case.

Surviving a restart

Approval state is stored by the server rather than held in a worker. The runtime-journey restart test stops both while approval is pending, restarts the daemon on the same SQLite database, records a decision with a new client and restarts the worker. The traverser succeeds with its checkpoint intact. Its minimal handler is:

from __future__ import annotations

from atfgraph import Agent

app = Agent(node_id="approval", capacity=1)


@app.skill()
async def reviewed() -> dict:
    saved = await app.load_checkpoint("travel brief")
    if saved is None:
        saved = {"origin": "BNE", "revision": 1}
        await app.save_checkpoint("travel brief", saved, expected_revision=0)
    decision = await app.request_approval("Review this plan", operation_key="review", checkpoint=saved)
    return {"decision": decision["status"], "saved": saved}

Keep a worker serving the original (label, version) while approvals for that version remain pending. Continuations are claimed for that pair. They execute the handler code currently deployed for it, so preserve compatible checkpoint and operation-key handling when changing the implementation.

Checkpoints without a pause

You can also use checkpoints without approval to resume preparation during retries. Read externally with Client.get_checkpoint(traverser_id, name), which returns value and revision or raises NotFoundError. Writes require a task lease; normally use app.save_checkpoint inside the handler.

For external effects, a checkpoint alone cannot guarantee exactly-once delivery across a crash between the effect and its marker. Use the external service's idempotency support or a recovery/reconciliation protocol as well.

A checkpoint belongs to one traverser. Use scoped memory for longer-lived or shared state; it supports the same revision-checked writes.

Handling continuation

Not verified: approval expiry and non-pausing approval requests. The wire accepts expires_in_seconds, but Agent.request_approval and Client.request_approval do not expose it, and the cited tests do not exercise expiry. Client.request_approval(..., pause=False) is accepted by the client but not used by the agent helper. Test these paths against a real daemon before relying on them.

Next

• Understand child steps and retries in Calling other nodes.
• Keep shared state in Sessions and memory.
• Check reviewer permissions in Identity and access.