如何防止過時工具呼叫覆寫較新的變更

Two callers have write access to the same workspace. Both read the same automation. One saves a new interval; the other sends a tool call based on the earlier value.

Every permission check can pass, and the second call can still overwrite a newer decision.

The previous example put workspace ownership inside the database write. That prevents a write from crossing the workspace boundary. This example adds a different condition: the record must still have the version the caller read.

Authorized does not mean current

Consider one record, initially at version 1 with an interval of 60 minutes. The numbers below belong to a local teaching model; they are not product settings.

Step	Caller A	Caller B	Stored state
1	Reads interval 60, version 1		60 / v1
2		Saves interval 30, expecting v1	30 / v2
3	Sends interval 15, still expecting v1		Should remain 30 / v2

A statement that filters by ID and workspace can accept step 3. Ownership has not changed. It is the caller's view of the record that is out of date.

The unsafe control in the runnable example deliberately performs that overwrite. It ends at interval 15, version 3. The test passes because it confirms the unwanted behavior was reproduced.

Compare the version inside the update

The guarded mutation is:

UPDATE automations
SET interval_minutes = ?, version = version + 1
WHERE id = ? AND workspace_id = ? AND version = ?
RETURNING id, workspace_id, interval_minutes, running, version;

The workspace parameter comes from the authenticated principal supplied by trusted server code. The expected version comes from the record the caller previously read through an authorized path.

The expected version is a precondition, not a credential. Knowing version 1 gives a caller no write authority.

In SQLite, an UPDATE affects the rows that satisfy its WHERE condition. If B has already advanced the row to version 2, A's version-1 predicate matches nothing. The version increment belongs to the same mutation, so a later call using version 1 cannot reuse it successfully.

RETURNING gives this example the row directly updated by the statement. No returned row is handled as a refusal, rather than a successful update with an empty payload. The code consumes and closes the cursor before returning.

The same refusal message covers a stale version, an inaccessible record and a missing ID. It does not reveal which case occurred. A real service can choose a more detailed conflict response after an authorized read, while preserving its resource-visibility rules.

Make the stale call deterministic

The fixture uses two connections to a temporary SQLite file. A actually reads version 1 and closes its read cursor. B then writes and commits version 2. Only then does A attempt the stale write.

This controlled ordering reproduces the failure without depending on thread timing. It is not a load test or a claim that two SQLite writers run simultaneously: SQLite serializes writers.

Save the following as versioned_write.py, then run:

python versioned_write.py

It uses the Python standard library. The example requires Python 3.9 or later and SQLite 3.35 or later for RETURNING. Connections use isolation_level=None; there is no open application transaction waiting for a later commit.

The accepted interval range, 5–1440 minutes, is a demonstration constraint. Scope checks and input validation still run before the mutation. Updating an interval does not start the automation.

"""A local SQLite teaching model, not a SMTasker implementation."""
from dataclasses import dataclass
from pathlib import Path
import sqlite3
import sys
import tempfile
import unittest


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


class NotApplied(Exception):
    pass


def set_interval(db, principal, automation_id, minutes, expected_version):
    # Trusted authentication code supplies principal, not the model's arguments.
    if "automation:write" not in principal.scopes:
        raise PermissionError("Write access required.")
    if type(minutes) is not int or not 5 <= minutes <= 1440:
        raise ValueError("minutes must be an integer from 5 to 1440.")
    if type(expected_version) is not int or expected_version < 1:
        raise ValueError("expected_version must be a positive integer.")
    cursor = db.execute(
        """UPDATE automations
           SET interval_minutes = ?, version = version + 1
           WHERE id = ? AND workspace_id = ? AND version = ?
           RETURNING id, workspace_id, interval_minutes, running, version""",
        (minutes, automation_id, principal.workspace_id, expected_version),
    )
    try:
        rows = cursor.fetchall()
    finally:
        cursor.close()
    if not rows:
        # Do not reveal whether an inaccessible record exists.
        raise NotApplied("Update not applied. Reload an accessible record.")
    return dict(rows[0])


