前言

会后行动助手整理完当天的会议文字，用户又问蓝桥项目还有哪些工作没有完成。之前的任务保存在业务系统中，当天的会议记录并不能说明它们现在的状态。如果助手只根据对话补出清单，就可能漏掉旧任务，或者把已经完成的工作再次列出。我们需要查询保存这些记录的服务，再让助手依据实际返回回答。

我们先从只读查询接入工具，因为项目和完成状态能够明确限定读取范围，返回内容也可以与数据库页面逐项比较。查询不会改变任务，便于集中排查参数和结果。先在电脑上取得真实 HTTP 响应，再把同一组字段配置成云工具，后面就能分别检查服务收到的参数、返回的数据和助手给出的答复。

一、从用户问题确定项目与完成状态

用户要求查询蓝桥项目的未完成行动时，服务器需要收到两个筛选条件：蓝桥对应业务标识 blue-bridge，未完成对应 open。项目名称供用户辨认，业务标识供服务器筛选，两者不能随意替换。输入没有指定项目时，技能应该先询问查询对象。如果把蓝桥设成隐藏的固定项目，其他项目的请求也可能读到蓝桥数据。

当前服务还接受 done 和 all。前者只返回已完成行动，后者返回这个项目的两种状态。用户询问全部行动时才使用 all，用户询问还有哪些未完成工作时继续使用 open。截止日期已经过去并不能代替完成状态，服务器没有根据时间自动把行动改成已完成，助手也不能从日期推断任务已经结束。

打开 examples/demo-service/README.md，在 HTTP 契约中找到 GET /api/actions。它要求 projectId，允许省略 status，省略时查询未完成行动。这项缺省只减少状态参数，不决定项目。首次核对时仍然显式传入两个字段，实际返回中也有相同的 projectId 和 status，便于确认筛选范围没有在途中变化。

这个接口没有接收会议原文，也没有保存开关。它读取已经保存的记录，响应中的 meetingId 说明记录来自哪场会议，但当前请求没有会议筛选参数。用户只想查看某场会议时，不能把项目查询描述成已经按会议过滤。可以先说明项目查询的范围，再决定是否为真实服务增加会议筛选能力。

参数范围明确以后，先让本机服务按这两个条件读取一次。这里使用固定的虚构项目和姓名，能够重复比较输入与返回，也不会把当前会议的提取结果混进已有任务。接口检查通过以后，云工具才能采用同一组字段；如果服务本身还无法返回正确记录，先修改平台描述不会解决数据库查询问题。

二、启动本机服务，打开真实记录页面

打开终端，进入专题目录，再进入 examples/demo-service。运行环境只需要 Python 3 和它自带的 SQLite 模块，不需要安装第三方依赖。下面的命令与服务启动说明一致，其中数据库与请求记录都保存在当前目录的 runtime/ 下。第一次启动会建立数据库并加入三条教学记录，之后继续读取这个数据库，不会每次启动都清空行动。

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

保留这个终端窗口，让服务继续运行。浏览器地址栏输入 127.0.0.1:8765，打开首页后检查表格中是否出现核对接口字段、整理验收清单和提交页面草图。页面直接读取同一个 SQLite 文件，因此可以先核对标题、负责人、日期与原文依据。它没有显示完成状态，下一步还要打开查询响应，才能确认未完成记录是哪两条。

这张截图展示本机 HTTP 服务已经保存的页面。表格里同时有林舟、许岚的未完成行动和陈禾的已完成行动，因此三行记录不能直接回答用户还有多少未完成工作。继续读取 status=open 的响应，才能确认其中哪些符合查询条件，再将结果与首页核对。

浏览器无法连接时，先回到启动窗口检查进程是否已经退出，以及地址是否使用了 8765。端口被其他程序占用时，启动会失败；如果更换端口，后续浏览器与查询命令必须一起更换。页面能够连接却出现别的内容时，还要检查自己启动的文件和数据库路径，避免在另一个目录读到另一份 runtime/actions.sqlite。

