A Version Check Won't Recover a Lost Response

Both attempts used version 1. The first changed the row to version 2. Its response never reached the client.

The retry now fails the version check.

That is a useful refusal: the old request cannot overwrite another edit. But it doesn't answer the client's question: did this operation succeed? A conflict alone cannot identify which request produced the newer version.

The previous example put an expected version inside the database write. This follow-up adds a saved result for one intended operation. It keeps the version check for new operations and recovers the original result for recognized retries.

The example is a local Python/SQLite handler with 12 tests. It does not implement an MCP transport or describe SMTasker's production code.

Consider a hypothetical tool that changes an automation's interval:

client saves request_id = edit-42
client sends: id=1, expected_version=1, interval_minutes=30
server commits: interval=30, version=2
response is lost
client resends the same request_id and the same arguments

The second attempt should recover the result of edit-42. It should not create a new intent, repeat the write, or quietly change expected_version to whatever the database contains now.

An expected version and a request identifier solve different problems:

Neither value grants permission. The handler still checks the trusted principal and the resource's workspace.

The handler stores a small receipt alongside the row change:

(workspace, principal, tool revision, request_id)
    -> validated arguments + committed result

The caller creates and persists the request ID before the first send. Retries reuse it. Two genuinely separate edits get different IDs, even if their arguments happen to match.

The service binds that key to a narrow, validated payload. Reusing a key with a different target, expected version or interval is an error. Sorting these three integer fields is sufficient for this example; it is not a general JSON normalization scheme.

For an authorized, accessible resource, the transaction does this:

The receipt lookup comes before the version condition. A successful original request has already advanced the version, so testing the old version first would reject the very retry we want to recognize.

Current access checks come before the receipt lookup. Moving the record to another workspace or deleting it causes a refusal, even if an old receipt remains. This policy prioritizes current access over replay availability. A real service should choose and document its own equivalent policy.

Save the following as retry_receipt.py and run:

python retry_receipt.py

It requires Python 3.12+ and SQLite 3.35+ for RETURNING. It uses the standard library and temporary databases beneath a local test-db directory. The interval range is an illustrative domain rule, not a product limit.

Principal represents trusted request context. Authentication code must supply the actual caller, workspace and current permission decision. This example does not validate tokens, resolve project membership, or implement revocation propagation.

"""Local SQLite demonstration, not an MCP server or a product implementation.

Run with Python 3.12+ and SQLite 3.35+: python retry_receipt.py
Only temporary databases beneath this file's test-db directory are used.
"""
from concurrent.futures import ThreadPoolExecutor
from contextlib import closing
from dataclasses import dataclass, replace
import json
from pathlib import Path
import re
import sqlite3
import sys
import tempfile
from threading import Barrier
import unittest

TOOL = "set_interval.v1"

@dataclass(frozen=True)
class Principal:
    user_id: str
    workspace_id: str
    scopes: frozenset[str]

@dataclass(frozen=True)
class Reply:
    result: dict
    replayed: bool

class Refused(Exception):
    pass

class KeyReuse(Exception):
    pass

def connect(path):
    db = sqlite3.connect(path, timeout=5, autocommit=True)
    db.row_factory = sqlite3.Row
    return db

def one(db, sql, params=()):
    cursor = db.execute(sql, params)
    try:
        return cursor.fetchone()
    finally:
        cursor.close()