class VersionedWriteTests(unittest.TestCase):
    def setUp(self):
        fixture_root = Path(__file__).resolve().parent / "test-db"
        fixture_root.mkdir(exist_ok=True)
        self.tmp = tempfile.TemporaryDirectory(prefix="version-", dir=fixture_root)
        assert Path(self.tmp.name).resolve().is_relative_to(fixture_root.resolve())
        filename = str(Path(self.tmp.name) / "example.sqlite")
        self.a = sqlite3.connect(filename, isolation_level=None)
        self.b = sqlite3.connect(filename, isolation_level=None)
        self.a.row_factory = self.b.row_factory = sqlite3.Row
        self.a.executescript("""
            CREATE TABLE automations (
                id TEXT PRIMARY KEY,
                workspace_id TEXT NOT NULL,
                interval_minutes INTEGER NOT NULL
                    CHECK(interval_minutes BETWEEN 5 AND 1440),
                running INTEGER NOT NULL CHECK(running IN (0, 1)),
                version INTEGER NOT NULL CHECK(version >= 1)
            );
            INSERT INTO automations VALUES ('alpha-job', 'alpha', 60, 0, 1);
            INSERT INTO automations VALUES ('beta-job', 'beta', 90, 0, 1);
        """)
        self.editor = Principal("alpha", frozenset({"automation:write"}))

    def tearDown(self):
        self.a.close()
        self.b.close()
        self.tmp.cleanup()

    def row(self, db, automation_id="alpha-job"):
        cursor = db.execute("SELECT * FROM automations WHERE id = ?", (automation_id,))
        try:
            row = cursor.fetchone()
            return None if row is None else dict(row)
        finally:
            cursor.close()

    def newer_edit_after_a_reads(self):
        seen = self.row(self.a)
        self.assertEqual((seen["interval_minutes"], seen["version"]), (60, 1))
        # A's read cursor is closed before B writes and commits.
        saved = set_interval(self.b, self.editor, "alpha-job", 30, 1)
        self.assertEqual((saved["interval_minutes"], saved["version"]), (30, 2))
        return seen

    def test_matching_version_updates_once_and_preserves_other_state(self):
        beta_before = self.row(self.b, "beta-job")
        result = set_interval(self.a, self.editor, "alpha-job", 20, 1)
        self.assertEqual(result, {
            "id": "alpha-job", "workspace_id": "alpha",
            "interval_minutes": 20, "running": 0, "version": 2,
        })
        self.assertEqual(self.row(self.b), result)
        self.assertEqual(self.row(self.b, "beta-job"), beta_before)

    def test_unsafe_control_overwrites_the_newer_edit(self):
        self.newer_edit_after_a_reads()
        self.a.execute("""UPDATE automations
            SET interval_minutes = ?, version = version + 1
            WHERE id = ? AND workspace_id = ?""", (15, "alpha-job", "alpha")).close()
        row = self.row(self.b)
        self.assertEqual((row["interval_minutes"], row["version"]), (15, 3))

    def test_stale_version_refuses_the_same_overwrite(self):
        seen = self.newer_edit_after_a_reads()
        newer = self.row(self.b)
        with self.assertRaises(NotApplied):
            set_interval(self.a, self.editor, "alpha-job", 15, seen["version"])
        self.assertEqual(self.row(self.b), newer)

    def test_replaying_a_successful_request_is_a_conflict_not_a_receipt(self):
        saved = set_interval(self.a, self.editor, "alpha-job", 20, 1)
        with self.assertRaises(NotApplied):
            set_interval(self.b, self.editor, "alpha-job", 20, 1)
        self.assertEqual(self.row(self.b), saved)

    def test_read_scope_cannot_write_even_with_the_current_version(self):
        before = self.row(self.b)
        reader = Principal("alpha", frozenset({"automation:read"}))
        with self.assertRaises(PermissionError):
            set_interval(self.a, reader, "alpha-job", 20, 1)
        self.assertEqual(self.row(self.b), before)

    def test_foreign_and_missing_rows_use_the_same_rejection(self):
        before = self.row(self.b, "beta-job")
        for automation_id in ("beta-job", "missing"):
            with self.subTest(automation_id=automation_id):
                with self.assertRaisesRegex(NotApplied, "^Update not applied"):
                    set_interval(self.a, self.editor, automation_id, 20, 1)
        self.assertEqual(self.row(self.b, "beta-job"), before)

    def test_workspace_transfer_is_still_refused(self):
        seen = self.row(self.a)
        self.b.execute("""UPDATE automations SET workspace_id = 'beta'
                          WHERE id = 'alpha-job'""").close()
        before = self.row(self.b)
        with self.assertRaises(NotApplied):
            set_interval(self.a, self.editor, "alpha-job", 20, seen["version"])
        self.assertEqual(self.row(self.b), before)

    def test_invalid_inputs_leave_the_row_unchanged(self):
        before = self.row(self.b)
        cases = [(value, 1) for value in (True, "20", 4, 1441)]
        cases += [(20, value) for value in (True, "1", 0, -1)]
        for minutes, version in cases:
            with self.subTest(minutes=minutes, version=version):
                with self.assertRaises(ValueError):
                    set_interval(self.a, self.editor, "alpha-job", minutes, version)
        self.assertEqual(self.row(self.b), before)

    def test_sql_like_id_remains_a_bound_value(self):
        before = (self.row(self.b), self.row(self.b, "beta-job"))
        with self.assertRaises(NotApplied):
            set_interval(self.a, self.editor, "alpha-job' OR 1=1 --", 20, 1)
        self.assertEqual((self.row(self.b), self.row(self.b, "beta-job")), before)