三、用真实响应比较正常记录、空结果与参数错误

在第二个终端输入下面的只读请求。引号用于保护地址中的 &，使两个参数一起交给服务；没有引号时，终端可能把后半段当成另一项操作。-i 会同时展示 HTTP 状态和 JSON，便于把请求成功与业务内容分开观察。这个命令不会新增行动，也不会修改已有记录。

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

正常响应应当包含 HTTP 200、ok=true、projectId=blue-bridge 和 status=open。items 中的 A-001 是林舟负责的核对接口字段，日期为 2026-10-06；A-002 是许岚负责的整理验收清单，日期为 2026-10-07。这两项来自已保存记录，会议编号为 M-20261001-01，不属于当天新整理的 M-20261004-01。如果把两个会议编号写成相同值，就会改变记录来源。

阅读返回时，先确认根对象中的查询条件，再展开 items，每个数组对象对应一条行动。不要把根部的 status 与行动内部的 status 混用，前者说明本次筛选条件，后者说明这条记录保存的状态。查询 all 时，根部会是 all，各条记录仍然分别是 open 或 done。字段放在不同位置是为了保留这两种含义，转换成工具输出时也需要维持层次。

实际保存的响应在 samples/04-query/normal.json。它还保留请求方法、路径和当时的 HTTP 状态，打开时可以从外层定位 actualResponse。这一画面是已经取得的响应文件视图，当前启动服务后的数据如果曾经新增过行动，条数可能不同。发现新增记录时，先比较自己使用的数据库，不要改写历史响应使它符合当前页面。

服务在查询以前检查项目格式和状态范围。下面是 Store.actions 中已经工作的检查代码。它限制项目使用小写字母、数字和连字符，并把其他状态拒绝为参数错误；这些值属于当前业务接口的约定。

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')

参数通过检查以后，服务器使用参数化 SQL 读取记录。格式合法却没有匹配行动时，查询仍然完成，会返回空数组。现在把项目改为 empty-project，继续保留未完成状态，就能观察这个区别。这个标识是合法的教学查询对象，当前没有对应行动。

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

items=[] 表示服务器在当前查询条件下没有找到匹配项，助手可以据此回答没有查到该项目的未完成行动。接口没有另外查询项目目录，因此空数组不能证明项目不存在，也不能成为补出姓名或日期的依据。配置工具输出时需要允许数组为空，让没有匹配记录成为正常返回，避免用虚构任务填满清单。

然后保持蓝桥项目，把状态改为 unknown。这次请求故意使用不受支持的状态，用于观察失败怎样返回。它不应该触发数据库状态修改，也不应该被当作查询全部记录的快捷写法。

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

实际返回为 HTTP 400，错误码是 INVALID_STATUS。调用方应该修正状态以后再查询，助手不能向用户说明蓝桥没有行动。正常空结果已经完成读取，参数错误却没有完成所要求的查询，两种反馈会引导用户采取不同动作。工具调试因此需要同时保留 HTTP 状态和返回内容，只看 JSON 有没有出现记录还不足以判断结果。

图中三个分支对应本机接口的三种处理结果：有记录与空数组都可以返回 HTTP 200，参数错误返回 HTTP 400。SQLite 节点表示服务读取的数据来源；这些分支没有增加写入动作。

HTTP 路由还会拒绝未知字段、缺少项目及同名参数出现多次的请求。下面的代码来自真实路由，它确保服务不会从两个项目值中随意挑选一个。状态省略时，这一层把 open 交给后面的查询函数。

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]))

同名参数冲突时，先修正请求生成方式，再继续核对数据库。服务能够确认字段格式及数量，无法判断项目是不是用户想查询的对象。因此，到了模型调用环节，还需要把用户原始问题、工具参数和实际 HTTP 请求连起来比较；服务返回正确数据也可能对应了一次错误的项目选择。

