Trading AI Architecture For Trade Breaks: Checkpoints, Retries, A Step Budget And A Human Queue

"""A trade-break workflow that survives a crash: checkpointed steps, idempotent side effects,
and an error handler that sends failures to a person instead of retrying for ever.

The same ideas appear in LangGraph 1.2.18 (26 October 2026), which lets error_handler recover on every
path, and checkpoint 4.3.1, which rejects NUL bytes in namespace labels. Python 3.14 standard library
only (sqlite3). Sample trades only.
"""
from __future__ import annotations

import sqlite3
from collections.abc import Callable

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE steps (break_id TEXT, step TEXT, result TEXT, PRIMARY KEY (break_id, step))")
db.execute("CREATE TABLE human_queue (break_id TEXT PRIMARY KEY, reason TEXT)")
SENT: list[str] = []
CRASH = {"armed": True}


def checkpoint(break_id: str, step: str, fn: Callable[[], str]) -> str:
    row = db.execute("SELECT result FROM steps WHERE break_id=? AND step=?", (break_id, step)).fetchone()
    if row:
        return str(row[0])  # already done before the crash: do not repeat the side effect
    result = fn()
    db.execute("INSERT INTO steps VALUES (?,?,?)", (break_id, step, result))
    return result


def draft_chaser(break_id: str) -> str:
    if CRASH["armed"]:
        CRASH["armed"] = False
        raise ConnectionError("model endpoint timed out")
    return f"Draft chaser for {break_id}: confirm settlement date"


def run(break_id: str, max_attempts: int = 2) -> str:
    if "\x00" in break_id:
        raise ValueError("NUL byte in break id")
    checkpoint(break_id, "load", lambda: "loaded TRD-88302 and custodian record")
    for attempt in range(1, max_attempts + 1):
        try:
            draft = checkpoint(break_id, "draft", lambda: draft_chaser(break_id))
            return checkpoint(break_id, "queue_for_approval", lambda: (SENT.append(draft) or "awaiting ops lead"))
        except ConnectionError as exc:
            if attempt == max_attempts:
                db.execute("INSERT OR IGNORE INTO human_queue VALUES (?,?)", (break_id, str(exc)))
                return "sent to human queue"
    return "unreachable"


if __name__ == "__main__":
    print("run 1:", run("BRK-1001"))
    print("run 2 (resume):", run("BRK-1001"))
    print("chasers queued once:", len(SENT))
    CRASH["armed"] = True
    print("max_attempts=1:", run("BRK-1002", max_attempts=1), db.execute("SELECT * FROM human_queue").fetchall())

Python 3.14.4, standard library only. Run once as shown. Sample trades only.

"""A step budget for agent loops on trade breaks: stop at a fixed limit and hand over to a person.

LangGraph 1.2.17 (24 October 2026) applies an explicit recursion_limit equal to the default, and 1.2.18
streams the last checkpoint of a run stopped by the recursion limit. The same rule applies in any
framework: decide the limit, log the stop, and keep the partial work for the reviewer.
Python 3.14 standard library only. Sample data only.
"""
from __future__ import annotations

from collections.abc import Iterator


class StepBudgetExceeded(RuntimeError):
    pass


def guarded(steps: Iterator[str], limit: int) -> list[str]:
    done: list[str] = []
    for n, step in enumerate(steps, start=1):
        if n > limit:
            raise StepBudgetExceeded(f"stopped after {limit} steps, last: {done[-1]}")
        done.append(step)
    return done


def agent_that_keeps_searching() -> Iterator[str]:
    n = 0
    while True:  # e.g. a model that keeps asking for one more custodian file
        n += 1
        yield f"search custodian file {n}"


if __name__ == "__main__":
    print(guarded(iter(["load break", "match trade", "draft chaser"]), limit=8))
    try:
        guarded(agent_that_keeps_searching(), limit=8)
    except StepBudgetExceeded as exc:
        print("handed to ops lead:", exc)

Python 3.14.4, standard library only. Run once as shown. Sample data only.