if __name__ == "__main__":
    if sqlite3.sqlite_version_info < (3, 35, 0):
        raise SystemExit("SQLite 3.35 or later is required for RETURNING.")
    print(f"Python {sys.version.split()[0]}; SQLite {sqlite3.sqlite_version}", flush=True)
    unittest.main(verbosity=2)

The nine tests cover the successful update, the unsafe overwrite, the refused stale call, replay of a previously successful call, read-only access, inaccessible or missing records, a workspace transfer, invalid input values and SQL-like text passed as a bound ID. They also check the saved state and fields that should remain unchanged.

A conflict needs a new decision

Do not handle a refused stale call by replacing its expected version with the newest number and silently trying again. That makes the check pass while preserving the stale intent that the check was meant to catch.

For an agent-facing tool, a useful recovery flow is:

Report that the requested update was not applied.
Reload the record through the caller's authorized read path.
Compare the new state with the requested change.
Obtain a fresh decision, including renewed human approval when the workflow requires it.

A transport adapter should map NotApplied into the application's explicit failure response. This file does not implement an MCP transport or its error envelope, authentication middleware, a permission service or SMTasker's internal API.

A version check is not an idempotency receipt

Suppose the first update committed, but its response was lost. Retrying with the original expected version will be refused: the successful write already consumed that version.

That refusal does not establish whether the first request succeeded or someone else changed the record. The replay test makes the distinction visible. If the caller needs to recover an operation's outcome, design an operation identifier and an authorized lookup or receipt separately.

The rule has to cover every writer

The example protects this one update path. For the version to represent the state a decision depended on, other writers that change that state must participate in the same versioning policy. An admin endpoint or background job that changes the interval without advancing the version can make a stale precondition appear current.

Decide what belongs to a revision: all configuration, a smaller independently edited field group, or something else. That choice determines which changes should conflict.

The integer version here also assumes the same record keeps its identity. Deleting it and recreating the same ID with a reset version needs a separate incarnation identifier or a policy that prevents ID reuse. Membership revocation, multi-record transactions and merging two valid edits are outside this example.

Which stale edit would matter most in your tool integration: an interval, an audience source, or a different setting? Leave a small example in the comments, especially if several write paths can change it.