如果终端只显示连接失败，没有收到 HTTP 状态，就先检查地址、端口和服务进程，此时修改项目标识还不能解决连接问题。已经收到 INVALID_STATUS 则说明请求到达了服务，应修正状态参数后再查询。根据实际反馈选择检查位置，可以避免在错误参数与网络故障之间反复切换，改动与问题无关的配置。

四、在云工具中保持同一份参数含义

本机的正常记录、空结果和参数错误都核对清楚以后，才继续配置云工具。先准备云端可达的服务地址和插件资源，再进入资源库，打开该云插件的创建工具入口。当前账号尚未为这项服务创建插件，因此可以先阅读 examples/tools/cloud-plugin-mapping.md 中的接口映射，资源 ID 和部署地址仍要来自自己的实际配置，不能填入猜测值或本机回环地址。

工具名称可以围绕会后行动查询填写，描述要说明读取指定项目的已有记录、仅查询、状态取值范围，以及空结果和 HTTP 错误的区别。名称与描述会参与模型选择工具，因此保存、管理之类宽泛描述会扩大动作含义。当前接口只负责查看，把这个范围说清楚以后，用户询问任务时才能得到一个明确的候选工具。

在工具基本信息中选择 GET，数据处理方式选择批处理。当前接口一次返回完整 JSON，没有持续到达的输出，使用流式配置会引入与响应不对应的字段。输入参数方式选择 Query，让参数进入地址的查询部分。把项目放进请求体以后，当前 GET 路由不会按预期读到它，参数必填开关也无法弥补这个位置错误。

然后配置两个输入字段。projectId 使用字符串，保持必填和模型可见；status 使用字符串并保持可选，说明只接受 open、done、all，缺省为 open，同样向模型开放。平台表单如果没有提供枚举设置，就在参数描述中写明范围，服务继续检查实际值。不要把 status 设置为必填以后又漏掉缺省说明，让未指定状态的正常问题变成参数缺失。

平台允许关闭参数对模型可见的开关。固定字段可以使用缺省值减少模型填写，必填字段关闭以后必须有缺省值。当前项目由用户指定，所以保持可见，也不配置 blue-bridge 作为所有用户的固定项目。只有服务确实限定一个项目时，才可以采用固定值，并同时说明工具只查询这个项目；这个取舍由业务范围决定。

输入配置结束以后，继续配置输出。根字段包含 ok、projectId、status 和 items，其中 items 是允许为空的数组。数组里的 id、meetingId、title、owner、dueDate、source 与 status 都来自真实响应，不能只保留标题后再让模型自行补其他字段。嵌套数组怎样填写及怎样映射，需要在实际表单中确认，调试返回是否保留这些字段才是继续绑定的条件。

输出里的行动标识用于辨认具体记录，标题用于阅读，两者不能互相替代。两条任务可以有相同标题，却来自不同会议或由不同人员负责；如果映射时只保留标题，最终清单就失去区分依据。当前查询也没有把日期转换成自然语言的接口要求，先保留原来的日期字符串，再由助手调整排版，能够减少转换阶段丢失年份或改变期限的机会。

数组映射完成后，先用正常响应核对两项行动是否都被保留。最终答复即使读起来通顺，也可能因为映射只取到第一项而漏掉许岚的任务。再用空数组检查工具在没有元素时怎样返回，确认它不会补入占位记录。这两次检查分别确认内容是否齐全、空结果是否如实保留，都满足以后再继续绑定。

同一云插件内的每个工具都需要配置输入和输出。发现查询工具无法使用时，还要检查插件中是否有其他工具的参数配置为空，不能只修改当前两个字段。错误响应也需要实际调试：本地接口返回 ok=false 和 error，平台怎样交付非 2xx 状态尚未取得结果。请保留真实工具失败信息，不能先假定所有错误对象都会完整传给技能。

