Autonomous Agent Workflows

Real-World Scenario

An agent runs every night to pay approved supplier invoices. One night the process is killed halfway through. The next night’s run starts over and pays the first three suppliers a second time. Nothing in the agent’s reasoning was wrong: it was simply built for a world where processes never die. Long-running agents need the same engineering as any batch job that moves money, including checkpoints, idempotency, timeouts, approval gates and budgets.

The examples run real processes against a real SQLite “bank”, crash them on purpose, and rerun them. No model or API key is involved, because none of these problems depend on the model. They are in the code around it.

The Invoices and the Bank

Five invoices, and a payment function where every call moves money. Save this as bank.py.

import sqlite3

INVOICES = [("INV-101", "Acme Bolts", 1200), ("INV-102", "Brightline", 860),
            ("INV-103", "Coastal Freight", 4300), ("INV-104", "Delta Paper", 310),
            ("INV-105", "Evergreen IT", 2750)]

db = sqlite3.connect("bank.db")
db.execute("CREATE TABLE IF NOT EXISTS payments (invoice TEXT, supplier TEXT, amount INTEGER)")

def pay(invoice, supplier, amount):
    """Stands in for a payment API: every call moves money."""
    db.execute("INSERT INTO payments VALUES (?, ?, ?)", (invoice, supplier, amount))
    db.commit()

def paid():
    return db.execute("SELECT invoice, COUNT(*) FROM payments GROUP BY invoice").fetchall()

The Failure: Rerunning From the Start

The simplest workflow pays each invoice in turn. The first run crashes at the fourth invoice, and the second run starts over.

import sys
from bank import INVOICES, pay, paid

crash_at = sys.argv[1] if len(sys.argv) > 1 else None
for invoice, supplier, amount in INVOICES:
    if invoice == crash_at:
        sys.exit(f"crashed while processing {invoice}")      # the process dies mid-run
    pay(invoice, supplier, amount)
    print(f"paid {invoice} {supplier} ${amount}")
print("payments per invoice:", paid())

$ rm -f bank.db
$ python naive.py INV-104
paid INV-101 Acme Bolts $1200
paid INV-102 Brightline $860
paid INV-103 Coastal Freight $4300
crashed while processing INV-104
$ python naive.py
paid INV-101 Acme Bolts $1200
paid INV-102 Brightline $860
paid INV-103 Coastal Freight $4300
paid INV-104 Delta Paper $310
paid INV-105 Evergreen IT $2750
payments per invoice: [('INV-101', 2), ('INV-102', 2), ('INV-103', 2), ('INV-104', 1), ('INV-105', 1)]

The second run paid every invoice, including the three already paid, so Acme Bolts, Brightline and Coastal Freight each received two payments. A rerun is the obvious response to a crash, and without state it repeats every side effect.

Checkpoints: Remember What Is Done

Record each completed invoice in a checkpoint file, and skip those on the next run. The script can crash either just before or just after a payment.

import json, os, sys
from bank import INVOICES, pay, paid

STATE = "checkpoint.json"
done = json.load(open(STATE)) if os.path.exists(STATE) else []
crash_at, when = (sys.argv[1:] + [None, None])[:2]

for invoice, supplier, amount in INVOICES:
    if invoice in done:
        print(f"skip {invoice} (done in an earlier run)")
        continue
    if invoice == crash_at and when == "before":
        sys.exit(f"crashed before paying {invoice}")
    pay(invoice, supplier, amount)
    if invoice == crash_at and when == "after":
        sys.exit(f"crashed after paying {invoice}, before saving the checkpoint")
    done.append(invoice)
    json.dump(done, open(STATE, "w"))                   # checkpoint after every step
    print(f"paid {invoice}")
print("payments per invoice:", paid())

$ rm -f bank.db checkpoint.json
$ python checkpointed.py INV-104 before
paid INV-101
paid INV-102
paid INV-103
crashed before paying INV-104
$ python checkpointed.py
skip INV-101 (done in an earlier run)
skip INV-102 (done in an earlier run)
skip INV-103 (done in an earlier run)
paid INV-104
paid INV-105
payments per invoice: [('INV-101', 1), ('INV-102', 1), ('INV-103', 1), ('INV-104', 1), ('INV-105', 1)]

A crash before paying is handled cleanly: the rerun skips the three finished invoices and pays each remaining one once. Now crash in the gap between the payment and the checkpoint save:

$ rm -f bank.db checkpoint.json
$ python checkpointed.py INV-104 after
paid INV-101
paid INV-102
paid INV-103
crashed after paying INV-104, before saving the checkpoint
$ python checkpointed.py
skip INV-101 (done in an earlier run)
skip INV-102 (done in an earlier run)
skip INV-103 (done in an earlier run)
paid INV-104
paid INV-105
payments per invoice: [('INV-101', 1), ('INV-102', 1), ('INV-103', 1), ('INV-104', 2), ('INV-105', 1)]

