Build a Retryable CRM Handoff for an AI Workflow

Use a transactional outbox and stable action IDs to recover CRM handoffs, with a tested local example that exposes duplicate-task and lost-reply failures.

Your AI-assisted app prepares a follow-up task. A person approves it, the app sends it to the CRM, and the request times out. Did the task fail to arrive, or did the CRM create it and lose the reply? Retrying blindly can leave two employees working the same follow-up. Giving up can leave the work unassigned.

This tutorial builds a small handoff that keeps the approved action in a durable outbox and retries it with the same action identifier. Its receiver recognizes a replay and returns the existing task. The useful outcome is a handoff you can inspect and recover, with clear states for pending, sent and blocked work.

Scope: Harbor Service and its repair follow-up are fictional. The example uses two local SQLite databases and a simulated CRM with an explicit idempotency contract. It makes no model or network calls and does not test a real CRM integration. The AI draft is assumed to have already been reviewed; the code starts at recording that approved action. The featured image is AI-generated conceptual artwork.

Keep approval and delivery intent together

The fragile version saves “approved” in one system and then makes an external request. Those are separate writes. A failure between them can leave the application and receiver disagreeing about what happened.

The transactional outbox pattern described by AWS puts the local business change and delivery intent in one database transaction. A worker sends the committed intent later. That solves the local two-write gap, but delivery can still repeat, so duplicate handling belongs at the receiver too.

Here, record_approved_task() writes the approved payload and an outbox row in the same transaction. The payload is then reused unchanged. A test injects a failure between those writes and verifies that neither survives. The delivery function ends its local transaction before contacting the receiver.

Validation and approval remain separate requirements. Validating AI output before it reaches your CRM checks its shape; approval decides whether this exact action should happen. In a real app, trusted server code must authorize the user and bind approval to the stored payload. An LLM must not grant itself permission by returning an approved field.

Identify the business action, not the delivery attempt

The fictional policy permits one initial follow-up task per repair case. The application derives the operation key from that identity:

harbor:create-followup:v1:case-104

The business namespace, action type and case ID stay fixed across retries. The key comes from application-controlled identity, not model-generated prose. Approving the same case and title twice returns the same key. Reusing it with a different title raises a conflict; it does not silently replace the earlier instruction.

This is the distinction in Amazon’s guidance on idempotent APIs: a caller-provided identifier expresses one intended operation. The receiver must atomically associate it with the created resource and return an equivalent result on replay. Reusing an identifier for different intent needs explicit handling.

Do not generate a fresh UUID every time a request times out. Do not bump the v1 prefix to force a blocked request through. The example supports one create action per case, not later reminders or task edits. Those need their own approved action identities and update/cancellation rules. Two genuinely different cases still produce two tasks even if their titles match.

Run the local example

Use Python 3.12 or newer because the code sets the SQLite autocommit option explicitly. The actual test environment was Windows 11, Python 3.14.3 and SQLite 3.50.4. No third-party packages, API keys or cloud account are required.

Save the three complete files below in a new folder. handoff.py contains the local outbox and durable receiver simulation; test_handoff.py exercises failures; demo_handoff.py produces an inspectable lost-reply demonstration.

Complete handoff.py

"""Local teaching fixture: one business, one create-task action, one worker."""
from contextlib import closing, contextmanager
import json
import re
import sqlite3

MAX_ATTEMPTS = 3  # Demonstration policy, not a recommended universal threshold.


class KeyConflict(Exception):
    pass


class TransientDeliveryError(Exception):
    pass


class SimulatedCrash(Exception):
    pass


def connect(path):
    db = sqlite3.connect(path, autocommit=True)
    db.row_factory = sqlite3.Row
    db.execute("PRAGMA foreign_keys = ON")
    return db


@contextmanager
def transaction(path):
    with closing(connect(path)) as db:
        db.execute("BEGIN IMMEDIATE")
        try:
            yield db
            db.execute("COMMIT")
        except BaseException:
            if db.in_transaction:
                db.execute("ROLLBACK")
            raise


