智能体说“任务完成”，数据库却没有：如何用可验证提交消除假成功

{
  "ok": true,
  "operation_id": "op_01JXYZ",
  "entity_id": "order-42",
  "previous_state": "pending",
  "current_state": "completed",
  "affected_rows": 1,
  "committed": true,
  "verified": true
}

UPDATE orders
SET status = 'completed', completed_at = CURRENT_TIMESTAMP
WHERE id = :order_id
  AND status = 'pending';

import json
import sqlite3
import uuid

DB_PATH = "agent_demo.db"


def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("DROP TABLE IF EXISTS orders")
        conn.execute("DROP TABLE IF EXISTS operations")
        conn.execute("""
            CREATE TABLE orders (
                id TEXT PRIMARY KEY,
                status TEXT NOT NULL,
                completed_by_operation TEXT
            )
        """)
        conn.execute("""
            CREATE TABLE operations (
                operation_id TEXT PRIMARY KEY,
                entity_id TEXT NOT NULL,
                result_json TEXT NOT NULL
            )
        """)
        conn.execute(
            "INSERT INTO orders(id, status) VALUES (?, ?)",
            ("order-42", "pending"),
        )


def complete_order(order_id, operation_id):
    with sqlite3.connect(DB_PATH) as conn:
        conn.row_factory = sqlite3.Row

        # 同一个操作被重试时，返回第一次执行的结果。
        cached = conn.execute(
            "SELECT result_json FROM operations WHERE operation_id = ?",
            (operation_id,),
        ).fetchone()
        if cached:
            return json.loads(cached["result_json"])

        cursor = conn.execute("""
            UPDATE orders
            SET status = 'completed', completed_by_operation = ?
            WHERE id = ? AND status = 'pending'
        """, (operation_id, order_id))

        affected_rows = cursor.rowcount
        row = conn.execute(
            "SELECT id, status, completed_by_operation FROM orders WHERE id = ?",
            (order_id,),
        ).fetchone()

        verified = (
            affected_rows == 1
            and row is not None
            and row["status"] == "completed"
            and row["completed_by_operation"] == operation_id
        )

        result = {
            "ok": verified,
            "operation_id": operation_id,
            "entity_id": order_id,
            "affected_rows": affected_rows,
            "committed": verified,
            "verified": verified,
            "current_state": row["status"] if row else None,
            "error": None if verified else "State transition was not applied",
        }

        # 操作结果和业务更新处于同一事务中。
        conn.execute(
            "INSERT INTO operations(operation_id, entity_id, result_json) VALUES (?, ?, ?)",
            (operation_id, order_id, json.dumps(result)),
        )
        return result


if __name__ == "__main__":
    init_db()
    operation_id = str(uuid.uuid4())

    first = complete_order("order-42", operation_id)
    retry = complete_order("order-42", operation_id)

    print("first:", json.dumps(first, ensure_ascii=False, indent=2))
    print("retry:", json.dumps(retry, ensure_ascii=False, indent=2))

python verified_agent_write.py

只有当工具结果同时满足以下条件时，才可以向用户声称操作完成：
- ok == true
- committed == true
- verified == true
- affected_rows == 1

如果结果未知、超时或验证失败，不得表述为成功。
应告知用户当前状态未知或操作未完成，并返回 operation_id 供查询。
不得因为工具调用已发出，就推断数据库已提交。