INV-104 was paid twice. The payment happened, but the process died before recording it, so the checkpoint says it’s still to do. No ordering of “act” and “save” closes this gap, because there are always two separate writes to two separate systems. The fix has to be on the receiving side.

Idempotency: Make the Second Payment a No-Op

Give every side effect a key that identifies it, here the invoice number, and have the receiving system refuse a duplicate.

import json, os, sys
from bank import INVOICES, db, paid

db.execute("CREATE UNIQUE INDEX IF NOT EXISTS one_payment ON payments (invoice)")

def pay_once(invoice, supplier, amount):
    """The payment side refuses a second payment with the same key."""
    cur = db.execute("INSERT OR IGNORE INTO payments VALUES (?, ?, ?)", (invoice, supplier, amount))
    db.commit()
    return "paid" if cur.rowcount else "already paid, skipped"

if __name__ == "__main__":
    STATE = "checkpoint.json"
    done = json.load(open(STATE)) if os.path.exists(STATE) else []
    crash_at = sys.argv[1] if len(sys.argv) > 1 else None
    for invoice, supplier, amount in INVOICES:
        if invoice in done:
            continue
        result = pay_once(invoice, supplier, amount)
        if invoice == crash_at:
            sys.exit(f"crashed after paying {invoice}, before saving the checkpoint")
        done.append(invoice)
        json.dump(done, open(STATE, "w"))
        print(f"{invoice}: {result}")
    print("payments per invoice:", paid())

$ rm -f bank.db checkpoint.json
$ python idempotent.py INV-104
INV-101: paid
INV-102: paid
INV-103: paid
crashed after paying INV-104, before saving the checkpoint
$ python idempotent.py
INV-104: already paid, skipped
INV-105: paid
payments per invoice: [('INV-101', 1), ('INV-102', 1), ('INV-103', 1), ('INV-104', 1), ('INV-105', 1)]

The same crash, after paying and before checkpointing, now ends with exactly one payment per invoice. The rerun tries INV-104 again, and the unique index turns it into “already paid, skipped”. Many payment and messaging APIs accept an idempotency key header for exactly this reason. For systems you own, a unique constraint does the job. The checkpoint still earns its keep, because it saves the rerun from repeating work that has no side effect.

Parallel Calls With a Timeout

The run checks each supplier’s bank details through an external API before paying, in parallel. One supplier’s API hangs.

import time
from concurrent.futures import ThreadPoolExecutor, wait

DELAY = {"Acme Bolts": 0.2, "Brightline": 0.3, "Coastal Freight": 3.0,   # one supplier API hangs
         "Delta Paper": 0.1, "Evergreen IT": 0.2}

def check_supplier(name):
    time.sleep(DELAY[name])
    return f"{name}: bank details verified"

start = time.perf_counter()
pool = ThreadPoolExecutor(max_workers=5)
futures = {pool.submit(check_supplier, name): name for name in DELAY}
finished, pending = wait(futures, timeout=1.0)            # stop waiting after 1 second
pool.shutdown(wait=False, cancel_futures=True)            # don't block on the hung call
for f in sorted(finished, key=futures.get):
    print(f.result())
print("timed out:", [futures[f] for f in pending])
print(f"workflow continued after {time.perf_counter() - start:.1f}s")

$ python timeouts.py
Acme Bolts: bank details verified
Brightline: bank details verified
Delta Paper: bank details verified
Evergreen IT: bank details verified
timed out: ['Coastal Freight']
workflow continued after 1.0s

Four checks finish, and Coastal Freight’s is recorded as timed out; the workflow moves on after 1.0 seconds instead of 3. wait(..., timeout=1.0) stops waiting, and shutdown(wait=False) returns immediately instead of blocking on the hung call, as a with ThreadPoolExecutor() block would. (cancel_futures=True also drops calls still queued, which matters when there are more calls than workers.) The hung thread keeps running until its call returns, and Python can’t kill a thread, so the process itself still can’t exit before then: time the script and it takes 3 seconds overall. That is why the external call needs its own timeout too. The timed-out supplier’s invoice should wait for the next run, not be paid unchecked.

Approval Gates That Survive Restarts

Large payments need a person’s approval, and the person won’t answer within the run. Record the pending decision, pay what doesn’t need it, and pick up approvals next run.

import json, os, sys
from bank import INVOICES
from idempotent import pay_once

LIMIT, FILE = 2000, "approvals.json"
approvals = json.load(open(FILE)) if os.path.exists(FILE) else {}

if sys.argv[1:2] == ["approve"]:                   # a person runs: python gated.py approve INV-103
    approvals[sys.argv[2]] = "approved"