def init_app(path):
    with closing(connect(path)) as db:
        db.executescript("""
        CREATE TABLE IF NOT EXISTS approvals (
            operation_key TEXT PRIMARY KEY, payload TEXT NOT NULL);
        CREATE TABLE IF NOT EXISTS outbox (
            operation_key TEXT PRIMARY KEY REFERENCES approvals(operation_key),
            state TEXT NOT NULL CHECK(state IN ('pending','sent','blocked')),
            attempts INTEGER NOT NULL DEFAULT 0,
            remote_id INTEGER, last_error TEXT);
        """)


def record_approved_task(path, case_id, title, *, fail_before_outbox=False):
    # Call only after real server-side authorization and human approval.
    # The fixture itself does not authenticate users or implement an approval UI.
    if not isinstance(case_id, str) or not re.fullmatch(r"[A-Za-z0-9-]{1,80}", case_id):
        raise ValueError("Invalid case identifier")
    if not isinstance(title, str) or not 1 <= len(title.strip()) <= 160:
        raise ValueError("Title must contain 1 to 160 characters")
    key = "harbor:create-followup:v1:" + case_id
    payload = json.dumps({"case_id": case_id, "title": title.strip()},
                         sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    with transaction(path) as db:
        existing = db.execute("SELECT payload FROM approvals WHERE operation_key=?", (key,)).fetchone()
        if existing:
            if existing["payload"] != payload:
                raise KeyConflict("Same action key with different approved content")
            return key
        db.execute("INSERT INTO approvals VALUES (?,?)", (key, payload))
        if fail_before_outbox:
            raise SimulatedCrash("Failure between the two local writes")
        db.execute("INSERT INTO outbox(operation_key,state) VALUES (?,'pending')", (key,))
    return key


class FakeCRM:
    """Durable local receiver simulation; no network or real CRM is contacted."""
    def __init__(self, path):
        self.path = path
        with closing(connect(path)) as db:
            db.execute("""CREATE TABLE IF NOT EXISTS tasks (
                task_id INTEGER PRIMARY KEY,
                operation_key TEXT NOT NULL UNIQUE, payload TEXT NOT NULL)""")

    def create_task(self, key, payload):
        with transaction(self.path) as db:
            old = db.execute("SELECT * FROM tasks WHERE operation_key=?", (key,)).fetchone()
            if old:
                if old["payload"] != payload:
                    raise KeyConflict("Receiver has different content for this key")
                return old["task_id"]
            cur = db.execute("INSERT INTO tasks(operation_key,payload) VALUES (?,?)", (key, payload))
            return cur.lastrowid


def deliver_one(path, receiver, *, crash_after_reply=False):
    # Deliberately single-worker: no claim/lease or multi-worker scheduling.
    with transaction(path) as db:
        row = db.execute("""SELECT o.*,a.payload FROM outbox o
            JOIN approvals a USING(operation_key)
            WHERE state='pending' ORDER BY o.rowid LIMIT 1""").fetchone()
        if row is None:
            return "idle"
        key = row["operation_key"]
        if row["attempts"] >= MAX_ATTEMPTS:
            db.execute("UPDATE outbox SET state='blocked',last_error='Retry budget exhausted' WHERE operation_key=?", (key,))
            return "blocked"
        attempt = row["attempts"] + 1
        db.execute("UPDATE outbox SET attempts=? WHERE operation_key=?", (attempt, key))
    # Never hold the application transaction open across a remote call.
    try:
        remote_id = receiver.create_task(key, row["payload"])
    except (TransientDeliveryError, KeyConflict) as exc:
        state = "blocked" if isinstance(exc, KeyConflict) or attempt >= MAX_ATTEMPTS else "pending"
        with transaction(path) as db:
            db.execute("UPDATE outbox SET state=?,last_error=? WHERE operation_key=?",
                       (state, type(exc).__name__, key))
        return state
    if crash_after_reply:
        raise SimulatedCrash("Receiver succeeded; local acknowledgment not recorded")
    with transaction(path) as db:
        db.execute("UPDATE outbox SET state='sent',remote_id=?,last_error=NULL WHERE operation_key=?",
                   (remote_id, key))
    return "sent"


def snapshot(app_path, crm_path):
    with closing(connect(app_path)) as db:
        approvals = db.execute("SELECT COUNT(*) FROM approvals").fetchone()[0]
        outbox = [dict(r) for r in db.execute("SELECT * FROM outbox ORDER BY rowid")]
    with closing(connect(crm_path)) as db:
        tasks = [dict(r) for r in db.execute("SELECT * FROM tasks ORDER BY task_id")]
    return {"approval_count": approvals, "outbox": outbox, "crm_tasks": tasks}

Complete test_handoff.py

import tempfile
import unittest
from pathlib import Path
from handoff import (FakeCRM, KeyConflict, MAX_ATTEMPTS, SimulatedCrash,
                     TransientDeliveryError, deliver_one, init_app,
                     record_approved_task, snapshot)


class Offline:
    def create_task(self, key, payload):
        raise TransientDeliveryError("Connection failed before acceptance")


class LostReply:
    def __init__(self, crm):
        self.crm = crm

    def create_task(self, key, payload):
        self.crm.create_task(key, payload)
        raise TransientDeliveryError("Reply lost after acceptance")


class HandoffTests(unittest.TestCase):
    def setUp(self):
        # Test-owned temporary databases only; no application data is used.
        self.folder = tempfile.TemporaryDirectory(prefix="algo-handoff-test-")
        self.addCleanup(self.clean_test_folder)
        self.app = Path(self.folder.name) / "app.sqlite"
        self.remote = Path(self.folder.name) / "fake-crm.sqlite"
        init_app(self.app)
        self.crm = FakeCRM(self.remote)

    def clean_test_folder(self):
        target = Path(self.folder.name).resolve()
        if target.parent != Path(tempfile.gettempdir()).resolve() or not target.name.startswith("algo-handoff-test-"):
            raise ValueError("Unexpected test cleanup target")
        self.folder.cleanup()

    def approve(self, case="case-104", title="Call about the approved repair plan"):
        return record_approved_task(self.app, case, title)

    def state(self):
        return snapshot(self.app, self.remote)

    def test_approval_and_outbox_roll_back_together(self):
        with self.assertRaises(SimulatedCrash):
            record_approved_task(self.app, "case-104", "Call customer", fail_before_outbox=True)
        self.assertEqual(self.state(), {"approval_count": 0, "outbox": [], "crm_tasks": []})

    def test_same_approval_is_recorded_once(self):
        self.assertEqual(self.approve(), self.approve())
        self.assertEqual(self.state()["approval_count"], 1)
        self.assertEqual(len(self.state()["outbox"]), 1)

    def test_changed_content_rejected_for_same_action(self):
        self.approve()
        before = self.state()
        with self.assertRaises(KeyConflict):
            self.approve(title="Different approved instruction")
        self.assertEqual(self.state(), before)

    def test_failure_before_acceptance_can_retry(self):
        self.approve()
        self.assertEqual(deliver_one(self.app, Offline()), "pending")
        self.assertEqual(len(self.state()["crm_tasks"]), 0)
        self.assertEqual(deliver_one(self.app, self.crm), "sent")
        self.assertEqual(len(self.state()["crm_tasks"]), 1)

    def test_lost_reply_does_not_create_second_task(self):
        self.approve()
        self.assertEqual(deliver_one(self.app, LostReply(self.crm)), "pending")
        first = self.state()["crm_tasks"][0]["task_id"]
        # A new receiver object uses the existing durable database.
        self.assertEqual(deliver_one(self.app, FakeCRM(self.remote)), "sent")
        self.assertEqual(len(self.state()["crm_tasks"]), 1)
        self.assertEqual(self.state()["outbox"][0]["remote_id"], first)

    def test_crash_after_reply_recovers_same_remote_task(self):
        self.approve()
        with self.assertRaises(SimulatedCrash):
            deliver_one(self.app, self.crm, crash_after_reply=True)
        self.assertEqual(self.state()["outbox"][0]["state"], "pending")
        self.assertEqual(deliver_one(self.app, FakeCRM(self.remote)), "sent")
        self.assertEqual(len(self.state()["crm_tasks"]), 1)

    def test_remote_content_conflict_blocks_delivery(self):
        key = self.approve()
        self.crm.create_task(key, '{"title":"different preexisting receiver content"}')
        self.assertEqual(deliver_one(self.app, self.crm), "blocked")
        self.assertEqual(deliver_one(self.app, self.crm), "idle")
        self.assertEqual(len(self.state()["crm_tasks"]), 1)

    def test_sent_action_is_not_selected_again(self):
        self.approve()
        self.assertEqual(deliver_one(self.app, self.crm), "sent")
        self.approve()
        self.assertEqual(deliver_one(self.app, self.crm), "idle")
        self.assertEqual(self.state()["outbox"][0]["attempts"], 1)

    def test_distinct_cases_remain_distinct_actions(self):
        self.approve(case="case-104")
        self.approve(case="case-105")
        self.assertEqual(deliver_one(self.app, self.crm), "sent")
        self.assertEqual(deliver_one(self.app, self.crm), "sent")
        self.assertEqual(len(self.state()["crm_tasks"]), 2)

    def test_retry_budget_stops_repeated_failures(self):
        self.approve()
        for _ in range(MAX_ATTEMPTS - 1):
            self.assertEqual(deliver_one(self.app, Offline()), "pending")
        self.assertEqual(deliver_one(self.app, Offline()), "blocked")
        self.assertEqual(deliver_one(self.app, self.crm), "idle")
        self.assertEqual(self.state()["outbox"][0]["attempts"], MAX_ATTEMPTS)
        self.assertEqual(len(self.state()["crm_tasks"]), 0)

    def test_invalid_input_never_enqueues(self):
        for case, title in [("", "Call"), ("case/104", "Call"), (None, "Call"),
                            ("case-104", " "), ("case-104", None), ("case-104", "x" * 161)]:
            with self.subTest(case=case), self.assertRaises(ValueError):
                record_approved_task(self.app, case, title)
        self.assertEqual(self.state()["approval_count"], 0)

    def test_crashes_exhaust_budget_without_duplicate_tasks(self):
        self.approve()
        for _ in range(MAX_ATTEMPTS):
            with self.assertRaises(SimulatedCrash):
                deliver_one(self.app, self.crm, crash_after_reply=True)
        self.assertEqual(deliver_one(self.app, self.crm), "blocked")
        self.assertEqual(len(self.state()["crm_tasks"]), 1)
        self.assertIsNone(self.state()["outbox"][0]["remote_id"])


if __name__ == "__main__":
    unittest.main(verbosity=2)

Complete demo_handoff.py

"""Run in a new folder. The databases and result are preserved for inspection."""
from pathlib import Path
import json
from handoff import (FakeCRM, TransientDeliveryError, deliver_one, init_app,
                     record_approved_task, snapshot)

P = Path(__file__).resolve().parent
DEMO = P / "demo-data"
DEMO.mkdir()  # Refuse to overwrite an earlier run.
APP, REMOTE = DEMO / "app.sqlite", DEMO / "fake-crm.sqlite"
init_app(APP)
crm = FakeCRM(REMOTE)
key = record_approved_task(APP, "case-104", "Call about the approved repair plan")


class LoseReply:
    def create_task(self, operation_key, payload):
        crm.create_task(operation_key, payload)
        raise TransientDeliveryError("Injected lost reply")


rows = []
for label, receiver in [("reply_lost", LoseReply()),
                        ("retry_same_action", FakeCRM(REMOTE)),
                        ("run_again", FakeCRM(REMOTE))]:
    result = deliver_one(APP, receiver)
    state = snapshot(APP, REMOTE)
    rows.append({"step": label, "result": result,
                 "crm_task_count": len(state["crm_tasks"]),
                 "attempts": state["outbox"][0]["attempts"],
                 "remote_id": state["outbox"][0]["remote_id"]})
report = {"fictional_business": "Harbor Service", "operation_key": key,
          "network_calls": 0, "steps": rows}
(P / "demo-result.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
print(json.dumps(report, indent=2))

python -X utf8 -m unittest -v test_handoff.py
python -X utf8 demo_handoff.py

The demo creates demo-data/app.sqlite, demo-data/fake-crm.sqlite and demo-result.json. It refuses to reuse an existing demo-data folder. To repeat it, copy the source files to a new empty folder and preserve the earlier results. Tests use their own temporary databases.

The transaction helper uses autocommit=True with explicit SQL BEGIN IMMEDIATE, COMMIT and ROLLBACK. It does not rely on Python’s commit() method, which has no effect in that mode. Values are bound through SQL placeholders and connections are closed explicitly. These choices follow the behavior documented in the Python sqlite3 reference.

Read the lost-reply result

On the first attempt, the fake CRM commits a task and the wrapper raises a simulated delivery error instead of returning its ID. The outbox remains pending. On retry, a new receiver object opens the same database, finds the original key and returns task 1. A third worker call finds no pending work.

Demo step	Worker result	CRM tasks	Attempts recorded	Known remote ID
Reply lost after acceptance	pending	1	1	Unknown locally
Retry the same action	sent	1	2	1
Run worker again	idle	1	2	1

Those are observed fixture results, not CRM reliability statistics. The important assertion is that the task count stays at one while the sender recovers the resource ID. The sender did not infer success from an exception; it retried the same action against a receiver that remembers it.

Twelve tests passed. They cover atomic rollback, repeated approval, changed-content rejection, failure before acceptance, a lost reply after acceptance, an interruption before local acknowledgment, receiver conflict, avoiding reselection of sent work, distinct cases, exhausted retry budgets, invalid input and repeated interruptions after remote success. Faults were injected as exceptions; no real network outage, power loss or OS process termination was tested.

A blocked action may already exist in the CRM

The worker records an attempt before sending. Known transient failures stay pending until the demonstration’s three-attempt budget is exhausted. Conflicting receiver content blocks immediately. Unexpected errors propagate for inspection instead of being classified as harmless retries. There is no automatic retry scheduler in this example; each function call makes at most one delivery attempt.

The most revealing test interrupts the workflow after the receiver succeeds three times. The outbox then blocks with no locally recorded remote ID, while the receiver still has exactly one task. “Blocked” means operator attention is needed; it does not prove that no side effect occurred.

State	Meaning in this example	Operator response
pending	No final local acknowledgment	Use the original key and payload for an eligible retry
sent	The receiver returned a task ID that was recorded locally	Keep the receipt; this does not mean the employee completed the task
blocked	Content conflict or retry budget exhausted	Reconcile by action key before any further delivery

idle is a worker return value, not an outbox state. It means there was no pending row to select; blocked rows can still exist. A dashboard that reports “all done” from an idle worker would hide the work that needs attention.

Check the real receiver before replacing FakeCRM

The receiver in this example stores the unique operation key and task in the same row and transaction. An arbitrary CRM endpoint may not offer equivalent behavior. A local deduplication table wrapped around a non-idempotent create request still has a failure gap: the remote task can exist before your local receipt is saved.

Before adapting this example, verify the exact endpoint’s contract:

It supports an idempotency key, or a server-enforced unique external action ID with atomic upsert behavior. Searching for a task and then creating it is not equivalent.
Retries return or let you retrieve the same resource, and conflicting payloads have documented handling.
The key’s scope and retention window cover your account and recovery period. Once deduplication expires, an ambiguous retry may create another task.
The adapter distinguishes retryable failures from validation, authorization and conflict errors, and observes provider rate limits.

If those guarantees are missing, preserve the ambiguous action for reconciliation instead of promising automatic duplicate prevention. Confirm what the CRM actually contains before issuing another create request.

A production worker also needs delayed retries with backoff and jitter, an attempt or time budget, alerts for old pending and blocked actions, and a reviewed recovery path. This example has one worker and no claim lease, parallel-delivery coordination, cross-action ordering or automatic unblock operation. Approval authentication, provider credentials, real HTTP behavior and backup recovery remain outside its tests. No global “exactly once” claim follows from the local result.

For a business building with AI, the useful boundary is after a draft becomes an approved action. Persist that decision, give it a stable identity, and require the receiving system to recognize a replay. That makes interrupted handoffs explainable instead of turning every timeout into a guess.

Jason Ead publishes AlgoThoughts for people building practical business technology. My Biz Heroes focuses on websites, automation, workflows, AI tools and training. Read more about Jason and the publication.