"""第 33 章：真实 SQLite 状态、十进制报价、受控取消与幂等回执。只用标准库。"""

from contextlib import closing, contextmanager
from datetime import datetime, timedelta, timezone
from decimal import Decimal, ROUND_HALF_UP
import json
from pathlib import Path
import re
import sqlite3
from tempfile import TemporaryDirectory
from prompt_contracts import fingerprint, read_json
from retrieval_qa import bm25, chunk_documents

CLOCK = datetime(2026, 10, 20, 10, tzinfo=timezone(timedelta(hours=8)))
NOW = int(CLOCK.timestamp())
FIXTURES = [
    ("B100", "lin", NOW+4*3600, "12.5", "active", 1),
    ("B101", "lin", NOW+30*60, "25", "active", 1),
    ("B102", "lin", NOW+6*3600, "50", "active", 1),
    ("B103", "lin", NOW+5*3600, "25", "cancelled", 2),
    ("B200", "mei", NOW+4*3600, "80", "active", 1),
]


class ToolError(Exception):
    """给调用方的稳定错误码；不直接暴露数据库异常或其他人的数据。"""


def initialize(path):
    if Path(path).exists():
        raise FileExistsError("演示初始化只允许新数据库")
    with closing(sqlite3.connect(path)) as db:
        db.executescript("""
        CREATE TABLE bookings(
          id TEXT PRIMARY KEY, owner TEXT, starts INTEGER, grams TEXT,
          status TEXT, version INTEGER);
        CREATE TABLE proposals(
          id INTEGER PRIMARY KEY, actor TEXT, booking TEXT,
          version INTEGER, expires INTEGER, approved INTEGER DEFAULT 0);
        CREATE TABLE receipts(
          actor TEXT, request_key TEXT, payload TEXT, response TEXT,
          PRIMARY KEY(actor, request_key));
        CREATE TABLE events(
          id INTEGER PRIMARY KEY, actor TEXT, booking TEXT, request_key TEXT);
        """)
        db.executemany("INSERT INTO bookings VALUES(?,?,?,?,?,?)", FIXTURES)
        db.commit()