def set_interval(path, principal, request_id, automation_id,
                 expected_version, interval_minutes):
    if "automation:write" not in principal.scopes:
        raise Refused("write not permitted")
    if not isinstance(request_id, str) or not re.fullmatch(
            r"[A-Za-z0-9_-]{1,64}", request_id):
        raise ValueError("request_id must be 1-64 letters, digits, _ or -")
    if any(type(v) is not int or v < 1
           for v in (automation_id, expected_version)):
        raise ValueError("id and version must be positive integers")
    if type(interval_minutes) is not int or not 5 <= interval_minutes <= 1440:
        raise ValueError("illustrative interval range is 5-1440 minutes")

    payload = json.dumps({"id": automation_id, "version": expected_version,
                          "interval": interval_minutes},
                         sort_keys=True, separators=(",", ":"))
    key = (principal.workspace_id, principal.user_id, TOOL, request_id)
    with closing(connect(path)) as db:
        try:
            db.execute("BEGIN IMMEDIATE")
            if one(db, "SELECT 1 FROM automations WHERE id=? AND workspace_id=?",
                   (automation_id, principal.workspace_id)) is None:
                raise Refused("resource unavailable")
            old = one(db, """SELECT payload, response FROM receipts
                WHERE workspace_id=? AND user_id=? AND tool=? AND request_id=?""",
                      key)
            if old is not None:
                if old["payload"] != payload:
                    raise KeyReuse("request_id reused with different intent")
                result = json.loads(old["response"])
                db.execute("COMMIT")
                return Reply(result, replayed=True)

            row = one(db, """UPDATE automations
                SET interval_minutes=?, version=version+1
                WHERE id=? AND workspace_id=? AND version=?
                RETURNING id, interval_minutes, version""",
                      (interval_minutes, automation_id,
                       principal.workspace_id, expected_version))
            if row is None:
                raise Refused("version conflict; reload before a new intent")
            result = {"id": row["id"], "interval_minutes": row["interval_minutes"],
                      "version": row["version"]}
            db.execute("INSERT INTO receipts VALUES(?,?,?,?,?,?)",
                       (*key, payload, json.dumps(result, sort_keys=True)))
            db.execute("COMMIT")
            return Reply(result, replayed=False)
        except BaseException:
            if db.in_transaction:
                db.execute("ROLLBACK")
            raise

SCHEMA = """
CREATE TABLE automations(
    id INTEGER PRIMARY KEY, workspace_id TEXT NOT NULL,
    interval_minutes INTEGER NOT NULL, version INTEGER NOT NULL);
CREATE TABLE receipts(
    workspace_id TEXT NOT NULL, user_id TEXT NOT NULL, tool TEXT NOT NULL,
    request_id TEXT NOT NULL, payload TEXT NOT NULL, response TEXT NOT NULL,
    PRIMARY KEY(workspace_id,user_id,tool,request_id));
INSERT INTO automations VALUES(1,'team-a',15,1),(2,'team-a',15,1),(20,'team-b',15,1);
"""
ALICE = Principal("alice", "team-a", frozenset({"automation:write"}))