else:
    for invoice, supplier, amount in INVOICES:
        if amount > LIMIT and approvals.get(invoice) != "approved":
            approvals.setdefault(invoice, "waiting")
            print(f"{invoice} ${amount}: waiting for approval")
        else:
            print(f"{invoice} ${amount}: {pay_once(invoice, supplier, amount)}")
json.dump(approvals, open(FILE, "w"))

$ rm -f bank.db approvals.json
$ python gated.py
INV-101 $1200: paid
INV-102 $860: paid
INV-103 $4300: waiting for approval
INV-104 $310: paid
INV-105 $2750: waiting for approval
$ python gated.py approve INV-103
$ python gated.py
INV-101 $1200: already paid, skipped
INV-102 $860: already paid, skipped
INV-103 $4300: paid
INV-104 $310: already paid, skipped
INV-105 $2750: waiting for approval
$ cat approvals.json
{"INV-103": "approved", "INV-105": "waiting"}

The first run pays the three small invoices and parks the two large ones as waiting. A person approves INV-103, and the next run pays it. The small invoices are skipped as already paid, thanks to the idempotency key. INV-105 keeps waiting. The approval state lives in a file, not in the agent’s memory, so it survives any number of restarts, and the file records what was decided. In production, store who approved each item and when, as well.

Practice

1. checkpointed.py writes its checkpoint with a plain json.dump. If the process dies during that write, the next run can’t even load the file. Show the failure, then write the checkpoint atomically.

   Solution

   import json, os
   
   def save(done, path="checkpoint.json"):
       tmp = path + ".tmp"
       with open(tmp, "w") as f:
           json.dump(done, f)
           f.flush()
           os.fsync(f.fileno())                   # on disk before we swap it in
       os.replace(tmp, path)                      # atomic: readers see old or new, never half
   
   open("checkpoint.json", "w").write('["INV-101", "INV-1')    # what a crash mid-write leaves
   try:
       json.load(open("checkpoint.json"))
   except json.JSONDecodeError as e:
       print("plain write, crash mid-way:", e)
   
   save(["INV-101", "INV-102"])
   print("atomic write:", json.load(open("checkpoint.json")))

   $ python sol_atomic.py
   plain write, crash mid-way: Unterminated string starting at: line 1 column 13 (char 12)
   atomic write: ['INV-101', 'INV-102']

   A half-written checkpoint is worse than none: the next run crashes at startup. Writing to a temporary file and swapping it in with os.replace, which is atomic on the same filesystem, means a reader always sees either the complete old checkpoint or the complete new one.

2. Agents that plan their own subtasks can generate work faster than they finish it. Run a loop where every task spawns two more, with a limit of 12 steps and $0.50, checked before each step.

   Solution

   def planner(task):
       """A stand-in model that keeps finding 'one more thing' to research."""
       return [f"{task}.{i}" for i in (1, 2)]
   
   queue, steps, cost = ["research competitors"], 0, 0.0
   MAX_STEPS, MAX_COST, COST_PER_STEP = 12, 0.50, 0.04        # dollars
   while queue:
       if steps >= MAX_STEPS or cost + COST_PER_STEP > MAX_COST:
           print(f"stopped: {steps} steps, ${cost:.2f} spent, {len(queue)} tasks left unexplored")
           break
       task = queue.pop(0)
       steps, cost = steps + 1, cost + COST_PER_STEP
       queue += planner(task)
   print("last task run:", task)

   $ python sol_budget.py
   stopped: 12 steps, $0.48 spent, 13 tasks left unexplored
   last task run: research competitors.2.1.1

   After 12 steps and $0.48 the queue still holds 13 tasks and is still growing, so without the limits this loop never ends. The cost check runs before each step (cost + COST_PER_STEP > MAX_COST), so the run can’t overshoot the budget by one step. The report of what was left undone is as important as the stop: someone has to decide whether the remaining work matters.

3. gated.py only knows “approved” and “waiting”. Add “rejected”, so a rejected invoice is never paid and never asked about again.

   Solution

   import json
   from bank import INVOICES
   from idempotent import pay_once
   
   approvals = {"INV-103": "approved", "INV-105": "rejected"}     # decisions a person recorded
   for invoice, supplier, amount in INVOICES:
       decision = approvals.get(invoice, "not needed" if amount <= 2000 else "waiting")
       if decision in ("approved", "not needed"):
           print(f"{invoice}: {pay_once(invoice, supplier, amount)}")
       else:
           print(f"{invoice}: {decision}, not paid")
   json.dump(approvals, open("approvals.json", "w"))

   $ rm -f bank.db
   $ python sol_reject.py
   INV-101: paid
   INV-102: paid
   INV-103: paid
   INV-104: paid
   INV-105: rejected, not paid

   INV-105 is rejected and stays unpaid, and because the decision is recorded, no later run will park it as waiting again. Anything not explicitly approved is not paid: “waiting” and “rejected” both mean no. Default-deny is the only safe reading of an approval file.