class Workshop:
    def __init__(self, path, actor="lin", now=NOW):
        # actor 来自已认证会话；now 是可替换的教学时钟，均不接受模型参数。
        self.actor, self.now = actor, now
        self.db = sqlite3.connect(path, isolation_level=None)
        self.db.row_factory = sqlite3.Row
        self.chunks = chunk_documents(read_json("workshop-documents.json"))

    def close(self):
        self.db.close()

    @contextmanager
    def transaction(self):
        self.db.execute("BEGIN IMMEDIATE")
        try:
            yield
            self.db.execute("COMMIT")
        except BaseException:
            if self.db.in_transaction:
                self.db.execute("ROLLBACK")
            raise

    def owned(self, booking_id):
        row = self.db.execute(
            "SELECT * FROM bookings WHERE id=? AND owner=?",
            (booking_id, self.actor)).fetchone()
        if row is None:
            raise ToolError("NOT_FOUND_OR_DENIED")
        return row

    def eligible(self, row):
        if row["status"] == "cancelled":
            raise ToolError("ALREADY_CANCELLED")
        if row["starts"]-self.now < 2*3600:
            raise ToolError("TOO_LATE")

    def booking(self, booking_id):
        row = self.owned(booking_id)
        start = datetime.fromtimestamp(row["starts"], CLOCK.tzinfo).isoformat()
        return dict(kind="booking", id=row["id"], starts=start,
                    estimated_grams=row["grams"], status=row["status"],
                    version=row["version"])

    def quote(self, grams):
        if (not isinstance(grams, str)
                or not re.fullmatch(r"(?:0|[1-9]\d{0,3})(?:\.\d{1,2})?", grams)):
            raise ToolError("BAD_GRAMS")
        weight = Decimal(grams)
        if not 0 < weight <= 1000:
            raise ToolError("BAD_GRAMS")
        amount = max(Decimal("8"), weight*Decimal("0.40"))
        return dict(kind="quote", grams=grams,
                    amount=str(amount.quantize(Decimal(".01"), rounding=ROUND_HALF_UP)),
                    currency="CNY", estimate_only=True, source="price:1",
                    policy_version="2026-09-01")

    def search(self, query):
        if not isinstance(query, str) or not 1 <= len(query.strip()) <= 100:
            raise ToolError("BAD_QUERY")
        hits = [dict(id=c["id"], version=c["version"], text=c["text"])
                for c, score in bm25(query, self.chunks)[:2] if score > 0]
        return dict(kind="search", hits=hits)

    def prepare_cancel(self, booking_id):
        with self.transaction():
            row = self.owned(booking_id)
            self.eligible(row)
            cursor = self.db.execute(
                "INSERT INTO proposals(actor,booking,version,expires) VALUES(?,?,?,?)",
                (self.actor, booking_id, row["version"], self.now+300))
            proposal_id = cursor.lastrowid
        return dict(kind="proposal", proposal=proposal_id, booking=booking_id,
                    version=row["version"], expires_at=self.now+300,
                    effect="取消预约，不扣费", requires_confirmation=True)

    def proposal(self, proposal_id):
        if type(proposal_id) is not int:
            raise ToolError("BAD_PROPOSAL")
        row = self.db.execute(
            "SELECT * FROM proposals WHERE id=? AND actor=?",
            (proposal_id, self.actor)).fetchone()
        if row is None:
            raise ToolError("NOT_FOUND_OR_DENIED")
        return row

    def approve(self, proposal_id, expected_booking):
        """由可信确认界面调用；不在模型工具表中。界面展示并绑定具体预约。"""
        with self.transaction():
            proposal = self.proposal(proposal_id)
            if proposal["booking"] != expected_booking:
                raise ToolError("WRONG_CONFIRMATION_TARGET")
            if self.now >= proposal["expires"]:
                raise ToolError("EXPIRED")
            row = self.owned(proposal["booking"])
            if row["version"] != proposal["version"]:
                raise ToolError("STALE")
            self.eligible(row)
            self.db.execute("UPDATE proposals SET approved=1 WHERE id=?", (proposal_id,))

    def commit_cancel(self, proposal_id, request_key):
        """由应用在确认后调用；同一用户的同一请求键只能对应同一取消提案。"""
        if (type(proposal_id) is not int or not isinstance(request_key, str)
                or not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", request_key)):
            raise ToolError("BAD_REQUEST")
        payload = fingerprint(dict(operation="cancel", proposal=proposal_id))
        with self.transaction():
            saved = self.db.execute(
                "SELECT payload,response FROM receipts WHERE actor=? AND request_key=?",
                (self.actor, request_key)).fetchone()
            if saved:
                if saved["payload"] != payload:
                    raise ToolError("IDEMPOTENCY_CONFLICT")
                return json.loads(saved["response"])
            proposal = self.proposal(proposal_id)
            if not proposal["approved"]:
                raise ToolError("NOT_CONFIRMED")
            if self.now >= proposal["expires"]:
                raise ToolError("EXPIRED")
            row = self.owned(proposal["booking"])
            if row["version"] != proposal["version"]:
                raise ToolError("STALE")
            self.eligible(row)
            changed = self.db.execute(
                "UPDATE bookings SET status='cancelled',version=version+1 "
                "WHERE id=? AND owner=? AND version=?",
                (row["id"], self.actor, row["version"])).rowcount
            if changed != 1:
                raise ToolError("STALE")
            event = self.db.execute(
                "INSERT INTO events(actor,booking,request_key) VALUES(?,?,?)",
                (self.actor, row["id"], request_key)).lastrowid
            response = dict(kind="cancellation", booking=row["id"],
                            status="cancelled", version=row["version"]+1, event=event)
            self.db.execute("INSERT INTO receipts VALUES(?,?,?,?)",
                            (self.actor, request_key, payload,
                             json.dumps(response, ensure_ascii=False, sort_keys=True)))
            return response

    def call(self, tool, arguments):
        """模型入口：精确字段、类型和工具白名单；禁止注入身份、SQL 或确认结果。"""
        fields = {"search": "query", "quote": "grams",
                  "booking": "id", "prepare_cancel": "id"}
        try:
            if not isinstance(tool, str) or tool not in fields:
                raise ToolError("UNKNOWN_TOOL")
            key = fields[tool]
            if not isinstance(arguments, dict) or set(arguments) != {key}:
                raise ToolError("BAD_FIELDS")
            argument = arguments[key]
            if not isinstance(argument, str):
                raise ToolError("BAD_TYPE")
            if key == "id" and not re.fullmatch(r"B[0-9]{3}", argument):
                raise ToolError("BAD_ID")
            result = getattr(self, tool)(argument)
            return dict(ok=True, **result)
        except ToolError as error:
            return dict(ok=False, error=str(error))

    def state(self):
        """仅供本地测试/审计读取完整快照，未暴露给模型。"""
        rows = self.db.execute("SELECT id,status,version FROM bookings ORDER BY id").fetchall()
        return dict(bookings=[dict(row) for row in rows],
                    events=self.db.execute("SELECT count(*) FROM events").fetchone()[0])


def cancel_with_retry(service, proposal, key, lose_first_response=False):
    """模拟回执在提交之后丢失；只重试传输超时，最多两次，始终使用原请求键。"""
    for attempt in range(2):
        try:
            receipt = service.commit_cancel(proposal, key)
            if lose_first_response and attempt == 0:
                raise TimeoutError("服务端已提交，但回执未到达客户端")
            return receipt, attempt+1
        except TimeoutError:
            if attempt == 1:
                raise


def main():
    with TemporaryDirectory() as directory:
        path = Path(directory)/"workshop.sqlite"
        initialize(path)
        service = Workshop(path)
        print("教学时钟：", CLOCK.isoformat())
        print("25 克报价：", service.call("quote", {"grams": "25"})["amount"], "元")
        print("12.5 克报价：", service.call("quote", {"grams": "12.5"})["amount"], "元")
        print("开放时间来源：", service.search("工坊周三几点闭馆")["hits"][0]["id"])
        print("B100 初始状态：", service.booking("B100")["status"])
        print("查询他人的预约：", service.call("booking", {"id": "B200"})["error"])
        print("不足两小时：", service.call("prepare_cancel", {"id": "B101"})["error"])
        proposal = service.call("prepare_cancel", {"id": "B100"})["proposal"]
        try:
            service.commit_cancel(proposal, "cancel-B100-1")
        except ToolError as error:
            print("未确认就提交：", error)
        service.approve(proposal, expected_booking="B100")  # 模拟人点击明确的确认按钮
        receipt, attempts = cancel_with_retry(
            service, proposal, "cancel-B100-1", lose_first_response=True)
        print("丢失首次回执后的请求次数：", attempts)
        print("最终状态：", receipt["status"], "；取消事件数：", service.state()["events"])
        service.close()
        restored = Workshop(path, now=NOW+600)
        again = restored.commit_cancel(proposal, "cancel-B100-1")
        print("重开数据库、提案过期后，原请求仍返回原回执：", again == receipt)
        restored.close()


if __name__ == "__main__":
    main()

1. 对 35 克的计算正确，但任务参数错了。执行正确不代表意图理解正确。
2. 分别为 8.00、8.00、8.01 元。$20.02\times0.40=8.008$，按本例规则舍入到分。
3. 身份是服务器已经验证的信息，不能让模型通过参数改写。编号格式正确也不代表有读取权限。
4. 剩余时间为 7199 秒，返回 TOO_LATE，不修改预约。准备阶段的检查不能替代提交检查。
5. 拒绝，返回 IDEMPOTENCY_CONFLICT。新的业务意图应有新的键，并重新满足确认条件。
6. 可能已经修改预约，却没有留下能识别重试的回执；也可能留下回执，却没有完成相应业务变化。
7. 返回的是已经发生的结果，不是再次执行过期操作。若之前没有成功回执，就必须执行当前有效期检查。
8. 三类变化应一起回滚。仓库 scripts/test_tool_workflows.py 使用 SQLite 触发器模拟回执写入失败；移除故障后才能成功提交，最终仍只有一条取消事件。