class ReceiptTests(unittest.TestCase):
    def setUp(self):
        base = Path(__file__).resolve().parent / "test-db"
        base.mkdir(exist_ok=True)
        self.temp = tempfile.TemporaryDirectory(prefix="receipt-", dir=base)
        assert Path(self.temp.name).resolve().is_relative_to(base.resolve())
        self.addCleanup(self.temp.cleanup)
        self.path = Path(self.temp.name) / "demo.sqlite"
        with closing(connect(self.path)) as db:
            db.executescript(SCHEMA)

    def query(self, sql, params=()):
        with closing(connect(self.path)) as db:
            cursor = db.execute(sql, params)
            try:
                return [tuple(row) for row in cursor.fetchall()]
            finally:
                cursor.close()

    def call(self, key="edit-42", *, who=ALICE, target=1, version=1, minutes=30):
        return set_interval(self.path, who, key, target, version, minutes)

    def test_version_only_control_cannot_recover_its_own_result(self):
        sql = """UPDATE automations SET interval_minutes=30,version=version+1
            WHERE id=1 AND workspace_id='team-a' AND version=1 RETURNING version"""
        self.assertEqual(self.query(sql), [(2,)])
        self.assertEqual(self.query(sql), [])  # same retry now conflicts
        self.assertEqual(self.query("SELECT count(*) FROM receipts"), [(0,)])

    def test_commit_then_lost_response_replays_from_fresh_connection(self):
        def transport():
            self.call()  # handler has committed and closed its connection
            raise TimeoutError("simulated response loss after commit")
        with self.assertRaises(TimeoutError):
            transport()
        reply = self.call()
        self.assertTrue(reply.replayed)
        self.assertEqual(reply.result, {"id": 1, "interval_minutes": 30, "version": 2})
        self.assertEqual(self.query("SELECT interval_minutes,version FROM automations WHERE id=1"), [(30, 2)])
        self.assertEqual(self.query("SELECT count(*) FROM receipts"), [(1,)])

    def test_replay_returns_historical_result_without_undoing_newer_edit(self):
        original = self.call()
        later = self.call("edit-43", version=2, minutes=60)
        retry = self.call()
        self.assertEqual(retry.result, original.result)
        self.assertTrue(retry.replayed)
        self.assertEqual(later.result["version"], 3)
        self.assertEqual(self.query("SELECT interval_minutes,version FROM automations WHERE id=1"), [(60, 3)])

    def test_same_key_cannot_change_target_version_or_value(self):
        self.call()
        for changes in ({"minutes": 60}, {"version": 2}, {"target": 2}):
            with self.subTest(changes=changes), self.assertRaises(KeyReuse):
                self.call(**changes)
        self.assertEqual(self.query("SELECT count(*) FROM receipts"), [(1,)])
        self.assertEqual(self.query("SELECT interval_minutes,version FROM automations WHERE id=2"), [(15, 1)])

    def test_new_key_does_not_bypass_old_version(self):
        self.call()
        with self.assertRaises(Refused):
            self.call("another-key", minutes=60)
        self.assertEqual(self.query("SELECT count(*) FROM receipts"), [(1,)])

    def test_same_text_key_is_scoped_to_principal_and_workspace(self):
        self.call()
        bob = replace(ALICE, user_id="bob")
        other_team = replace(ALICE, workspace_id="team-b")
        self.assertFalse(self.call(who=bob, version=2, minutes=60).replayed)
        self.assertFalse(self.call(who=other_team, target=20).replayed)
        self.assertEqual(self.query("SELECT count(*) FROM receipts"), [(3,)])

    def test_current_scope_is_checked_even_for_a_saved_receipt(self):
        self.call()
        revoked = replace(ALICE, scopes=frozenset())
        with self.assertRaises(Refused):
            self.call(who=revoked)
        self.assertEqual(self.query("SELECT version FROM automations WHERE id=1"), [(2,)])

    def test_moved_or_deleted_resource_does_not_replay_or_resurrect(self):
        self.call()
        self.query("UPDATE automations SET workspace_id='team-b' WHERE id=1")
        with self.assertRaises(Refused):
            self.call()
        self.query("DELETE FROM automations WHERE id=1")
        with self.assertRaises(Refused):
            self.call()
        self.assertEqual(self.query("SELECT count(*) FROM automations WHERE id=1"), [(0,)])
        self.assertEqual(self.query("SELECT count(*) FROM receipts"), [(1,)])

    def test_receipt_failure_rolls_back_the_row_change(self):
        self.query("""CREATE TRIGGER fail_receipt BEFORE INSERT ON receipts
            BEGIN SELECT RAISE(ABORT,'receipt store unavailable'); END""")
        with self.assertRaises(sqlite3.IntegrityError):
            self.call()
        self.assertEqual(self.query("SELECT interval_minutes,version FROM automations WHERE id=1"), [(15, 1)])
        self.assertEqual(self.query("SELECT count(*) FROM receipts"), [(0,)])
        self.query("DROP TRIGGER fail_receipt")
        self.assertFalse(self.call().replayed)

    def test_simultaneous_duplicate_calls_share_one_result(self):
        start = Barrier(2)
        def submit():
            start.wait(timeout=5)
            return self.call()
        with ThreadPoolExecutor(max_workers=2) as pool:
            futures = [pool.submit(submit) for _ in range(2)]
            replies = [f.result(timeout=10) for f in futures]
        self.assertEqual(replies[0].result, replies[1].result)
        self.assertEqual(sorted(r.replayed for r in replies), [False, True])
        self.assertEqual(self.query("SELECT version FROM automations WHERE id=1"), [(2,)])
        self.assertEqual(self.query("SELECT count(*) FROM receipts"), [(1,)])

    def test_simultaneous_key_mismatch_has_one_winner(self):
        start = Barrier(2)
        def submit(minutes):
            start.wait(timeout=5)
            try:
                return self.call(minutes=minutes)
            except KeyReuse:
                return "mismatch"
        with ThreadPoolExecutor(max_workers=2) as pool:
            futures = [pool.submit(submit, m) for m in (30, 60)]
            outcomes = [f.result(timeout=10) for f in futures]
        self.assertEqual(sum(isinstance(x, Reply) for x in outcomes), 1)
        self.assertEqual(outcomes.count("mismatch"), 1)
        winner = next(x for x in outcomes if isinstance(x, Reply))
        self.assertEqual(self.query("SELECT interval_minutes,version FROM automations WHERE id=1"), [(winner.result["interval_minutes"], 2)])
        self.assertEqual(self.query("SELECT count(*) FROM receipts"), [(1,)])

    def test_invalid_or_inaccessible_requests_do_not_write(self):
        for changes in ({"target": True}, {"version": 0}, {"minutes": 1},
                        {"minutes": True}, {"target": "1 OR 1=1"}):
            with self.subTest(changes=changes), self.assertRaises(ValueError):
                self.call(**changes)
        for key in ("", "x" * 65, None, "unsafe key"):
            with self.subTest(key=key), self.assertRaises(ValueError):
                self.call(key)
        for target in (20, 999):
            with self.subTest(target=target), self.assertRaises(Refused):
                self.call(target=target)
        self.assertEqual(self.query("SELECT count(*) FROM receipts"), [(0,)])
        self.assertEqual(self.query("SELECT version FROM automations ORDER BY id"), [(1,), (1,), (1,)])