已有普通 HTTP API 可以采用这条云插件路线。只有服务已经遵循 MCP 协议时，才考虑注册 MCP 插件；把路径改名不能让普通接口符合协议。外部平台导入还有自己的渠道限制，当前服务没有执行这项注册。查询接入先使用现有 API，能够减少与当前问题无关的协议转换，后续调用方确实需要 MCP 时再单独处理。

五、把工具加入技能以后，从请求核对到最终清单

工具完成配置和调试以后，回到 Skill 创建对话，点击下方 @，选择插件，再按来源与类型找到自己的云插件。打开卡片详情，确认工具列表中确实包含刚才调试的查询工具，再点击 + 添加。添加之前要核对实际名称与资源版本，避免把同名旧插件加入当前项目。当前账号尚未完成这一步，所以这里的继续条件是取得真实资源详情。

添加依赖以后，使用查询蓝桥项目未完成行动这一固定输入调试。检查实际工具参数是否为 projectId=blue-bridge、status=open，再比较服务请求与最终清单。助手应该返回林舟和许岚的两项行动，保留各自负责人、日期及会议来源，不加入已完成的页面草图。如果平台没有发出请求，先检查资源依赖、可达地址和访问条件；请求已经到达却用了 all，再检查参数描述与模型选择。

最终反馈需要忠实保留查询范围。source 帮助用户核对任务来自哪句会议记录，它不是内部推理，也不能保证原始记录没有错误。服务字段经过正确映射以后，助手可以整理排版，不能换负责人、改截止日期，或者按自己的判断删掉看似重复的行动。用户需要修改任务时，应该进入有明确写入条件的操作，当前查询不会承担变更。

排查时保留请求和响应，比只截最终答复更有用。当前服务把本机请求写入指定审计文件，响应文件又保存实际字段，二者可以说明错误发生在参数进入以前还是数据返回以后。正式系统记录这些内容时，还需要按业务要求脱敏和限制读取，不能把凭据与完整业务正文无条件保存。这个建议来自接口排查经验，日志本身不提供项目访问授权。

总结

本机接口已经提供了能够核对的项目查询结果，正常记录、合法空结果和参数错误也有各自的响应。接入自己的服务时，可以先按这三种情况检查，再将相同的字段含义保留到云工具中。如果后面得到的清单不符，就能沿参数、响应和最终答复逐步比较，找到最先出现差异的位置。

继续到小艺调试的条件是服务可通过 HTTPS 到达、真实插件与工具已经创建，并且输入输出经过实际核对。取得调用记录以后，再比较用户问题、工具参数和最终反馈。当前查询不会保存新行动，不能借一次读取成功推断其他业务动作也已经接入。

完整配套文件

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()

examples/demo-service/test_service.py

#!/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)

examples/demo-service/README.md

# 会后行动助手教学服务

2026-10-04 创建；Python 标准库 HTTP + SQLite。接口、字段、批准和幂等规则均为作者设计，不是小艺平台内置能力。服务默认只监听 `127.0.0.1:8765`，禁止将本地地址填成云插件可访问地址。平台调用前需要读者自行部署 HTTPS、接入用户身份与权限；本轮不公开暴露或部署。

## 运行与结果页

```sh
/usr/bin/python3 server.py --port 8765 --database runtime/actions.sqlite --audit runtime/http.jsonl
```

浏览器打开 `http://127.0.0.1:8765/` 查看真实 SQLite 数据。创建提案后打开响应的 `approvalUrl`，逐项核对再点击批准。测试模式仅供本地检查。

```sh
/usr/bin/python3 test_service.py
```

测试自动启动真实 HTTP 子进程、生成临时 SQLite、记录请求响应、停止并重启服务验证持久化。产物保存到 `test-results/`。它只证明自建服务，不证明 Skill、云插件、小艺或备忘录运行。

## 统一数据

