前言

用户让助手整理会议文字，接着要求核对已有任务并保存清单。查询只读取记录，保存却需要先确认字段是否完整、用户批准的是哪份内容，以及服务最终是否写入。负责人还没有确定就提交，业务系统会收到不完整的安排；写入后没有收到响应，如果另建一份清单重试，又可能生成重复记录。服务需要保留可查询的操作状态，让调用方依据实际结果继续处理。

当前服务先保存提案，让用户在结果页核对并批准，再提交这份内容。创建提案时就返回操作标识，因此提交响应丢失后仍有对象可以查询。我们按缺少字段、补齐提案、批准、提交和结果核对的顺序操作，逐步判断何时能够继续。

一、先检查完整信息，再让服务创建提案

启动方式与查询相同。在 examples/demo-service 中运行普通模式服务，数据库使用 runtime/actions.sqlite，请求记录使用 runtime/http.jsonl。保持服务终端运行，再打开另一个终端发送请求。普通模式不接受故障注入，先用它观察正常的批准与提交，测试故障的历史结果随后单独阅读。

cd "examples/demo-service"
/usr/bin/python3 server.py --port 8765 --database runtime/actions.sqlite --audit runtime/http.jsonl

保存以前先确认会议中已经决定执行的工作。建议和取消项仍然单独保留，未知负责人或日期仍然使用空值。可批准提案要求项目、会议编号、会议日期以及行动的标题、负责人、截止日期和原文完整，missingFields 为空。这个要求来自当前业务选择，便于用户在批准页核对可以直接执行的安排；允许无日期任务的其他系统需要采用自己的真实接口规则。

创建提案前，需要从完整提取结果中选出已经确定的行动。提取结果还包含建议与取消项，而提案接口只接收三个会议字段和行动数组，整份 JSON 直接提交会因额外字段被拒绝。另外两类内容仍留在预览和说明中，方便用户理解会议结论，但不会进入任务系统。这个转换需要保留原来的分类，不能为了让接口接收就把建议改成行动。

行动数组需要包含一至二十项。没有确定行动时，不创建空提案；其他项目需要真实服务支持，不能改一个标识就宣称能够保存。标题、负责人和原文还有类型与长度限制，错误详情指出哪项不合法时，应回到对应输入修正。为了通过长度检查随意截断原文，会失去用户核对的依据，应该保留业务所需内容再决定接口是否适用。

下面的请求故意让结构中的负责人缺失，尽管原文已经提到林舟。它与历史检查第六条请求一致，用于观察服务器是否拒绝不完整结构。这不是一次模型漏提负责人，也不意味着原句没有姓名。服务器不会重新解析 source 补回字段，调用方应该先把提取结果核对完整。

curl -i -H 'Content-Type: application/json' \
  --data-binary @- 'http://127.0.0.1:8765/api/proposals' <<'JSON'
{
  "projectId": "blue-bridge",
  "meetingId": "M-20261004-01",
  "meetingDate": "2026-10-04",
  "actions": [
    {
      "title": "确认教学清单",
      "owner": null,
      "dueDate": "2026-10-08",
      "source": "林舟在八日前确认教学清单。",
      "missingFields": ["owner"]
    }
  ]
}
JSON

这份只读文件画面展示 HTTP 422 和 MISSING_INFORMATION。details 指向第零项行动的 owner 与 missingFields，服务没有创建提案或业务行动。实际技能收到这类返回后，应该回到缺失位置询问用户或修正提取，不继续调用提交。服务可以验证字段是否完整，任务是否真的得到会议决定仍然要由原文规则和用户核对。

字段补齐以后，先查询已有行动，让用户比较可能重复的工作。使用下面的蓝桥未完成查询，检查已保存标题、负责人、日期和会议来源；查询失败时停在这个问题，不把失败当作空清单，也不创建可批准提案。标题相似只能提示比较，当前接口没有跨会议语义去重。后面的幂等控制保护同一提交请求，不会自动判断两场会议的相似任务是否应该合并。

curl -i 'http://127.0.0.1:8765/api/actions?projectId=blue-bridge&status=open'

已有记录核对完成后，把上面请求的 owner 改为林舟，同时把 missingFields 改为空数组，再发送完整请求。不要只清空缺失数组而保留 owner=null，两处必须对应。历史第七条正是补齐后的输入，返回 HTTP 201 和 pending。它说明提案已经形成，业务记录还没有新增。

二、保存本次标识，让用户从结果页批准确切内容

HTTP 201 返回 proposalId、operationId、digest、expiresAt 与 approvalUrl。提案标识用于批准和提交，操作标识用于查询执行状态，摘要用于比较内容是否相同。先保存这些返回，再打开批准页面；如果只保留标题，连接中断以后就无法准确找到这次操作。历史第七条的 P-ea7276fd54134960b472249dc312b309 属于检查数据库，自己复现时使用新返回。

在发送请求的终端依次运行两行 read，第一行等待粘贴本次 proposalId，第二行等待粘贴本次 operationId，每次粘贴后回车。这两个变量供后面的请求使用，不能从旧截图抄入。然后打开浏览器，把本机服务地址与响应中的 approvalUrl 拼在一起，进入这份提案的页面。

read -r proposal_id
read -r operation_id
curl -i "http://127.0.0.1:8765/api/proposals/$proposal_id"

提案 API 可以读到服务保存的完整内容与状态，不返回批准令牌。浏览器结果页显示项目、会议、行动、摘要和有效期，用户在这里核对负责人及日期。当前默认有效期为十分钟，过期以后不能继续批准或提交，需要重新创建提案。刷新页面不会延长有效期，旧页面中的内容也不能自动转成一份新的批准。

expiresAt 包含时间和时区，历史响应使用协调世界时，不能只拿小时数字与电脑本地时间比较。这个字段决定批准还能否使用，与行动的截止日期没有关系。用户看到提案过期时，需要重新生成批准对象，不能修改任务日期来延长提案。会议日期同样保持原来的业务日期，不随批准页面的时间变化。

截图中的提案在当时仍为 pending，没有点击批准或提交。它与前面历史检查中的确认教学清单属于不同提案，只能用来观察页面怎样展示内容。自己打开当次 approvalUrl 时，应确认页面对应刚提交的清单；如果内容不符，先检查提案标识，不能沿用截图里的摘要或批准状态。

内容确认无误后，点击页面的已核对上述确切内容，批准写入。服务器读取这份页面表单的一次性令牌，把提案和操作更新为 approved，随后清除令牌。下面是已经工作的批准代码，状态不是待处理或已经过期时会拒绝，令牌不符时也不能取得批准。

if r['status'] != 'pending' or time.time() >= r['expires_at']:
    raise HTTPError(409, 'NOT_APPROVABLE', '提案已处理或已过期')
if not isinstance(token, str) or not secrets.compare_digest(token, r['approve_token'] or ''):
    raise HTTPError(403, 'APPROVAL_REQUIRED', '请从结果页核对并批准确切内容')
c.execute("UPDATE proposals SET status='approved',approve_token=NULL,approved_at=? WHERE id=?", (now_iso(), pid))
c.execute("UPDATE operations SET status='approved' WHERE id=?", (r['operation_id'],))
c.commit()