if __name__ == "__main__":
    print(f"Python {sys.version.split()[0]} / SQLite {sqlite3.sqlite_version}", flush=True)
    unittest.main(verbosity=2)

Local run: Python 3.12.14, SQLite 3.53.1; all 12 tests passed. These are local tests, not a live MCP interoperability or power-loss durability test.

The first test is the control: a version-scoped update succeeds once and conflicts on its own retry. There is no saved result to recover.

The lost-response test calls the real handler, lets it commit and close its connection, then raises a simulated transport timeout. Retrying through a new connection returns the saved version-2 result. The row stays at version 2 and there is one receipt.

The later-edit test is just as important:

edit-42 commits interval=30, version=2
edit-43 commits interval=60, version=3
retry edit-42 returns its original interval=30, version=2 result
the database remains interval=60, version=3

That replay result is historical evidence about an operation, not a fresh resource snapshot. The replayed flag makes the distinction visible. A client must not use the old result to replace newer displayed state without its own version check or a fresh read.

Two barrier-based tests release requests from separate threads; each handler opens its own SQLite connection. Identical requests yield one new result and one replay. Different payloads under the same scoped key yield one accepted request and one mismatch. Which payload wins is deliberately unspecified.

The rollback test installs a trigger that rejects receipt insertion after the update has executed. The handler rolls the transaction back: no row change and no receipt remain. This tests the failure between the two writes, rather than assuming that a successful happy path proves atomicity.

The remaining cases cover key scope, rejected input, a new key with a stale version, removed write scope, and moved or deleted resources. The scope test supplies newly restricted trusted context; it does not prove that a production identity system detects revocation.

SQLite permits one writer at a time. BEGIN IMMEDIATE obtains the write transaction before the lookup, so competing first attempts cannot both pass an unprotected lookup. That is a small-database implementation choice, not a high-throughput design recommendation. Keep this transaction short and handle lock contention explicitly in the surrounding service.

There is no external API call inside the transaction. Saving a receipt and sending an email, publishing a post or charging a card are not automatically one atomic operation. Those workflows need a recoverable handoff and a downstream deduplication or reconciliation strategy. This example makes no end-to-end exactly-once claim.

There is also no receipt expiry policy. Deleting receipts removes the ability to recover their results. In this versioned example, an old version normally still prevents a repeated write, but that is not a substitute for a replay contract. Define retention, backup/restore behavior, stable identities and what clients should do after the supported retry window. Reusing a deleted record ID or resetting versions needs additional protection.

The client has work to do too: persist the operation ID, preserve the original arguments, distinguish a retry from a new user decision, and avoid an unbounded retry loop. A request ID regenerated after every timeout defeats the contract.

BEGIN IMMEDIATE and contention. Which lost response would be hardest to reconcile in your tool integration: changing a setting, starting a job, or calling an external service? Leave the failure sequence in the comments, including what evidence the client still has after the timeout.