项目 `blue-bridge`；会议 `M-20261004-01`，日期 `2026-10-04`，时区 `Asia/Shanghai`。内置记录 A-001 至 A-003，两个 open、一个 done。`empty-project` 用于合法空结果。姓名和内容均为虚构教学材料。

行动字段为 `title`、`owner`、`dueDate`、`source`、`missingFields`。提取阶段允许未知负责人及日期为 null；创建可批准提案时必须补齐，`missingFields` 必须为空。日期为 YYYY-MM-DD。服务检查字段类型与范围，不能识别会议原文是否真的确定了行动；这项业务判断仍要由规则、测试与用户核对承担。

## HTTP 契约 v1

所有 JSON 正常响应含 `ok:true`，错误含 `ok:false,error:{code,message,details}`。下表描述作者服务，不是平台工具标识。

| 方法及路径 | 输入 | 结果与约束 |
|---|---|---|
| GET `/api/actions` | Query `projectId` 必填；`status` 为 open/done/all，默认 open | 200 `{ok,projectId,status,items}`；合法空项目返回 items:[]；不合法字段或状态 400 |
| POST `/api/proposals` | JSON `{projectId,meetingId,meetingDate,actions:[{title,owner,dueDate,source,missingFields:[]}]}` | 201 `{ok,proposalId,operationId,status:'pending',digest,expiresAt,approvalUrl}`；缺失信息 422，不创建提案 |
| GET `/api/proposals/{proposalId}` | 无 | 提案内容、摘要与状态，不返回批准令牌 |
| GET `/proposals/{proposalId}` | 无；这是结果页 | 显示确切清单、摘要、有效期及批准/取消表单；批准 token 只出现于页面，不进入工具响应 |
| POST `/ui/proposals/{proposalId}/approve` | 结果页 form token | 在同一 SQLite 中持久批准。普通 API 不提供批准接口；客户端 confirmed 不构成授权 |
| POST `/api/proposals/{proposalId}/cancel` | 空 JSON | pending/approved 时保存 canceled；已提交不能撤回，409 |
| POST `/api/commits` | JSON `{proposalId,idempotencyKey}` | 200 `{ok,operationId,status:'committed',actionIds,proposalId,digest}`；仅批准且未过期的提案能写入；幂等键为 8..100 字符 |
| GET `/api/operations/{operationId}` | 无；提案创建时已返回 operationId | 200 `{ok,operationId,proposalId,status,result,localCopy}`；独立核对 pending/approved/canceled/committed/failed/expired |
| GET `/api/local-copies/{operationId}` | 无 | 独立目标库中记录的本地副本状态；默认 not_requested；此目标是本地测试替身，不是备忘录 |

批准只批准页面显示的不可变提案。修改内容需要新建提案、重新批准。批准 token 一次使用，过期后不能提交。正式部署必须使用真实用户登录、CSRF 防护和权限校验；当前页面仅为 loopback 教学授权界面，不证明华为账号授权或生产安全。

提交先持有 `BEGIN IMMEDIATE` 事务，再核对幂等键和批准状态、插入业务行、保存操作结果、保存幂等记录、消耗批准，最后同一事务 COMMIT。同键同参返回已存的相同 HTTP 状态和 JSON，同键异参返回 409。不同键重复提交同一提案同样被拒绝，避免第二次消耗批准。幂等仅覆盖同一个数据库实例和这项请求约定，不负责语义相似行动去重。

## 故障与取消范围

仅 `--test-mode` 启用 `X-Test-Fault` 请求头。`fail_before_write` 在写入前保存明确失败，业务行数不变；`drop_after_commit` 在事务提交后关闭连接，客户端收不到响应，但 operation 已 committed。正式模式拒绝故障注入请求。收到未知结果时先使用已知 operationId 查询，不能通过超时推断数据未改变。

仅测试模式接受 POST `/__test/local-copies`，JSON `{operationId,outcome:'saved'|'failed'}`。它使用**独立 SQLite 文件**保存目标内容或失败状态，展示业务写入成功、第二目标失败的部分成功。业务事务与副本事务互不回滚。该接口不得作为备忘录调用成功证据；真实备忘录工具尚未绑定。