批准成功页面会说明批准已经保存，并展示操作标识。随后继续读取提案 API，确认状态为 approved，再提交同一份内容。提交接口不接受客户端传来的 confirmed，增加这个字段会被拒绝；工具也没有获得一个可以代替用户批准的普通 API。用户身份与确切业务批准属于两项不同条件，正式服务还需要把登录会话、项目权限和受保护表单关联起来。

如果用户在核对页面时修改负责人或日期，就需要重新创建提案并再次批准。保存后的提案不接受提交时改写内容，旧摘要和旧批准只能对应原来的输入。虽然这会增加一次确认，却能避免页面显示林舟、数据库最终写入许岚。当前回环页面没有真实用户登录，只提供教学批准过程，历史程序读取表单的结果也不能写成设备用户已经批准。

三、提交同一份提案，并保留同一把幂等键

批准完成以后，再向 POST /api/commits 发送提案标识和幂等键。幂等键用于辨认同一次提交，重试时保留这个值，服务就可以取回原结果。下面的本机操作把本次提案标识同时作为教学键，它的长度符合当前八至一百字符要求，也便于保留。正式调用方可以采用自己的键生成方式，关键是同一提交重试时不要更换。

curl -i -H 'Content-Type: application/json' \
  --data "{\"proposalId\":\"$proposal_id\",\"idempotencyKey\":\"$proposal_id\"}" \
  'http://127.0.0.1:8765/api/commits'

服务返回 HTTP 200 和 committed 以后，核对 operationId、proposalId、digest 与创建时的对应关系，再读取 actionIds。这些行动标识说明这次新增了哪些记录，不能用最终对话中的完成文字替代。历史并发检查使用 key_parallel_01，下面显示其中一份真实返回，操作和提案与历史第七条保持相同。

提交返回参数错误时，先检查终端变量是否仍保存本次标识，以及请求是否只包含两个约定字段。尚未批准时，服务会返回 NOT_APPROVED，应回到本次结果页核对，不能在提交对象里补一个确认开关。提案已经过期时则返回 PROPOSAL_EXPIRED，旧批准不能继续使用。这些响应分别指出格式、批准与时间问题，调用方应该按具体原因停止当前步骤。

同键同参再次到达时，服务器先读取已经保存的结果，返回相同 HTTP 状态和 JSON，不新增另一条行动。同一个键换成不同提案，服务器返回 HTTP 409 和 IDEMPOTENCY_CONFLICT。冲突应该让调用方检查提案与键的关系，不能把它隐藏后继续更换键提交。历史第二十六条就是 key_parallel_01 对应另一份提案的真实拒绝。

服务器先用 BEGIN IMMEDIATE 开始写事务，再读取幂等表，判断当前键是否已有结果。请求摘要对应提案标识，测试故障头不计入业务参数。下面的代码处理同键同参的结果复用，以及同键异参的冲突；这些检查还要与后面的行动写入处在同一事务中，才能共同约束一次提交。

c.execute('BEGIN IMMEDIATE')
prior = c.execute('SELECT * FROM idempotency WHERE key=?', (key,)).fetchone()
if prior:
    if prior['fingerprint'] != fingerprint:
        raise HTTPError(409, 'IDEMPOTENCY_CONFLICT', '同一幂等键不能提交不同提案')
    return prior['http_status'], json.loads(prior['result'])

第一次提交时，服务器核对提案是否批准与过期，插入行动，再保存操作结果、提案状态和幂等结果，最后一起提交。下面是事务末尾的原始代码，前面的行动插入也在当前事务里。进程不能只留下新增行动而漏掉对应幂等结果，否则重启后的请求就无法找到上一次执行。

encoded = canonical(result)
c.execute('UPDATE operations SET status=?,result=? WHERE id=?', (state, encoded, r['operation_id']))
c.execute('UPDATE proposals SET status=?,approve_token=NULL WHERE id=?', (state, pid))
c.execute('INSERT INTO idempotency VALUES(?,?,?,?)', (key, fingerprint, code, encoded))
c.commit()

历史检查同时发出了八个同键同参请求，响应中的操作与行动标识相同，数据库只增加一条行动。同键异参得到冲突，同一提案换新键也因为批准已经消耗而被拒绝。单张响应图只能看到一个返回，八并发与行数结论来自完整检查记录。这个控制覆盖当前 SQLite 和请求约定，不保证不同数据库实例或新提案之间的相似内容去重。

四、提交没有响应时，先查询本次操作

客户端没有收到 HTTP 成功响应时，还不能确定业务结果。操作标识在创建提案时已经取得，可以立即用只读接口查询，不需要先建立另一份提案。下面的命令继续使用当次保存的变量，读取与本次提交对应的操作。

curl -i "http://127.0.0.1:8765/api/operations/$operation_id"

查询 status 和 result，再决定反馈。committed 表示业务已提交，可以读取原来的行动标识；failed 表示服务记录了明确失败；pending 或 approved 说明还没有形成已提交结果。查询也不可达时，当前仍然未知，应该等待服务恢复或查看实际页面。网络报错不能替服务决定数据库有没有变化。

上方是自建服务的提交过程，行动与幂等结果在同一 SQLite 事务中保存。下方的查询箭头表示读取原操作已经保存的状态与结果，不会创建另一份提案；查询不可达时仍需保留结果未知的判断。

读取操作结果时，先确认它返回的提案是否对应这次清单。旧操作也可能合法返回成功，但这不能说明当前输入已经保存。确认提案一致后，再检查该操作的状态与 result；待处理结果为空时，也不能从旧请求补入行动标识。只有这些内容对应同一次提交，返回结果才能回答用户这次请求发生了什么。

历史故障检查使用另一份提案和操作。drop_after_commit 让服务先提交事务，再关闭连接，客户端记录 RemoteDisconnected，HTTP 状态为 null。下面保留了完整请求与客户端结果，故障头只在显式测试模式有效。它显示提交响应丢失，不能解释成服务返回了一个失败状态码。

然后读取同一份操作，历史第四十七条得到 HTTP 200 和 committed。返回的提案仍为 P-d0e5998a231d47f2ba00196fd31668db，操作为 O-f3c4c932e6f640cca4de03ce33fc2992，结果里有已经保存的行动标识。这个操作与普通并发提交那份不同，核对时沿当前标识比较，不能因为两次都是成功状态就混用记录。

同键重新请求取回了原结果，业务行没有重复增加，因此响应丢失后应先查询原操作，避免另建请求造成第二次保存。另一项 fail_before_write 检查则在写入前保存 failed，返回 HTTP 503，业务行数保持不变；同键重试仍返回这份失败。用户决定继续执行时，需要先处理失败原因，再新建提案并重新批准，不能将旧失败记录直接改写为成功。

五、取消与第二个目标都要遵守已经发生的结果

用户在批准以前取消时，点击结果页取消尚未执行的提交，或者由已绑定的取消工具请求对应提案。服务器保存 canceled，批准令牌作废，后续提交被拒绝。已批准但尚未提交的提案也允许取消，因此用户改变决定以后，调用方应该停止提交及后续备忘录保存。下面的命令只针对当前变量中的提案，不会撤销已经新增的行动。