取消与提交分别持有数据库写事务，因此先取得事务的一方决定当前变化。取消先完成时，提交被拒绝；提交先完成时，取消不能撤回业务行。测试检查确认前取消和已提交后取消，不承诺中断已经执行的远端动作。

## 平台待绑定

完整云工具配置映射见 `../tools/cloud-plugin-mapping.md`。当前资源库工具 ID、插件 ID、HTTPS endpoint、鉴权方式及端侧备忘录工具名均未取得，不填写虚构值。平台用户确认如何打开外部结果页、如何回传 operationId、模型是否调用正确工具，都需要真实账号与设备补验。

samples/04-query/README.md

# 04 只读查询输入与真实结果

输入分别为查询蓝桥未完成行动、查询empty-project未完成行动、使用非法status。对应请求与真实HTTP响应见normal.json、empty.json、invalid.json。2026-10-04从正在监听127.0.0.1:8765的教学服务取得。正常响应为A-001、A-002；空响应items为空；非法状态HTTP400/INVALID_STATUS。

Skill核对输入：查询蓝桥项目未完成的行动，不创建任何新行动。
预期工具参数：projectId=blue-bridge，status=open；预期仅调用只读查询。Skill真实调用状态：NOT_RUN。平台工具和HTTPS endpoint尚未绑定，不能把上述HTTP检查填入模型调用结果。

samples/04-query/normal.json

{
  "method": "GET",
  "path": "/api/actions?projectId=blue-bridge&status=open",
  "actualHttpStatus": 200,
  "actualResponse": {
    "items": [
      {
        "dueDate": "2026-10-06",
        "id": "A-001",
        "meetingId": "M-20261001-01",
        "owner": "林舟",
        "source": "林舟在六日前核对接口字段。",
        "status": "open",
        "title": "核对接口字段"
      },
      {
        "dueDate": "2026-10-07",
        "id": "A-002",
        "meetingId": "M-20261001-01",
        "owner": "许岚",
        "source": "许岚在七日前整理验收清单。",
        "status": "open",
        "title": "整理验收清单"
      }
    ],
    "ok": true,
    "projectId": "blue-bridge",
    "status": "open"
  },
  "checkedAt": "2026-10-04",
  "scope": "real local HTTP; platform not run"
}

samples/04-query/empty.json

{
  "method": "GET",
  "path": "/api/actions?projectId=empty-project&status=open",
  "actualHttpStatus": 200,
  "actualResponse": {
    "items": [],
    "ok": true,
    "projectId": "empty-project",
    "status": "open"
  },
  "checkedAt": "2026-10-04",
  "scope": "real local HTTP; platform not run"
}

samples/04-query/invalid.json

{
  "method": "GET",
  "path": "/api/actions?projectId=blue-bridge&status=unknown",
  "actualHttpStatus": 400,
  "actualResponse": {
    "error": {
      "code": "INVALID_STATUS",
      "details": null,
      "message": "status只接受open、done或all"
    },
    "ok": false
  },
  "checkedAt": "2026-10-04",
  "scope": "real local HTTP; platform not run"
}

examples/tools/cloud-plugin-mapping.md

# 云插件配置映射与待绑定项

2026-10-04，S06/S17 已取得当期完整官方正文。本文为人工配置材料，**不是平台导入 Schema**。本地 `server.py` 是作者 HTTP 服务；JSON 结构以其真实响应为准。当前未创建平台插件、工具或资源 ID。

## 04 查询工具

建议显示名为会后行动查询，最终名称和标识从资源库已创建工具详情回填。描述建议使用：读取指定项目的既有行动记录；仅查询，不创建、不修改；status只接受open、done、all；正常空结果需要如实反馈，HTTP错误不能解释为没有行动。