curl -i -H 'Content-Type: application/json' --data '{}' \
  "http://127.0.0.1:8765/api/proposals/$proposal_id/cancel"

取消与提交分别取得数据库写事务。取消先完成时，提交读到作废状态；提交先完成时，取消返回冲突，业务行仍然存在。当前没有撤销行动接口，也没有中断已经执行的远端动作。用户看到取消失败以后，需要知道已保存结果，不能把停止后续步骤说成所有数据已经恢复。

同一提案过期或已取消以后，用户重新发起保存需要新建内容并再次批准。已经提交的操作则先读取结果，不直接把整份清单重新创建成新提案。提案、操作和幂等记录保存在 SQLite，历史停止进程再启动后仍能取回原结果。但技能对话被中断以后如何重新取得变量，尚未在小艺运行环境验证，不能从服务持久化推断平台会自动恢复整条多工具任务。

业务成功以后，用户还可能希望留一份备忘录副本。第二个目标失败时，业务记录仍然已经保存，助手需要分别说明各自结果，再让用户决定是否补做副本。当前没有跨服务事务，端目标失败不自动删除业务行动。真实备忘录超时且没有查询标识时，仍然按照设备核对处理，业务幂等键无法控制另一个插件的重复记录。

总结

当前服务让用户先核对提案，再提交已经批准的内容；历史检查也确认了同键请求、并发提交和响应丢失后的处理。接入调用方需要保留创建时取得的标识，并在异常时查询同一操作。这样，即使一次请求没有收到响应，也仍能找到已经发生的业务结果，再决定是否继续。

取消可以阻止尚未提交的提案，但不能撤回当前服务已经完成的写入。业务记录与副本也要分别反馈，某个目标未知时先核对该目标，不能重新提交已经成功的业务。

完整文件

1. 提案、批准与提交的完整服务

文件路径：examples/demo-service/server.py。

这份文件包含行动查询、提案校验、批准页面、提交事务、幂等结果保存和操作查询，也保留了取消与本地故障测试入口。保存文件以后，可以使用第一节的普通模式命令启动服务；故障注入和独立测试目标只有在显式测试模式下才能使用。服务只监听本机回环地址，没有连接小艺平台或真实备忘录。

#!/usr/bin/env python3
"""Loopback teaching service; no Huawei platform integration is implied."""
import argparse
import datetime as dt
import hashlib
import html
import json
import os
import re
import secrets
import socket
import sqlite3
import threading
import time
import uuid
from contextlib import contextmanager
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlsplit


def canonical(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'))


def now_iso():
    return dt.datetime.now(dt.timezone.utc).isoformat()


class HTTPError(Exception):
    def __init__(self, status, code, message, details=None):
        self.status, self.code, self.message, self.details = status, code, message, details

    def body(self):
        return {'ok': False, 'error': {'code': self.code, 'message': self.message, 'details': self.details}}


class Store:
    def __init__(self, database, ttl):
        self.database = str(database)
        self.copy_database = self.database + '.copies.sqlite'
        self.ttl = ttl
        Path(self.database).parent.mkdir(parents=True, exist_ok=True)
        with self.connect() as c:
            c.executescript('''
            CREATE TABLE IF NOT EXISTS actions(id TEXT PRIMARY KEY,project_id TEXT NOT NULL,meeting_id TEXT NOT NULL,
              title TEXT NOT NULL,owner TEXT NOT NULL,due_date TEXT NOT NULL,source TEXT NOT NULL,status TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS proposals(id TEXT PRIMARY KEY,operation_id TEXT UNIQUE NOT NULL,payload TEXT NOT NULL,
              digest TEXT NOT NULL,status TEXT NOT NULL,expires_at REAL NOT NULL,approve_token TEXT,approved_at TEXT);
            CREATE TABLE IF NOT EXISTS operations(id TEXT PRIMARY KEY,proposal_id TEXT NOT NULL,status TEXT NOT NULL,result TEXT);
            CREATE TABLE IF NOT EXISTS idempotency(key TEXT PRIMARY KEY,fingerprint TEXT NOT NULL,http_status INTEGER NOT NULL,result TEXT NOT NULL);
            ''')
            seeds = [('A-001', 'blue-bridge', 'M-20261001-01', '核对接口字段', '林舟', '2026-10-06', '林舟在六日前核对接口字段。', 'open'),
                     ('A-002', 'blue-bridge', 'M-20261001-01', '整理验收清单', '许岚', '2026-10-07', '许岚在七日前整理验收清单。', 'open'),
                     ('A-003', 'blue-bridge', 'M-20261001-01', '提交页面草图', '陈禾', '2026-10-03', '陈禾提交页面草图。', 'done')]
            c.executemany('INSERT OR IGNORE INTO actions VALUES(?,?,?,?,?,?,?,?)', seeds)
        with self.connect(self.copy_database) as c:
            c.execute('CREATE TABLE IF NOT EXISTS copies(operation_id TEXT PRIMARY KEY,status TEXT NOT NULL,content TEXT,error TEXT)')

    @contextmanager
    def connect(self, database=None):
        c = sqlite3.connect(database or self.database, timeout=10, isolation_level=None)
        c.row_factory = sqlite3.Row
        try:
            c.execute('PRAGMA foreign_keys=ON')
            c.execute('PRAGMA journal_mode=WAL')
            yield c
            c.commit()
        except Exception:
            c.rollback()
            raise
        finally:
            c.close()

    def actions(self, project, status):
        if not re.fullmatch(r'[a-z][a-z0-9-]{0,63}', project):
            raise HTTPError(400, 'INVALID_PROJECT', 'projectId需要使用1到64个小写字母、数字或连字符')
        if status not in ('open', 'done', 'all'):
            raise HTTPError(400, 'INVALID_STATUS', 'status只接受open、done或all')
        with self.connect() as c:
            rows = c.execute('SELECT * FROM actions WHERE project_id=?' + ('' if status == 'all' else ' AND status=?') + ' ORDER BY id',
                             (project,) if status == 'all' else (project, status)).fetchall()
        return {'ok': True, 'projectId': project, 'status': status,
                'items': [{'id': r['id'], 'meetingId': r['meeting_id'], 'title': r['title'], 'owner': r['owner'],
                           'dueDate': r['due_date'], 'source': r['source'], 'status': r['status']} for r in rows]}

    def validate_proposal(self, p):
        expected = {'projectId', 'meetingId', 'meetingDate', 'actions'}
        if not isinstance(p, dict) or set(p) != expected:
            raise HTTPError(400, 'INVALID_BODY', '提案字段需要与接口约定一致')
        if p['projectId'] != 'blue-bridge':
            raise HTTPError(422, 'UNKNOWN_PROJECT', '教学写入仅支持blue-bridge项目')
        if not isinstance(p['meetingId'], str) or not re.fullmatch(r'M-[0-9]{8}-[0-9]{2}', p['meetingId']):
            raise HTTPError(422, 'INVALID_MEETING', 'meetingId格式不合法')
        self.date(p['meetingDate'], 'meetingDate')
        if not isinstance(p['actions'], list) or not 1 <= len(p['actions']) <= 20:
            raise HTTPError(422, 'INVALID_ACTIONS', 'actions需要包含1到20项')
        missing = []
        for i, a in enumerate(p['actions']):
            if not isinstance(a, dict) or set(a) != {'title', 'owner', 'dueDate', 'source', 'missingFields'}:
                raise HTTPError(422, 'INVALID_ACTION', '行动字段需要与约定一致', {'index': i})
            for key, maximum in [('title', 120), ('owner', 40), ('source', 2000)]:
                if a[key] is None or a[key] == '':
                    missing.append({'index': i, 'field': key})
                elif not isinstance(a[key], str) or len(a[key]) > maximum or not a[key].strip():
                    raise HTTPError(422, 'INVALID_FIELD', key + '类型或长度不合法', {'index': i})
            if a['dueDate'] is None or a['dueDate'] == '':
                missing.append({'index': i, 'field': 'dueDate'})
            else:
                self.date(a['dueDate'], 'dueDate')
            if not isinstance(a['missingFields'], list) or a['missingFields']:
                missing.append({'index': i, 'field': 'missingFields'})
        if missing:
            raise HTTPError(422, 'MISSING_INFORMATION', '请补齐行动信息后再创建提案', missing)
        return p

    @staticmethod
    def date(value, field):
        try:
            if not isinstance(value, str) or not re.fullmatch(r'[0-9]{4}-[0-9]{2}-[0-9]{2}', value):
                raise ValueError()
            dt.date.fromisoformat(value)
        except ValueError:
            raise HTTPError(422, 'INVALID_DATE', field + '需要使用真实日期YYYY-MM-DD')

    def create(self, payload):
        self.validate_proposal(payload)
        encoded = canonical(payload)
        digest = hashlib.sha256(encoded.encode()).hexdigest()
        pid, oid = 'P-' + uuid.uuid4().hex, 'O-' + uuid.uuid4().hex
        expires = time.time() + self.ttl
        with self.connect() as c:
            c.execute('BEGIN IMMEDIATE')
            c.execute('INSERT INTO proposals VALUES(?,?,?,?,?,?,?,NULL)',
                      (pid, oid, encoded, digest, 'pending', expires, secrets.token_urlsafe(32)))
            c.execute('INSERT INTO operations VALUES(?,?,?,NULL)', (oid, pid, 'pending'))
            c.commit()
        return {'ok': True, 'proposalId': pid, 'operationId': oid, 'status': 'pending', 'digest': digest,
                'expiresAt': dt.datetime.fromtimestamp(expires, dt.timezone.utc).isoformat(), 'approvalUrl': '/proposals/' + pid}

    def proposal(self, pid):
        with self.connect() as c:
            row = c.execute('SELECT * FROM proposals WHERE id=?', (pid,)).fetchone()
        if not row:
            raise HTTPError(404, 'PROPOSAL_NOT_FOUND', '提案不存在')
        return dict(row)

    def public_proposal(self, pid):
        r = self.proposal(pid)
        status = 'expired' if r['status'] in ('pending', 'approved') and time.time() >= r['expires_at'] else r['status']
        return {'ok': True, 'proposalId': pid, 'operationId': r['operation_id'], 'status': status,
                'digest': r['digest'], 'payload': json.loads(r['payload']),
                'expiresAt': dt.datetime.fromtimestamp(r['expires_at'], dt.timezone.utc).isoformat()}

    def approve(self, pid, token):
        with self.connect() as c:
            c.execute('BEGIN IMMEDIATE')
            r = c.execute('SELECT * FROM proposals WHERE id=?', (pid,)).fetchone()
            if not r:
                raise HTTPError(404, 'PROPOSAL_NOT_FOUND', '提案不存在')
            if r['status'] != 'pending' or time.time() >= r['expires_at']:
                raise HTTPError(409, 'NOT_APPROVABLE', '提案已处理或已过期')
            if not isinstance(token, str) or not secrets.compare_digest(token, r['approve_token'] or ''):
                raise HTTPError(403, 'APPROVAL_REQUIRED', '请从结果页核对并批准确切内容')
            c.execute("UPDATE proposals SET status='approved',approve_token=NULL,approved_at=? WHERE id=?", (now_iso(), pid))
            c.execute("UPDATE operations SET status='approved' WHERE id=?", (r['operation_id'],))
            c.commit()
        return {'ok': True, 'proposalId': pid, 'status': 'approved', 'operationId': r['operation_id']}

    def cancel(self, pid):
        with self.connect() as c:
            c.execute('BEGIN IMMEDIATE')
            r = c.execute('SELECT * FROM proposals WHERE id=?', (pid,)).fetchone()
            if not r:
                raise HTTPError(404, 'PROPOSAL_NOT_FOUND', '提案不存在')
            if r['status'] not in ('pending', 'approved', 'canceled'):
                raise HTTPError(409, 'CANNOT_CANCEL', '当前提案已执行，取消不能撤回业务写入')
            c.execute("UPDATE proposals SET status='canceled',approve_token=NULL WHERE id=?", (pid,))
            c.execute("UPDATE operations SET status='canceled' WHERE id=?", (r['operation_id'],))
            c.commit()
        return {'ok': True, 'proposalId': pid, 'status': 'canceled', 'operationId': r['operation_id']}

    def commit(self, payload, fault=None):
        if not isinstance(payload, dict) or set(payload) != {'proposalId', 'idempotencyKey'}:
            raise HTTPError(400, 'INVALID_BODY', '提交只接受proposalId和idempotencyKey，confirmed不构成批准')
        pid, key = payload['proposalId'], payload['idempotencyKey']
        if not isinstance(pid, str) or not isinstance(key, str) or not re.fullmatch(r'[A-Za-z0-9_-]{8,100}', key):
            raise HTTPError(400, 'INVALID_COMMIT', '提案标识或幂等键不合法')
        fingerprint = hashlib.sha256(canonical({'proposalId': pid}).encode()).hexdigest()
        with self.connect() as c:
            c.execute('BEGIN IMMEDIATE')
            prior = c.execute('SELECT * FROM idempotency WHERE key=?', (key,)).fetchone()
            if prior:
                if prior['fingerprint'] != fingerprint:
                    raise HTTPError(409, 'IDEMPOTENCY_CONFLICT', '同一幂等键不能提交不同提案')
                return prior['http_status'], json.loads(prior['result'])
            r = c.execute('SELECT * FROM proposals WHERE id=?', (pid,)).fetchone()
            if not r:
                raise HTTPError(404, 'PROPOSAL_NOT_FOUND', '提案不存在')
            if time.time() >= r['expires_at'] and r['status'] in ('pending', 'approved'):
                raise HTTPError(409, 'PROPOSAL_EXPIRED', '提案已过期，请重新创建并批准')
            if r['status'] != 'approved':
                raise HTTPError(403 if r['status'] == 'pending' else 409, 'NOT_APPROVED', '提案必须从结果页批准且不能重复消耗')
            if fault == 'fail_before_write':
                result = {'ok': False, 'error': {'code': 'INJECTED_FAILURE', 'message': '测试模式在业务写入前明确失败', 'details': None},
                          'operationId': r['operation_id'], 'status': 'failed', 'proposalId': pid}
                code, state = 503, 'failed'
            else:
                p = json.loads(r['payload'])
                action_ids = []
                for a in p['actions']:
                    aid = 'A-' + uuid.uuid4().hex
                    c.execute('INSERT INTO actions VALUES(?,?,?,?,?,?,?,?)',
                              (aid, p['projectId'], p['meetingId'], a['title'], a['owner'], a['dueDate'], a['source'], 'open'))
                    action_ids.append(aid)
                result = {'ok': True, 'operationId': r['operation_id'], 'status': 'committed', 'actionIds': action_ids,
                          'proposalId': pid, 'digest': r['digest']}
                code, state = 200, 'committed'
            encoded = canonical(result)
            c.execute('UPDATE operations SET status=?,result=? WHERE id=?', (state, encoded, r['operation_id']))
            c.execute('UPDATE proposals SET status=?,approve_token=NULL WHERE id=?', (state, pid))
            c.execute('INSERT INTO idempotency VALUES(?,?,?,?)', (key, fingerprint, code, encoded))
            c.commit()
        return code, result

    def copy_status(self, oid):
        with self.connect(self.copy_database) as c:
            r = c.execute('SELECT status,content,error FROM copies WHERE operation_id=?', (oid,)).fetchone()
        return {'status': r[0], 'content': json.loads(r[1]) if r[1] else None, 'error': r[2], 'source': 'local_test_destination'} if r else {'status': 'not_requested', 'source': 'local_test_destination'}

    def operation(self, oid):
        with self.connect() as c:
            r = c.execute('SELECT o.*,p.expires_at FROM operations o JOIN proposals p ON p.id=o.proposal_id WHERE o.id=?', (oid,)).fetchone()
        if not r:
            raise HTTPError(404, 'OPERATION_NOT_FOUND', '操作不存在')
        status = 'expired' if r['status'] in ('pending', 'approved') and time.time() >= r['expires_at'] else r['status']
        return {'ok': True, 'operationId': oid, 'proposalId': r['proposal_id'], 'status': status,
                'result': json.loads(r['result']) if r['result'] else None, 'localCopy': self.copy_status(oid)}

    def test_copy(self, p):
        if not isinstance(p, dict) or set(p) != {'operationId', 'outcome'} or p['outcome'] not in ('saved', 'failed'):
            raise HTTPError(400, 'INVALID_COPY', '测试目标参数不合法')
        op = self.operation(p['operationId'])
        if op['status'] != 'committed':
            raise HTTPError(409, 'BUSINESS_NOT_COMMITTED', '业务提交完成后才允许请求独立目标')
        proposal = self.public_proposal(op['proposalId'])
        with self.connect(self.copy_database) as c:
            c.execute('INSERT OR REPLACE INTO copies VALUES(?,?,?,?)', (p['operationId'], p['outcome'],
                      canonical(proposal['payload']) if p['outcome'] == 'saved' else None,
                      '测试目标明确拒绝保存' if p['outcome'] == 'failed' else None))
        return {'ok': True, 'operationId': p['operationId'], 'localCopy': self.copy_status(p['operationId'])}


class Handler(BaseHTTPRequestHandler):
    server_version = 'MeetingActionsTeaching/1'

    def log_message(self, *_):
        pass

    def send(self, status, body, content_type='application/json; charset=utf-8'):
        data = canonical(body).encode() if isinstance(body, dict) else body.encode()
        self.send_response(status)
        self.send_header('Content-Type', content_type)
        self.send_header('Content-Length', str(len(data)))
        self.send_header('Cache-Control', 'no-store')
        self.send_header('X-Content-Type-Options', 'nosniff')
        self.send_header('Content-Security-Policy', "default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; frame-ancestors 'none'")
        self.end_headers()
        self.wfile.write(data)
        self.record(status, body)

    def record(self, status, body):
        if self.server.audit:
            event = {'at': now_iso(), 'request': {'method': self.command, 'path': self.path, 'body': getattr(self, 'raw', ''),
                     'testFault': self.headers.get('X-Test-Fault')}, 'response': {'status': status, 'body': body}}
            with self.server.audit_lock:
                with open(self.server.audit, 'a', encoding='utf-8') as f:
                    f.write(canonical(event) + '\n')

    def read_body(self):
        try:
            n = int(self.headers.get('Content-Length', '0'))
        except ValueError:
            raise HTTPError(400, 'INVALID_LENGTH', '请求长度不合法')
        if not 0 <= n <= 65536:
            raise HTTPError(413, 'BODY_TOO_LARGE', '请求不能超过65536字节')
        self.raw = self.rfile.read(n).decode('utf-8')
        if self.headers.get('Content-Type', '').startswith('application/x-www-form-urlencoded'):
            return {k: v[0] for k, v in parse_qs(self.raw).items()}
        try:
            return json.loads(self.raw or '{}')
        except ValueError:
            raise HTTPError(400, 'INVALID_JSON', '请求需要合法JSON')

    def do_GET(self):
        self.dispatch()

    def do_POST(self):
        self.dispatch()

    def dispatch(self):
        try:
            path = urlsplit(self.path).path
            store = self.server.store
            if self.command == 'GET':
                if path == '/health':
                    return self.send(200, {'ok': True, 'testMode': self.server.test_mode})
                if path == '/api/actions':
                    q = parse_qs(urlsplit(self.path).query, keep_blank_values=True)
                    if set(q) - {'projectId', 'status'} or not q.get('projectId') or any(len(v) != 1 for v in q.values()):
                        raise HTTPError(400, 'INVALID_QUERY', '需要唯一的projectId和可选status')
                    return self.send(200, store.actions(q['projectId'][0], q.get('status', ['open'])[0]))
                if path.startswith('/api/proposals/'):
                    return self.send(200, store.public_proposal(path.rsplit('/', 1)[1]))
                if path.startswith('/api/operations/'):
                    return self.send(200, store.operation(path.rsplit('/', 1)[1]))
                if path.startswith('/api/local-copies/'):
                    oid = path.rsplit('/', 1)[1]
                    store.operation(oid)
                    return self.send(200, {'ok': True, 'operationId': oid, 'localCopy': store.copy_status(oid)})
                if path == '/':
                    rows = store.actions('blue-bridge', 'all')['items']
                    return self.send(200, self.page('会后行动助手真实记录', self.table(rows) + '<p>记录来自当前 SQLite。提案结果页只有创建后才能打开。</p>'), 'text/html; charset=utf-8')
                if path.startswith('/proposals/'):
                    pid = path.rsplit('/', 1)[1]
                    r = store.proposal(pid)
                    p = store.public_proposal(pid)
                    content = '<p>项目 ' + html.escape(p['payload']['projectId']) + '；会议 ' + html.escape(p['payload']['meetingId']) + '</p>'
                    content += self.table(p['payload']['actions']) + '<p>内容摘要 <code>' + p['digest'] + '</code></p><p>状态 ' + p['status'] + '；有效期 ' + p['expiresAt'] + '</p>'
                    if p['status'] == 'pending':
                        content += '<form method="post" action="/ui/proposals/' + pid + '/approve"><input type="hidden" name="token" value="' + r['approve_token'] + '"><button>已核对上述确切内容，批准写入</button></form>'
                    if p['status'] in ('pending', 'approved'):
                        content += '<form method="post" action="/ui/proposals/' + pid + '/cancel"><button>取消尚未执行的提交</button></form>'
                    content += '<p>批准只保存在服务端，批准后客户端仍需提交同一提案。已提交写入无法通过取消撤回。</p>'
                    return self.send(200, self.page('行动提案批准页', content), 'text/html; charset=utf-8')
            else:
                payload = self.read_body()
                fault = self.headers.get('X-Test-Fault')
                if fault and (not self.server.test_mode or fault not in ('fail_before_write', 'drop_after_commit')):
                    raise HTTPError(403, 'FAULT_DISABLED', '故障注入只允许本地测试模式')
                if path == '/api/proposals':
                    return self.send(201, store.create(payload))
                if re.fullmatch(r'/ui/proposals/P-[a-f0-9]{32}/approve', path):
                    origin = self.headers.get('Origin')
                    if origin and origin != 'http://' + self.headers.get('Host', ''):
                        raise HTTPError(403, 'INVALID_ORIGIN', '批准请求必须来自当前结果页')
                    result = store.approve(path.split('/')[3], payload.get('token'))
                    return self.send(200, self.page('已批准', '<p>批准已经保存。操作标识 <code>' + result['operationId'] + '</code></p>'), 'text/html; charset=utf-8')
                if re.fullmatch(r'/(api|ui)/proposals/P-[a-f0-9]{32}/cancel', path):
                    return self.send(200, store.cancel(path.split('/')[3]))
                if path == '/api/commits':
                    status, result = store.commit(payload, fault)
                    if fault == 'drop_after_commit' and status == 200:
                        self.record(None, {'connectionDroppedAfterCommit': True, 'operationId': result['operationId']})
                        self.close_connection = True
                        self.connection.shutdown(socket.SHUT_RDWR)
                        self.connection.close()
                        return
                    return self.send(status, result)
                if path == '/__test/local-copies':
                    if not self.server.test_mode:
                        raise HTTPError(403, 'TEST_ONLY', '独立测试目标只允许本地测试模式')
                    return self.send(200, store.test_copy(payload))
            raise HTTPError(404, 'NOT_FOUND', '接口不存在')
        except HTTPError as e:
            self.send(e.status, e.body())
        except (ValueError, UnicodeDecodeError):
            self.send(400, HTTPError(400, 'INVALID_INPUT', '请求格式不合法').body())
        except Exception as e:
            self.send(500, HTTPError(500, 'INTERNAL_ERROR', '服务内部失败，请核对操作状态').body())
            print(type(e).__name__, str(e), flush=True)

    @staticmethod
    def table(rows):
        body = ''.join('<tr>' + ''.join('<td>' + html.escape(str(r.get(k, ''))) + '</td>' for k in ('title', 'owner', 'dueDate', 'source')) + '</tr>' for r in rows)
        return '<table><thead><tr><th>行动</th><th>负责人</th><th>截止日期</th><th>原文依据</th></tr></thead><tbody>' + body + '</tbody></table>'

    @staticmethod
    def page(title, content):
        return '<!doctype html><html lang="zh-CN"><meta charset="utf-8"><meta name="viewport" content="width=device-width"><title>' + title + '</title><style>body{font:16px system-ui;max-width:1000px;margin:40px auto;padding:0 20px;color:#23324b;background:#f6f8fc}table{background:white;border-collapse:collapse;width:100%}td,th{border:1px solid #dce4f0;padding:12px;text-align:left}button{padding:12px;margin:10px 0;background:#dde8ff;border:1px solid #9ab5e7;border-radius:8px}code{word-break:break-all}</style><h1>' + title + '</h1><p>仅本地虚构教学数据，未连接小艺平台或真实备忘录。</p>' + content + '</html>'


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--port', type=int, default=8765)
    parser.add_argument('--database', default='runtime/actions.sqlite')
    parser.add_argument('--audit', default='runtime/http.jsonl')
    parser.add_argument('--test-mode', action='store_true')
    parser.add_argument('--proposal-ttl', type=int, default=600)
    args = parser.parse_args()
    server = ThreadingHTTPServer(('127.0.0.1', args.port), Handler)
    server.store = Store(args.database, args.proposal_ttl)
    server.test_mode, server.audit, server.audit_lock = args.test_mode, args.audit, threading.Lock()
    if args.audit:
        Path(args.audit).parent.mkdir(parents=True, exist_ok=True)
    print('LISTENING http://127.0.0.1:' + str(server.server_port), flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


if __name__ == '__main__':
    main()

阅读提交逻辑时，可以把 Store.commit 中的行动插入、操作结果保存和幂等记录放在一起核对。它们使用同一个数据库事务；独立副本则使用另一份 SQLite 文件，不能作为设备保存结果。

2. 调用方的多步骤执行约定

文件路径：samples/06-multi-step/execution-rules.md。

这份文件保存调用方从整理、查询到提案、批准、提交和异常核对的完整约定。文件仍保留平台工具待绑定的说明，复制这些规则不会自动创建平台工具或取得设备能力。

# 06 会后行动多步骤执行约定

本地作者开发材料；平台工具标识待绑定。

1. 用户只要求整理时，输出清单与缺失信息，不调用任何写入接口。
2. 提取明确行动，保留source，建议、取消和重复分别处理。项目、负责人、截止日期未知时先询问，不猜测，不创建可提交提案。
3. 查询既有行动，向用户显示可能重复项。接口错误不能按空结果继续；标题相似不等于同一行动，不自动删除已有数据。
4. 创建包含完整信息的提案，保存proposalId、operationId、digest和approvalUrl。向用户展示确切清单和批准页地址；批准页的内容是服务端不可变提案。
5. 用户从真实结果页批准后，读取提案状态确认approved。客户端confirmed字段、对话口头确认和鉴权token不能替代该页面批准。
6. 为同一提案生成并保存一次幂等键，提交{proposalId,idempotencyKey}。重试必须复用同一键及同一提案；内容变更要新提案和新批准。
7. 失败响应准确报告error.code。连接超时或丢失响应时查询创建提案时已返回的operationId；未核对前保留结果待核对，不写成未执行。
8. 用户取消时停止未执行步骤，并取消未提交提案。已提交行动不能通过cancel撤回。
9. 用户确实要求备忘录副本时，在业务提交确认之后调用实际可用端工具。该工具未绑定时说明未完成，不能用/__test/local-copies替代。
10. 业务成功、端目标失败时分别报告两个结果。端工具超时而没有操作标识时，用户需要打开设备核对后决定重试，不承诺自动重试没有重复项。

当前实际检查覆盖自建服务；本材料未经过真实小艺对话或端插件调用。

接入时需要让调用方保留本次提案、操作标识和幂等键。规则中涉及的备忘录保存仍以真实端工具可用为前提，不能用服务的独立测试目标替代。

3. 完整 HTTP 检查程序

文件路径：examples/demo-service/test_service.py。

这份程序通过本机 HTTP 请求和 SQLite 行数检查缺少信息、批准限制、并发提交、响应丢失、取消及进程重启后的结果。程序启动时会删除 test-results/ 中匹配 actions.sqlite* 的文件，并清空请求与服务记录；需要保留旧记录时，应先另存该目录，再在复制的工程中执行。

#!/usr/bin/env python3
"""Real HTTP + SQLite tests. UI automation here is a local server test, not a human/device result."""
import concurrent.futures
import copy
import http.client
import json
import re
import shutil
import sqlite3
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

BASE = Path(__file__).resolve().parent
OUT = BASE / 'test-results'
OUT.mkdir(exist_ok=True)
DB = OUT / 'actions.sqlite'
for p in OUT.glob('actions.sqlite*'):
    p.unlink()
for p in ('requests-responses.jsonl', 'server-audit.jsonl'):
    (OUT / p).write_text('')
LOCK = threading.Lock()
RESULTS = []


def start(test=True, ttl=600):
    command = [sys.executable, str(BASE / 'server.py'), '--port', '0', '--database', str(DB),
               '--audit', str(OUT / 'server-audit.jsonl'), '--proposal-ttl', str(ttl)]
    if test:
        command.append('--test-mode')
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    line = process.stdout.readline().strip()
    if not line.startswith('LISTENING '):
        raise RuntimeError(line + process.stderr.read())
    return process, line.split(' ', 1)[1]


def stop(process):
    process.terminate()
    process.wait(timeout=10)


def request(method, path, body=None, headers=None, form=False):
    data = urllib.parse.urlencode(body).encode() if form else json.dumps(body, ensure_ascii=False).encode() if body is not None else None
    hd = {'Content-Type': 'application/x-www-form-urlencoded' if form else 'application/json'}
    hd.update(headers or {})
    req = urllib.request.Request(URL + path, data=data, method=method, headers=hd)
    raw, status = None, None
    try:
        with urllib.request.urlopen(req, timeout=5) as response:
            status, raw = response.status, response.read().decode()
    except urllib.error.HTTPError as e:
        status, raw = e.code, e.read().decode()
    except (http.client.RemoteDisconnected, urllib.error.URLError, ConnectionResetError) as e:
        raw = {'clientError': type(e).__name__, 'message': str(e)}
    value = json.loads(raw) if isinstance(raw, str) and raw.startswith('{') else raw
    with LOCK:
        with (OUT / 'requests-responses.jsonl').open('a') as f:
            f.write(json.dumps({'request': {'method': method, 'path': path, 'headers': hd, 'body': body},
                                'response': {'status': status, 'body': value}}, ensure_ascii=False) + '\n')
    return status, value


def check(name, condition):
    if not condition:
        raise AssertionError(name)
    RESULTS.append({'case': name, 'result': 'PASS'})
    print('PASS ' + name)


def payload(title='确认教学清单'):
    return {'projectId': 'blue-bridge', 'meetingId': 'M-20261004-01', 'meetingDate': '2026-10-04',
            'actions': [{'title': title, 'owner': '林舟', 'dueDate': '2026-10-08', 'source': '林舟在八日前确认教学清单。', 'missingFields': []}]}


def proposal(title='确认教学清单'):
    status, p = request('POST', '/api/proposals', payload(title))
    assert status == 201, p
    return p


def approve(p):
    status, page = request('GET', p['approvalUrl'])
    assert status == 200 and '<table>' in page
    token = re.search(r'name="token" value="([^"]+)"', page).group(1)
    status, _ = request('POST', '/ui/proposals/' + p['proposalId'] + '/approve', {'token': token}, form=True)
    assert status == 200
    return token


def count():
    with sqlite3.connect(DB) as c:
        return c.execute('SELECT count(*) FROM actions').fetchone()[0]


PROCESS, URL = start()
try:
    status, r = request('GET', '/api/actions?projectId=blue-bridge&status=open')
    check('Q01 正常查询两个未完成行动', status == 200 and [i['id'] for i in r['items']] == ['A-001', 'A-002'])
    status, r = request('GET', '/api/actions?projectId=empty-project&status=open')
    check('Q02 合法空结果', status == 200 and r['items'] == [])
    status, r = request('GET', '/api/actions?projectId=blue-bridge&status=unknown')
    check('Q03 状态错误与空结果区分', status == 400 and r['error']['code'] == 'INVALID_STATUS')
    status, r = request('GET', '/api/actions?projectId=blue-bridge&projectId=other')
    check('Q04 重复Query拒绝', status == 400)
    status, page = request('GET', '/')
    check('Q05 结果页显示真实数据库内容', status == 200 and '核对接口字段' in page and '整理验收清单' in page)
    p = payload()
    p['actions'][0]['owner'] = None
    p['actions'][0]['missingFields'] = ['owner']
    status, r = request('POST', '/api/proposals', p)
    check('W01 缺失负责人不创建提案不写入', status == 422 and r['error']['code'] == 'MISSING_INFORMATION' and count() == 3)
    p = proposal()
    status, r = request('POST', '/api/commits', {'proposalId': p['proposalId'], 'idempotencyKey': 'key_confirmed_01', 'confirmed': True})
    check('W02 confirmed字段不授权', status == 400 and count() == 3)
    status, r = request('POST', '/api/commits', {'proposalId': p['proposalId'], 'idempotencyKey': 'key_unapproved_01'})
    check('W03 未批准提交拒绝', status == 403 and count() == 3)
    status, public = request('GET', '/api/proposals/' + p['proposalId'])
    check('W04 工具提案响应无批准令牌', 'token' not in json.dumps(public))
    status, r = request('POST', '/ui/proposals/' + p['proposalId'] + '/approve', {'token': 'bad'}, form=True)
    check('W05 无结果页令牌不能批准', status == 403)
    token = approve(p)
    status, r = request('POST', '/ui/proposals/' + p['proposalId'] + '/approve', {'token': token}, form=True)
    check('W06 批准令牌单次使用', status == 409)
    commit_body = {'proposalId': p['proposalId'], 'idempotencyKey': 'key_parallel_01'}
    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
        parallel = list(executor.map(lambda _: request('POST', '/api/commits', commit_body), range(8)))
    success = parallel[0][1]
    check('W07 八并发同键同参返回同结果且仅写一次', all(s == 200 and b == success for s, b in parallel) and count() == 4)
    p2 = proposal('另一提案')
    approve(p2)
    status, r = request('POST', '/api/commits', {'proposalId': p2['proposalId'], 'idempotencyKey': 'key_parallel_01'})
    check('W08 同键异参409且无额外写入', status == 409 and r['error']['code'] == 'IDEMPOTENCY_CONFLICT' and count() == 4)
    status, r = request('POST', '/api/commits', {'proposalId': p['proposalId'], 'idempotencyKey': 'another_key_01'})
    check('W09 新键不能重复消耗同一批准', status == 409 and count() == 4)
    canceled = proposal('取消行动')
    status, r = request('POST', '/api/proposals/' + canceled['proposalId'] + '/cancel', {})
    status2, _ = request('POST', '/api/commits', {'proposalId': canceled['proposalId'], 'idempotencyKey': 'key_canceled_01'})
    check('W10 确认前取消阻止提交', status == 200 and r['status'] == 'canceled' and status2 == 409 and count() == 4)
    approved_cancel = proposal('批准后取消')
    approve(approved_cancel)
    status, r = request('POST', '/api/proposals/' + approved_cancel['proposalId'] + '/cancel', {})
    status2, _ = request('POST', '/api/commits', {'proposalId': approved_cancel['proposalId'], 'idempotencyKey': 'key_canceled_02'})
    check('W11 批准后提交前取消阻止写入', status == 200 and status2 == 409 and count() == 4)
    status, r = request('POST', '/api/proposals/' + p['proposalId'] + '/cancel', {})
    check('W12 提交后取消不撤回', status == 409 and count() == 4)
    failed = proposal('明确失败行动')
    approve(failed)
    fail_body = {'proposalId': failed['proposalId'], 'idempotencyKey': 'key_failure_01'}
    status, r = request('POST', '/api/commits', fail_body, {'X-Test-Fault': 'fail_before_write'})
    status2, r2 = request('POST', '/api/commits', fail_body)
    check('F01 明确失败保存同键失败结果且无业务写入', status == status2 == 503 and r == r2 and count() == 4)
    status, r = request('GET', '/api/operations/' + failed['operationId'])
    check('F02 查询明确失败操作', status == 200 and r['status'] == 'failed')
    lost = proposal('响应丢失行动')
    approve(lost)
    lost_body = {'proposalId': lost['proposalId'], 'idempotencyKey': 'key_lost_reply_01'}
    status, r = request('POST', '/api/commits', lost_body, {'X-Test-Fault': 'drop_after_commit'})
    status2, op = request('GET', '/api/operations/' + lost['operationId'])
    check('F03 提交后响应丢失仍可查询已提交', status is None and status2 == 200 and op['status'] == 'committed' and count() == 5)
    status, r = request('POST', '/api/commits', lost_body)
    check('F04 响应丢失后同键取回同一结果', status == 200 and r == op['result'] and count() == 5)
    status, r = request('POST', '/__test/local-copies', {'operationId': lost['operationId'], 'outcome': 'failed'})
    status2, op2 = request('GET', '/api/operations/' + lost['operationId'])
    check('F05 独立目标失败保留业务提交和失败状态', status == status2 == 200 and op2['status'] == 'committed' and op2['localCopy']['status'] == 'failed' and count() == 5)
    status, r = request('POST', '/__test/local-copies', {'operationId': p['operationId'], 'outcome': 'saved'})
    check('F06 独立测试目标实际保存内容', status == 200 and r['localCopy']['status'] == 'saved' and r['localCopy']['content']['actions'][0]['title'] == '确认教学清单')
    stop(PROCESS)
    PROCESS, URL = start()
    status, r = request('POST', '/api/commits', commit_body)
    status2, restarted = request('GET', '/api/operations/' + lost['operationId'])
    check('P01 重启后业务幂等操作及独立目标均持久化', status == 200 and r == success and status2 == 200 and restarted['status'] == 'committed' and restarted['localCopy']['status'] == 'failed' and count() == 5)
    stop(PROCESS)
    PROCESS, URL = start(test=False)
    status, r = request('POST', '/api/commits', lost_body, {'X-Test-Fault': 'drop_after_commit'})
    status2, _ = request('POST', '/__test/local-copies', {'operationId': lost['operationId'], 'outcome': 'saved'})
    check('S01 普通模式拒绝故障注入和测试目标', status == 403 and status2 == 403)
    stop(PROCESS)
    PROCESS, URL = start(ttl=1)
    expires = proposal('过期行动')
    approve(expires)
    time.sleep(1.1)
    status, r = request('POST', '/api/commits', {'proposalId': expires['proposalId'], 'idempotencyKey': 'key_expired_01'})
    status2, r2 = request('GET', '/api/operations/' + expires['operationId'])
    check('S02 过期批准拒绝提交并可核对', status == 409 and r['error']['code'] == 'PROPOSAL_EXPIRED' and status2 == 200 and r2['status'] == 'expired' and count() == 5)
    with sqlite3.connect(DB) as c:
        counts = {table: c.execute('SELECT count(*) FROM ' + table).fetchone()[0] for table in ['actions', 'proposals', 'operations', 'idempotency']}
    manifest = {'checkedAt': '2026-10-04', 'python': sys.version, 'transport': 'real 127.0.0.1 HTTP', 'storage': 'real SQLite plus independent copies SQLite',
                'cases': RESULTS, 'databaseCounts': counts, 'platform': 'NOT_RUN', 'cloudPlugin': 'NOT_BOUND', 'memoPlugin': 'NOT_BOUND', 'device': 'NOT_RUN'}
    (OUT / 'results.json').write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + '\n')
    (OUT / 'results.md').write_text('# 自建服务实际检查\n\n2026-10-04；真实 loopback HTTP 与 SQLite。\n\n' + '\n'.join('- ' + r['case'] + ' ' + r['result'] for r in RESULTS) + '\n\n平台、云插件、备忘录和设备尚未执行。\n')
finally:
    if PROCESS.poll() is None:
        stop(PROCESS)

程序中的 approve 会读取本地表单令牌并发送批准请求，用于检查服务行为，不能作为真实用户或设备批准的记录。文件末尾的检查日期是原文件已有内容，复制或重新执行不会自动把它更新为执行当天。

4. 历史故障输入与结果对照

文件路径：samples/06-multi-step/faults-and-results.md。

这份文件记录 2026-10-04 的故障输入和实际结果，可配合 HTTP 检查程序阅读。表中的返回与行数属于那次本地检查；重新执行以后，需要以新生成的请求记录和数据库为准。

# 故障输入与真实结果

2026-10-04真实HTTP检查，详见../../examples/demo-service/test-results/requests-responses.jsonl。

| 输入 | 实际结果 | 反馈约定 |
|---|---|---|
| 缺少owner创建提案 | HTTP422 MISSING_INFORMATION，actions仍3行 | 询问负责人，不声称已写入 |
| 未批准提交或传confirmed | 分别403 NOT_APPROVED与400 INVALID_BODY，无业务写入 | 打开批准页，不能把模型字段当批准 |
| 八个并发同键同参提交 | 八个200和相同JSON，actions从3到4 | 返回同一operationId/actionIds |
| 同键不同proposalId | 409 IDEMPOTENCY_CONFLICT，无新增行 | 检查本地保存的键与提案，不换键重复写原提案 |
| fail_before_write | HTTP503 INJECTED_FAILURE，操作failed，行数不变；同键再调返回同失败 | 明确失败；修正原因后需要新提案与批准 |
| drop_after_commit | 客户端RemoteDisconnected；GET operation为committed，actions增加一行 | 先核对，不能声称未执行 |
| 业务提交后独立测试目标failed | operation committed且localCopy failed | 分别报告业务完成与副本失败，不回滚业务 |
| 提交前cancel | canceled，后续提交409 | 停止尚未执行步骤 |
| 提交后cancel | 409 CANNOT_CANCEL，行数不变 | 说明无法撤回已执行写入 |
| 重启进程同键再调 | 相同结果、业务行数不增加，独立目标failed仍保存 | 使用持久操作查询，不依赖对话记忆 |

独立测试目标使用单独SQLite文件，不是备忘录或小艺端插件，saved测试结果同样不能作为设备保存证据。