| 配置项目 | 本服务值 | 平台核对 |
|---|---|---|
| 服务地址 | 当前本地 http://127.0.0.1:8765；平台待部署HTTPS | 不将loopback地址作为云端可达地址 |
| HTTP方法和路径 | GET /api/actions | GET入参方式必须为Query |
| 数据处理方式 | 完整JSON响应；批处理 | 不使用流式输出规则套用普通JSON |
| projectId | string，必填，示范blue-bridge | 模型可见；不自动用别的项目代替 |
| status | string，open/done/all，缺省open | 可见且有缺省值；枚举若页面不能配置则写清描述，服务仍检查 |
| ok | boolean | 配置输出参数 |
| projectId | string | 返回实际查询项目 |
| status | string | 返回实际查询状态 |
| items | array<object> | 输出数组允许为空 |
| items.id / meetingId / title / owner / dueDate / source / status | 均string | 逐项配置真实嵌套输出结构；是否支持当前数组映射以页面调试为准 |

S17要求同一云插件的每一个工具都配置输入和输出参数。关闭参数对模型可见的开关时，必填参数需要填写缺省值。固定服务字段可以考虑隐藏；项目选择需要用户指定，不能通过隐藏并固定blue-bridge冒充支持多个项目。保留服务器参数校验，不把页面字段配置当成数据验证。

错误响应为 `{ok:false,error:{code,message,details}}`，工具调试须保存原始HTTP状态。平台怎样暴露非2xx给Skill需要实际验证，不预设平台会保留所有响应字段。

## 06 提案、提交与核对

以下是候选显示名，平台实际 ID/调用名均待回填。批准界面**不注册为模型可调用工具**。

| 候选显示名 | 方法路径 | 入参 | 出参 |
|---|---|---|---|
| 创建行动提案 | POST /api/proposals | projectId string；meetingId string；meetingDate string；actions array，含title/owner/dueDate/source string及missingFields空数组 | ok boolean；proposalId/operationId/status/digest/expiresAt/approvalUrl string |
| 取消未提交提案 | POST /api/proposals/{proposalId}/cancel | path proposalId string | ok boolean；proposalId/operationId/status string |
| 提交已批准提案 | POST /api/commits | proposalId string；idempotencyKey string | ok boolean；operationId/status/proposalId/digest string；actionIds array<string> |
| 核对行动操作 | GET /api/operations/{operationId} | path operationId string | ok boolean；operationId/proposalId/status string；result object/null；localCopy object |

POST参数方式、path参数表达、嵌套数组、nullable输出和错误对象在当前平台表单中的准确填写方式需要从实际工具详情核对，未取得实配结果。不要将此映射表转换成猜测的module.json工具资源。正式HTTPS部署还需要用户身份、项目访问授权、可信批准页面与受保护的批准表单；本地批准令牌不构成生产账号体系。

`localCopy.source=local_test_destination` 只表示本地独立目标测试，不能映射为备忘录工具回执。05真实端工具名、资源ID、参数与权限从当前平台取得后才能添加，不使用本服务的/__test接口代替它。

## 原始配置留档模板

- 插件实际名称与ID：待绑定
- 工具实际名称与ID：待绑定
- HTTPS endpoint：待部署
- 鉴权字段和安全保存方式：待实际部署确定
- 输入、输出参数页面：待截图/导出
- 调试HTTP状态与响应：待云插件实调
- Skill所用资源版本：待绑定
- 设备入口与系统、小艺版本：待真机
- 核查人及日期：待实际操作

## MCP选型边界

现有普通HTTP业务API可以使用云插件。只有服务已遵循MCP协议时，才考虑MCP插件；把HTTP接口改名不能使其符合MCP。S20区分标准注册和外部平台导入。外部平台导入针对魔搭已有长期有效自部署MCP，且只支持本账号智能体渠道调试，不能由此推导插件市场或主对话公开发布权限。当前服务没有实现MCP，也没有执行任何公开发布。