完善课程订单准备与售后数量展示

README.md

@@ -23,6 +23,10 @@
 
 Grafana 支持匿名只读访问，管理操作使用表中的账号登录。消费者与商家凭据保存在 `.work/frontend-accounts.json`，平台凭据保存在 `.work/platform-accounts.json`；首次在新环境准备账户时，脚本生成该环境的密码，以对应账户文件为准。
 
+## 课程初始订单
+
+每次上课前，在项目根目录使用 PowerShell 7 执行 `./scripts/start_course_demo.ps1`。入口创建本次课程独立客户和已签收两天的键盘订单，同时登记 OMS 与平台，并在课堂期间持续同步。账户与订单信息保存在 `.work/course-demo/active.json`，使用方法见 [课程演示数据](docs/course-demo.md)。
+
 ## 产品与业务
 
 | 使用者 | 入口与职责 |

docs/course-demo.md

@@ -0,0 +1,78 @@
+# 课程初始订单
+
+每次课程使用独立客户和订单。商品为 `keyboard-01` 办公键盘，可申请数量为 1 件，创建时签收时间设为当前时间的两天前。课堂故障说明使用“键盘部分按键失灵，申请换货 1 件”。准备完成时保留空会话，换货申请由课堂操作创建。
+
+## 每次上课前执行
+
+在项目根目录使用 PowerShell 7：
+
+```powershell
+.\scripts\start_course_demo.ps1
+```
+
+入口为本次课程生成名称，将订单登记到独立 OMS，经已授权的 SERVICE 账户同步到平台，并核对当前客户能读取订单。课程记录保存到 `.work/course-demo/active.json`，历史记录保存到 `.work/course-demo/sessions/`。客户密码与 OMS 凭据保存在 `.work/course-demo/credentials.json`；这些文件受 `.gitignore` 排除。
+
+消费者使用本次记录中的 `customer.username` 登录 H5。主管使用现有 `supervisor01` 账户，员工密码保存在 `.work/frontend-accounts.json`。客户和主管使用不同浏览器会话。
+
+入口加载课程 OMS 凭据到 `fulfillment` 容器，然后在后台每 60 秒从 OMS 读取订单并同步到平台，默认持续 180 分钟。确认和审批要求订单观察时间在最近五分钟内，因此演示期间需要保持同步进程运行。
+
+可以指定课程名称和同步时长：
+
+```powershell
+.\scripts\start_course_demo.ps1 -Session lesson-02 -DurationMinutes 120
+```
+
+相同名称复用同一客户、订单和会话，保留签收时间、商品数量、审批和履约结果。需要一份初始数据时使用新名称，或者执行不带 `-Session` 的命令。每次启动只维护当前课程的后台同步，此前的课程记录继续保留。
+
+## 课中订单同步
+
+单次同步当前课程：
+
+```powershell
+.\.venv\Scripts\python.exe scripts\prepare_course_demo.py sync
+```
+
+同步读取 OMS 当前商品数量与来源版本，不改变 OMS 订单。每次请求使用新的 `Idempotency-Key`，使平台更新 `observed_at`。同一来源版本下的订单业务内容必须保持一致。
+
+同步日志位置记录在 `.work/course-demo/sync-process.json`。遇到网络、权限或业务检查失败，进程终止并将错误写入对应日志；检查服务后可以为该课程再次启动同步。登录会话到期前重新获取同步账户令牌。
+
+课后停止后台同步：
+
+```powershell
+.\scripts\start_course_demo.ps1 -Action Stop
+```
+
+该命令按进程标识、创建时间和命令路径核验目标，停止本入口创建的同步进程，并保留课程数据及项目服务。
+
+## 课堂登记仓储完成
+
+客户确认、主管批准后，等待平台工单进入 `PROCESSING`、OMS 操作进入 `ACCEPTED`。在项目根目录执行：
+
+```powershell
+.\.venv\Scripts\python.exe scripts\prepare_course_demo.py complete
+```
+
+命令读取当前课程记录，核验唯一的本订单履约操作，使用仓储凭据调用 `POST /exchanges/{operation_id}/complete`，并同步 OMS 当前订单数量。需要指定课程或操作时使用 `--session lesson-02` 和 `--operation-id <UUID>`。跨订单操作会被拒绝；重复完成返回原有完成结果。
+
+OMS 返回 `COMPLETED` 后，平台履约进程继续查询并将工单更新为 `COMPLETED`。客户在进度页面点击“读取最新方案”检查结果。完成证据保存在该课程记录的 `completion` 字段。
+
+## 环境与数据范围
+
+- 使用已经运行的完整项目：平台 API 默认端口 `28000`，OMS 默认端口 `28100`，前端端口 `18762`。
+- 本机 `.env`、`.env.oms`、`.work/runtime.env`、`.work/models` 和开发账户文件需要存在，Python 环境包含项目开发依赖。
+- 准备命令只支持 `development` 或 `test`，并限定现有开发商家和店铺。当前已发布的键盘规则和商品需要有效。
+- 课程客户仅拥有本人业务所需权限；同步账户仅拥有订单同步权限。平台 OMS 凭据与仓储完成凭据分别登记。
+- 已存在其他 OMS 连接器时，准备命令终止并报告冲突；现有规则和发布资料保持原有配置。
+- 新建 OMS 订单使用具有 `can_fulfill` 权限的凭据调用 `PUT /orders/{order_id}`，平台同步使用 SERVICE 身份调用 `PUT /internal/v1/orders/{order_id}`，客户查询使用本人会话。重复准备核对已有 OMS 订单并复用记录。
+- 单次 HTTP 请求最长 10 秒；准备流程最长 90 秒。持续同步有明确时长，PowerShell 入口允许 1 至 180 分钟。
+- 准备和同步不发送咨询消息，不创建 AgentRun，也不调用生成模型。课堂中的 AI 咨询使用项目当前已发布的模型配置。
+
+## 已执行的真实业务验证
+
+2026-09-28 使用独立验收课程 `validation-20260928-1823`，通过真实 API、当前履约进程、RabbitMQ 和独立 OMS 完成 `DRAFT → WAITING_REVIEW → APPROVED → PROCESSING → COMPLETED`。
+
+同键重复确认返回一致，确认与审批记录各 1 条；重复登记仓储完成保持同一操作标识和外部版本。完成后平台与 OMS 的可申请数量均为 0，来源版本均为 2。该客户 `AgentRun` 与 `ModelAttempt` 数据库记录数量均为 0。验收耗时 14.09 秒，直接 HTTP 请求 21 次，单并发，流程期限 120 秒；另外执行两次同步命令和一次重复完成命令。
+
+报告保存在 `.work/course-demo/validation-report.json`。课堂初始数据使用独立课程 `lesson-01-20260928`，保持数量 1、来源版本 1、空会话及零工单。模型生成与工具选择由课堂咨询触发，其运行结果需要按当次记录查看。
+
+同日通过真实浏览器检查课程客户登录、订单列表、订单详情及申请入口：列表与详情均显示可申请数量 1，申请表数量默认 1、故障说明为空。会话消息为 0，当前 Agent 运行为空，页面错误和失败响应均为 0。浏览器报告保存在 `.work/course-demo/lesson-01-20260928-browser.json`。

scripts/prepare_course_demo.py

@@ -0,0 +1,572 @@
+import argparse
+import asyncio
+import csv
+import hashlib
+import json
+import math
+import os
+import re
+import secrets
+import subprocess
+import time
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from uuid import UUID, uuid4
+
+import httpx
+from dotenv import dotenv_values, set_key
+from filelock import FileLock
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from zhifu.core.config import Settings
+from zhifu.core.database import create_engine
+from zhifu.modules.after_sales.models import BusinessEvent, ExchangeRule, Ticket
+from zhifu.modules.identity.access_models import Principal, StoreGrant
+from zhifu.modules.identity.auth import password_hasher
+from zhifu.modules.identity.models import Store, Tenant
+from zhifu.modules.integrations.models import Connector, ExternalOperation
+from zhifu.modules.storefront.models import CatalogProduct
+from zhifu.oms.database import oms_engine
+from zhifu.oms.models import Credential
+
+ROOT = Path(__file__).resolve().parents[1]
+DATA = ROOT / ".work" / "course-demo"
+TENANT = UUID("ac700000-2026-4000-8000-000000000001")
+STORE = UUID("ac700000-2026-4000-8000-000000000002")
+TOKEN_ENV = "ZHIFU_COURSE_OMS_TOKEN"
+CONNECTOR_URL = "http://oms:8100"
+CUSTOMER_PERMISSIONS = (
+    "orders:read", "tickets:read", "after_sales:create", "after_sales:confirm",
+    "after_sales:cancel",
+)
+
+
+def require(condition, message):
+    if not condition:
+        raise ValueError(message)
+
+
+def utc_now():
+    return datetime.now(UTC).isoformat()
+
+
+def read_json(path):
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def protect_file(path):
+    # Windows 使用当前用户 SID 设置访问权限，凭据文件仅当前用户和 SYSTEM 可读取。
+    if os.name == "nt":
+        result = subprocess.run(
+            ["whoami", "/user", "/fo", "csv", "/nh"],
+            check=True, capture_output=True, text=True, timeout=10,
+        )
+        sid = next(csv.reader(result.stdout.strip().splitlines()))[1]
+        subprocess.run(
+            ["icacls", str(path), "/inheritance:r", "/grant:r",
+             f"*{sid}:(F)", "*S-1-5-18:(F)"],
+            check=True, capture_output=True, timeout=10,
+        )
+    else:
+        path.chmod(0o600)
+
+
+def write_json(path, payload, *, private=False):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(f".{path.name}.{uuid4().hex}.new")
+    # 同目录临时文件与原子替换保证中断后仍能读取完整 JSON；临时文件也使用受限权限。
+    descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
+        if private:
+            protect_file(temporary)
+        json.dump(payload, stream, ensure_ascii=False, indent=2)
+        stream.write("\n")
+        stream.flush()
+        os.fsync(stream.fileno())
+    temporary.replace(path)
+
+
+def session_name(value):
+    require(
+        re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_-]{0,47}", value) is not None,
+        "课程名称须为 1 至 48 个英文字母、数字、下划线或连字符，并以字母或数字开头",
+    )
+    require(
+        value.upper() not in {"CON", "PRN", "AUX", "NUL"}
+        and re.fullmatch(r"(?:COM|LPT)[1-9]", value.upper()) is None,
+        "课程名称不能使用 Windows 保留文件名",
+    )
+    return value
+
+
+def validate_scope(value):
+    require(
+        value["tenant_id"] == str(TENANT) and value["store_id"] == str(STORE),
+        "课程数据仅允许使用固定开发商家和店铺",
+    )
+
+
+def validate_url(value):
+    url = httpx.URL(value)
+    require(
+        url.scheme in {"http", "https"} and url.host in {"127.0.0.1", "localhost", "::1"}
+        and not url.username and not url.password and not url.query and not url.fragment
+        and url.path in {"", "/"},
+        "课程接口地址必须为本机 HTTP 或 HTTPS 服务根地址",
+    )
+    return str(url).rstrip("/")
+
+
+def load_manifest(name=None):
+    path = DATA / "sessions" / f"{session_name(name)}.json" if name else DATA / "active.json"
+    manifest = read_json(path)
+    validate_scope(manifest)
+    session_name(manifest["session"])
+    require(manifest["schema_version"] == 1, "课程记录版本不受支持")
+    validate_url(manifest["api_url"])
+    validate_url(manifest["oms_url"])
+    require(manifest["order"]["item_id"] == "keyboard-01", "课程订单商品不匹配")
+    return manifest
+
+
+def save_manifest(manifest, *, activate=False):
+    write_json(DATA / "sessions" / f"{manifest['session']}.json", manifest)
+    active = DATA / "active.json"
+    if activate or (active.exists() and read_json(active)["session"] == manifest["session"]):
+        write_json(active, manifest)
+
+
+def credentials(accounts=None):
+    path = DATA / "credentials.json"
+    if path.exists():
+        value = read_json(path)
+        validate_scope(value)
+        require(value["schema_version"] == 1, "课程凭据版本不受支持")
+        protect_file(path)
+        return value
+    require(accounts is not None, "课程凭据不存在，请执行 prepare")
+    value = {
+        "schema_version": 1, "tenant_id": str(TENANT), "store_id": str(STORE),
+        "customer_password": accounts["password"],
+        "service": {
+            "principal_id": str(uuid4()), "username": "course-order-sync",
+            "password": secrets.token_urlsafe(32),
+        },
+        "oms_platform_token": secrets.token_urlsafe(32),
+        "oms_fulfillment_token": secrets.token_urlsafe(32),
+    }
+    # 先保存凭据，再创建数据库记录；后续显式重试继续使用相同身份与令牌。
+    write_json(path, value, private=True)
+    return value
+
+
+def prepare_files(args):
+    accounts = read_json(ROOT / ".work" / "frontend-accounts.json")
+    validate_scope(accounts)
+    name = session_name(args.session or datetime.now().strftime("%Y%m%d-%H%M%S-%f"))
+    path = DATA / "sessions" / f"{name}.json"
+    secret = credentials(accounts)
+    if path.exists():
+        manifest = load_manifest(name)
+        require(
+            manifest["api_url"] == args.api_url and manifest["oms_url"] == args.oms_url,
+            "已有课程接口地址与本次参数不同，请使用原有地址",
+        )
+    else:
+        manifest = {
+            "schema_version": 1, "session": name, "tenant_id": str(TENANT),
+            "store_id": str(STORE), "created_at": utc_now(), "status": "PREPARING",
+            "api_url": args.api_url, "oms_url": args.oms_url,
+            "customer": {"principal_id": str(uuid4()), "username": f"course-{name}"},
+            "order": {
+                "order_id": str(uuid4()), "item_id": "keyboard-01", "product_type": "keyboard",
+                "available_quantity": 1, "source_version": 1,
+                "signed_at": (datetime.now(UTC) - timedelta(days=2)).isoformat(),
+            },
+            "staff": {name: identifier for name, identifier in accounts["actors"].items()
+                      if name in {"staff01", "supervisor01", "admin01"}},
+        }
+        save_manifest(manifest)
+    return manifest, secret
+
+
+async def ensure_principal(session, actor, password, kind, permissions):
+    identifier = UUID(actor["principal_id"]).bytes
+    existing = await session.get(Principal, identifier, with_for_update=True)
+    if existing is None:
+        collision = await session.scalar(select(Principal).where(
+            Principal.tenant_id == TENANT.bytes, Principal.username == actor["username"],
+        ))
+        require(collision is None, "课程账户名已经被其他身份使用")
+        session.add(Principal(
+            id=identifier, tenant_id=TENANT.bytes, username=actor["username"],
+            password_hash=await asyncio.to_thread(password_hasher.hash, password), kind=kind,
+        ))
+        await session.flush()
+        for permission in permissions:
+            session.add(StoreGrant(
+                tenant_id=TENANT.bytes, store_id=STORE.bytes,
+                principal_id=identifier, permission=permission,
+            ))
+    else:
+        require(
+            existing.tenant_id == TENANT.bytes and existing.username == actor["username"]
+            and existing.kind == kind and existing.status == "ACTIVE",
+            "已有课程账户身份或状态与课程配置不一致",
+        )
+        require(
+            await asyncio.to_thread(password_hasher.verify, password, existing.password_hash),
+            "已有课程账户密码与本地受保护凭据不一致",
+        )
+        granted = set((await session.scalars(select(StoreGrant.permission).where(
+            StoreGrant.tenant_id == TENANT.bytes, StoreGrant.store_id == STORE.bytes,
+            StoreGrant.principal_id == identifier,
+        ))).all())
+        require(granted == set(permissions), "已有课程账户授权与课程要求不一致")
+
+
+async def bootstrap(settings, manifest, secret):
+    engine = create_engine(settings)
+    oms = oms_engine()
+    try:
+        sessions = async_sessionmaker(engine, expire_on_commit=False)
+        async with sessions.begin() as session:
+            tenant = await session.get(Tenant, TENANT.bytes)
+            store = await session.get(Store, STORE.bytes, with_for_update=True)
+            require(tenant is not None and tenant.status == "ACTIVE", "开发商家未启用")
+            require(
+                store is not None and store.tenant_id == TENANT.bytes and store.status == "ACTIVE",
+                "开发店铺未启用或不属于指定商家",
+            )
+            product = await session.get(CatalogProduct, (TENANT.bytes, STORE.bytes, "keyboard-01"))
+            require(product is not None and product.status == "ACTIVE", "键盘商品未发布")
+            rule = await session.scalar(select(ExchangeRule).where(
+                ExchangeRule.tenant_id == TENANT.bytes, ExchangeRule.store_id == STORE.bytes,
+            ).order_by(ExchangeRule.version.desc()).limit(1))
+            require(
+                rule is not None and not rule.revoked and rule.window_days > 2
+                and "keyboard" in rule.product_types,
+                "当前规则不支持签收两天的键盘订单，请检查已发布规则",
+            )
+            manifest["rule"] = {"version": rule.version, "window_days": rule.window_days,
+                                "citation": rule.citation}
+            manifest["product_name"] = product.name
+            manifest["existing_ticket_count"] = await session.scalar(
+                select(func.count()).select_from(Ticket).where(
+                    Ticket.tenant_id == TENANT.bytes,
+                    Ticket.order_id == UUID(manifest["order"]["order_id"]).bytes,
+                )
+            )
+            await ensure_principal(
+                session, manifest["customer"], secret["customer_password"],
+                "CUSTOMER", CUSTOMER_PERMISSIONS,
+            )
+            await ensure_principal(
+                session, secret["service"], secret["service"]["password"],
+                "SERVICE", ("orders:sync",),
+            )
+            connector = await session.get(Connector, (TENANT.bytes, STORE.bytes))
+            if connector is None:
+                session.add(Connector(
+                    tenant_id=TENANT.bytes, store_id=STORE.bytes, source_system="oms",
+                    base_url=CONNECTOR_URL, credential_env=TOKEN_ENV, enabled=True,
+                ))
+                session.add(BusinessEvent(
+                    id=uuid4().bytes, tenant_id=TENANT.bytes,
+                    actor_id=UUID(secret["service"]["principal_id"]).bytes,
+                    aggregate_id=STORE.bytes, aggregate_version=1,
+                    event_type="connector.configured", request_id=str(uuid4()),
+                    payload={"credential_env": TOKEN_ENV, "source": "prepare_course_demo"},
+                    occurred_at=datetime.now(UTC).replace(tzinfo=None),
+                ))
+            else:
+                require(
+                    connector.source_system == "oms" and connector.base_url == CONNECTOR_URL
+                    and connector.credential_env == TOKEN_ENV and connector.enabled,
+                    "已有 OMS 连接器配置与课程环境不一致",
+                )
+        oms_sessions = async_sessionmaker(oms, expire_on_commit=False)
+        async with oms_sessions.begin() as session:
+            for name, can_fulfill in (
+                ("oms_platform_token", False), ("oms_fulfillment_token", True),
+            ):
+                token_hash = hashlib.sha256(secret[name].encode()).digest()
+                existing = await session.get(Credential, token_hash, with_for_update=True)
+                if existing is None:
+                    session.add(Credential(
+                        token_hash=token_hash, tenant_id=TENANT.bytes, store_id=STORE.bytes,
+                        can_fulfill=can_fulfill, enabled=True,
+                    ))
+                else:
+                    require(
+                        existing.tenant_id == TENANT.bytes and existing.store_id == STORE.bytes
+                        and existing.can_fulfill == can_fulfill and existing.enabled,
+                        "已有 OMS 凭据权限与课程配置不一致",
+                    )
+    finally:
+        await engine.dispose()
+        await oms.dispose()
+
+
+def configure_runtime(secret):
+    path = ROOT / ".work" / "runtime.env"
+    require(path.is_file(), "完整项目的 .work/runtime.env 不存在")
+    existing = dotenv_values(path).get(TOKEN_ENV)
+    require(
+        existing is None or existing == secret["oms_platform_token"],
+        "运行配置已存在不同的课程 OMS 凭据",
+    )
+    if existing is None:
+        set_key(path, TOKEN_ENV, secret["oms_platform_token"], quote_mode="always")
+
+
+async def request(client, method, url, *, token=None, body=None, key=None, allow_missing=False):
+    headers = {}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    if key:
+        headers["Idempotency-Key"] = key
+    response = await client.request(method, url, headers=headers, json=body)
+    if allow_missing and response.status_code == 404:
+        return None
+    response.raise_for_status()
+    return response.json()
+
+
+async def login(client, manifest, actor, password):
+    result = await request(client, "POST", manifest["api_url"] + "/openapi/v1/auth/sessions", body={
+        "tenant_id": str(TENANT), "username": actor["username"], "password": password,
+    })
+    return result["access_token"], time.monotonic() + result["expires_in"] - 60
+
+
+def verify_snapshot(manifest, snapshot):
+    expected = manifest["order"]
+    require(
+        snapshot["tenant_id"] == str(TENANT) and snapshot["store_id"] == str(STORE)
+        and snapshot["subject_id"] == manifest["customer"]["principal_id"]
+        and snapshot["order_id"] == expected["order_id"]
+        and snapshot["item_id"] == "keyboard-01" and snapshot["product_type"] == "keyboard"
+        and datetime.fromisoformat(snapshot["signed_at"]) == datetime.fromisoformat(
+            expected["signed_at"]),
+        "OMS 订单身份、商品或签收时间与课程记录不一致",
+    )
+    require(
+        snapshot["source_version"] >= 1 and 0 <= snapshot["available_quantity"] <= 1,
+        "OMS 订单版本或数量不符合课程订单约束",
+    )
+
+
+async def synchronize(client, manifest, secret, token):
+    snapshot = await request(
+        client, "GET", manifest["oms_url"] + "/orders/" + manifest["order"]["order_id"],
+        token=secret["oms_platform_token"],
+    )
+    verify_snapshot(manifest, snapshot)
+    body = {name: snapshot[name] for name in (
+        "store_id", "subject_id", "source_version", "item_id", "product_type",
+        "available_quantity", "signed_at",
+    )}
+    body["source_system"] = "oms"
+    # 每次可信观察使用新的幂等键，确保相同来源版本也更新 observed_at。
+    await request(
+        client, "PUT", manifest["api_url"] + "/internal/v1/orders/" + snapshot["order_id"],
+        token=token, body=body, key=f"course-sync-{uuid4().hex}",
+    )
+    manifest["last_sync_at"] = utc_now()
+    manifest["last_order_snapshot"] = snapshot
+    return snapshot
+
+
+async def prepare(settings, args):
+    async with asyncio.timeout(90):
+        manifest, secret = await asyncio.to_thread(prepare_files, args)
+        await bootstrap(settings, manifest, secret)
+        await asyncio.to_thread(configure_runtime, secret)
+        async with httpx.AsyncClient(timeout=10, trust_env=False) as client:
+            order_url = manifest["oms_url"] + "/orders/" + manifest["order"]["order_id"]
+            existing = await request(client, "GET", order_url,
+                                     token=secret["oms_platform_token"], allow_missing=True)
+            if existing is None:
+                body = {name: value for name, value in manifest["order"].items()
+                        if name != "order_id"}
+                body["subject_id"] = manifest["customer"]["principal_id"]
+                await request(client, "PUT", order_url,
+                              token=secret["oms_fulfillment_token"], body=body)
+            else:
+                verify_snapshot(manifest, existing)
+            service_token, _ = await login(
+                client, manifest, secret["service"], secret["service"]["password"],
+            )
+            snapshot = await synchronize(client, manifest, secret, service_token)
+            customer_token, _ = await login(
+                client, manifest, manifest["customer"], secret["customer_password"],
+            )
+            order = await request(
+                client, "GET", manifest["api_url"] + "/openapi/v1/orders/" + snapshot["order_id"],
+                token=customer_token,
+            )
+            require(
+                order["subject_id"] == manifest["customer"]["principal_id"]
+                and order["source_system"] == "oms"
+                and order["source_version"] == snapshot["source_version"]
+                and order["available_quantity"] == snapshot["available_quantity"],
+                "客户读取的订单与 OMS 当前订单不一致",
+            )
+            conversation = await request(
+                client, "POST", manifest["api_url"] + "/openapi/v1/conversations",
+                token=customer_token, body={"store_id": str(STORE)},
+                key=f"course-conversation-{manifest['customer']['principal_id']}",
+            )
+            manifest["conversation_id"] = conversation["conversation_id"]
+            unused = (
+                conversation["sequence"] == 0 and manifest["existing_ticket_count"] == 0
+                and snapshot["source_version"] == 1 and snapshot["available_quantity"] == 1
+            )
+            if manifest["status"] != "OMS_COMPLETED":
+                manifest["status"] = "READY" if unused else "IN_USE"
+            manifest["verification"] = {
+                "verified_at": utc_now(), "customer_order_read": True,
+                "conversation_sequence": conversation["sequence"],
+                "generation_model_calls": 0,
+            }
+        await asyncio.to_thread(save_manifest, manifest, activate=True)
+        return manifest
+
+
+async def sync_loop(args):
+    manifest = await asyncio.to_thread(load_manifest, args.session)
+    secret = await asyncio.to_thread(credentials)
+    duration = args.duration_seconds
+    deadline = time.monotonic() + duration
+    maximum = max(1, math.ceil(duration / args.interval_seconds))
+    token, expires = None, 0
+    count = 0
+    async with (
+        asyncio.timeout(duration + 35),
+        httpx.AsyncClient(timeout=10, trust_env=False) as client,
+    ):
+        for index in range(maximum):
+            if index and time.monotonic() >= deadline:
+                break
+            # 锁等待计入本次同步的 30 秒期限，短时间重叠的课程操作允许等待完成。
+            lock = FileLock(DATA / "course.lock", timeout=10, thread_local=False)
+            async with asyncio.timeout(30):
+                await asyncio.to_thread(lock.acquire)
+                try:
+                    manifest = await asyncio.to_thread(load_manifest, manifest["session"])
+                    if token is None or time.monotonic() >= expires:
+                        token, expires = await login(
+                            client, manifest, secret["service"], secret["service"]["password"],
+                        )
+                    await synchronize(client, manifest, secret, token)
+                    await asyncio.to_thread(save_manifest, manifest)
+                finally:
+                    await asyncio.to_thread(lock.release)
+            count += 1
+            print(json.dumps({
+                "session": manifest["session"], "sync_count": count,
+                "last_sync_at": manifest["last_sync_at"],
+                "source_version": manifest["last_order_snapshot"]["source_version"],
+                "available_quantity": manifest["last_order_snapshot"]["available_quantity"],
+            }, ensure_ascii=False), flush=True)
+            if index + 1 < maximum:
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    break
+                await asyncio.sleep(min(args.interval_seconds, remaining))
+    return {"session": manifest["session"], "sync_count": count, "generation_model_calls": 0}
+
+
+async def complete(settings, args):
+    async with asyncio.timeout(90):
+        manifest = await asyncio.to_thread(load_manifest, args.session)
+        secret = await asyncio.to_thread(credentials)
+        engine = create_engine(settings)
+        try:
+            sessions = async_sessionmaker(engine, expire_on_commit=False)
+            async with sessions() as session:
+                query = select(ExternalOperation).where(
+                    ExternalOperation.tenant_id == TENANT.bytes,
+                    ExternalOperation.store_id == STORE.bytes,
+                    ExternalOperation.order_id == UUID(manifest["order"]["order_id"]).bytes,
+                    ExternalOperation.status.in_(("ACCEPTED", "COMPLETED")),
+                )
+                if args.operation_id:
+                    query = query.where(ExternalOperation.id == args.operation_id.bytes)
+                operations = (await session.scalars(query)).all()
+                require(
+                    len(operations) == 1,
+                    "需要唯一的本课程已受理履约操作，请检查审批与履约进度",
+                )
+                operation_id = str(UUID(bytes=operations[0].id))
+                ticket_id = str(UUID(bytes=operations[0].ticket_id))
+        finally:
+            await engine.dispose()
+        async with httpx.AsyncClient(timeout=10, trust_env=False) as client:
+            url = manifest["oms_url"] + "/exchanges/" + operation_id
+            existing = await request(client, "GET", url, token=secret["oms_fulfillment_token"])
+            require(
+                existing["order_id"] == manifest["order"]["order_id"]
+                and existing["status"] in {"ACCEPTED", "COMPLETED"},
+                "OMS 操作不属于本课程订单或当前状态不能完成",
+            )
+            result = await request(client, "POST", url + "/complete",
+                                   token=secret["oms_fulfillment_token"])
+            require(result["status"] == "COMPLETED", "OMS 未返回履约完成状态")
+            manifest["completion"] = {**result, "ticket_id": ticket_id, "completed_at": utc_now()}
+            manifest["status"] = "OMS_COMPLETED"
+            service_token, _ = await login(
+                client, manifest, secret["service"], secret["service"]["password"],
+            )
+            await synchronize(client, manifest, secret, service_token)
+        await asyncio.to_thread(save_manifest, manifest)
+        return {
+            "session": manifest["session"], **manifest["completion"], "generation_model_calls": 0,
+        }
+
+
+async def locked_command(settings, args):
+    # 跨线程共享同一个锁状态；准备与完成的 90 秒期限包含最长 10 秒的锁等待。
+    lock = FileLock(DATA / "course.lock", timeout=10, thread_local=False)
+    async with asyncio.timeout(90):
+        await asyncio.to_thread(lock.acquire)
+        try:
+            if args.command == "prepare":
+                return await prepare(settings, args)
+            return await complete(settings, args)
+        finally:
+            await asyncio.to_thread(lock.release)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="准备、同步和完成完整项目的课程演示订单")
+    commands = parser.add_subparsers(dest="command", required=True)
+    prepare_parser = commands.add_parser("prepare", help="创建本次课程客户、OMS 订单与空会话")
+    prepare_parser.add_argument("--session", type=session_name)
+    prepare_parser.add_argument("--api-url", type=validate_url, default="http://127.0.0.1:28000")
+    prepare_parser.add_argument("--oms-url", type=validate_url, default="http://127.0.0.1:28100")
+    sync_parser = commands.add_parser("sync", help="从 OMS 读取并同步当前课程订单")
+    sync_parser.add_argument("--session", type=session_name)
+    sync_parser.add_argument("--duration-seconds", type=int, default=0)
+    sync_parser.add_argument("--interval-seconds", type=int, default=60)
+    complete_parser = commands.add_parser("complete", help="使用仓储身份登记本课程履约完成")
+    complete_parser.add_argument("--session", type=session_name)
+    complete_parser.add_argument("--operation-id", type=UUID)
+    args = parser.parse_args()
+    settings = Settings()
+    require(settings.environment in {"development", "test"}, "课程工具仅允许 development 或 test")
+    require(Path.cwd().resolve() == ROOT, "请在完整项目根目录执行课程工具")
+    DATA.mkdir(parents=True, exist_ok=True)
+    if args.command == "sync":
+        require(0 <= args.duration_seconds <= 43200, "同步时长须为 0 至 43200 秒")
+        require(30 <= args.interval_seconds <= 240, "同步间隔须为 30 至 240 秒")
+        result = asyncio.run(sync_loop(args))
+    else:
+        result = asyncio.run(locked_command(settings, args))
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

scripts/start_course_demo.ps1

@@ -0,0 +1,74 @@
+#Requires -Version 7.0
+param(
+    [ValidateSet('Prepare', 'Stop')]
+    [string] $Action = 'Prepare',
+    [ValidatePattern('^[a-zA-Z0-9][a-zA-Z0-9_-]{0,47}$')]
+    [string] $Session,
+    [ValidateRange(1, 180)]
+    [int] $DurationMinutes = 180
+)
+
+$ErrorActionPreference = 'Stop'
+$demoRoot = (Resolve-Path (Join-Path $PSScriptRoot '..')).Path
+$demoDirectory = Join-Path $demoRoot '.work/course-demo'
+$demoScript = Join-Path $PSScriptRoot 'prepare_course_demo.py'
+$demoPython = Join-Path $demoRoot '.venv/Scripts/python.exe'
+$demoProcessFile = Join-Path $demoDirectory 'sync-process.json'
+
+function Stop-DemoSync {
+    if (-not (Test-Path -LiteralPath $demoProcessFile)) { return }
+    $record = Get-Content -LiteralPath $demoProcessFile -Raw | ConvertFrom-Json
+    $process = Get-Process -Id $record.pid -ErrorAction SilentlyContinue
+    if ($null -eq $process) { return }
+    $details = Get-CimInstance Win32_Process -Filter "ProcessId = $($record.pid)"
+    # 仅停止本入口登记且创建时间一致的同步进程，防止 PID 被其他程序复用。
+    if ($process.StartTime.ToUniversalTime().Ticks -ne [long]$record.started_ticks -or
+        $null -eq $details.CommandLine -or -not $details.CommandLine.Contains($demoScript)) {
+        throw '订单同步进程身份检查失败，请检查进程记录。'
+    }
+    $process.Kill($true)
+    if (-not $process.WaitForExit(10000)) { throw '订单同步进程未在十秒内停止。' }
+}
+
+Push-Location $demoRoot
+try {
+    if ($Action -eq 'Stop') {
+        Stop-DemoSync
+        Write-Output '课程订单同步已停止，课程数据保留。'
+        return
+    }
+    if (-not (Test-Path -LiteralPath $demoPython)) { throw '请准备项目 Python 环境。' }
+    if (-not $Session) { $Session = Get-Date -Format 'yyyyMMdd-HHmmss-fff' }
+    $env:PYTHONUTF8 = '1'
+    $env:PYTHONIOENCODING = 'utf-8'
+    & $demoPython $demoScript prepare --session $Session
+    if ($LASTEXITCODE -ne 0) { throw '课程订单准备失败。' }
+
+    # 履约进程读取已登记的专用 OMS 凭据；Compose 保留现有数据库与其他服务。
+    $env:ZHIFU_RUNTIME_ENV = (Resolve-Path '.work/runtime.env').Path
+    $env:ZHIFU_MODEL_PATH = (Resolve-Path '.work/models').Path
+    docker compose -f compose.runtime.yaml up -d --no-deps fulfillment
+    if ($LASTEXITCODE -ne 0) { throw '履约进程配置加载失败。' }
+
+    Stop-DemoSync
+    $durationSeconds = $DurationMinutes * 60
+    $arguments = @(('"' + $demoScript + '"'), 'sync', '--session', $Session,
+        '--duration-seconds', "$durationSeconds", '--interval-seconds', '60')
+    $stdout = Join-Path $demoDirectory "$Session-sync.stdout.log"
+    $stderr = Join-Path $demoDirectory "$Session-sync.stderr.log"
+    $process = Start-Process -FilePath $demoPython -ArgumentList $arguments `
+        -WorkingDirectory $demoRoot -WindowStyle Hidden -PassThru `
+        -RedirectStandardOutput $stdout -RedirectStandardError $stderr
+    @{pid = $process.Id; started_ticks = $process.StartTime.ToUniversalTime().Ticks;
+        session = $Session; duration_seconds = $durationSeconds; interval_seconds = 60;
+        stdout = $stdout; stderr = $stderr} |
+        ConvertTo-Json | Set-Content -LiteralPath $demoProcessFile -Encoding utf8
+    if ($process.WaitForExit(3000)) {
+        Get-Content -LiteralPath $stderr
+        throw '订单同步进程提前结束。'
+    }
+    Write-Output "课程 $Session 已准备；每 60 秒从 OMS 同步一次，最长 $DurationMinutes 分钟。"
+    Write-Output "本次账户及订单：$demoDirectory/active.json"
+} finally {
+    Pop-Location
+}

src/zhifu/core/api_docs.py

@@ -84,6 +84,7 @@ class OrderSummary(BaseModel):
     order_id: UUID = Field(description="订单标识")
     source_version: int = Field(description="OMS 来源版本")
     item_id: str = Field(description="订单商品标识")
+    available_quantity: int = Field(ge=0, description="最近一次 OMS 同步确认的可售后数量")
     store_name: str = Field(description="所属店铺名称")
     product_name: str | None = Field(description="同店铺商品目录名称，未收录时为 null")
     image_url: str | None = Field(description="同店铺商品图片，未收录时为 null")
@@ -97,7 +98,6 @@ class OrderSummary(BaseModel):
 class OrderView(OrderSummary):
     subject_id: UUID = Field(description="订单所属客户；客服用于核对当前会话客户")
     item_id: str = Field(description="订单商品明细标识")
-    available_quantity: int = Field(description="最近同步的可售后数量")
     source_system: str = Field(description="来源系统标识")
     observed_at: str = Field(description="最近同步时间，UTC ISO 8601 字符串，当前不带时区后缀")
 

src/zhifu/modules/storefront/orders.py

@@ -11,6 +11,8 @@ async def order_summary(session, order):
     return {
         "order_id": str(UUID(bytes=order.id)), "source_version": order.source_version,
         "item_id": order.item_id, "store_name": store.name,
+        # 列表中的可售后数量使用最近一次 OMS 同步保存的订单快照。
+        "available_quantity": order.available_quantity,
         "product_name": product.name if product else None,
         "image_url": product.image_url if product else None,
         "specifications": product.specifications if product else {},

tests/test_storefront.py

@@ -72,8 +72,24 @@ async def test_catalog_order_metadata_and_visibility(scenario):  # noqa: F811
     assert order.json()["product_name"] == "办公键盘"
     assert order.json()["specifications"] == {"款式": "标准款"}
     assert order.json()["fulfillment_status"] == "SIGNED"
+    assert order.json()["available_quantity"] == 1
+    orders_path = f"/openapi/v1/orders?store_id={scenario['store']}"
+    listed = await scenario["call"]("GET", orders_path)
+    assert listed.status_code == 200, listed.text
+    assert listed.json()["items"][0]["order_id"] == str(scenario["order_id"])
+    assert listed.json()["items"][0]["available_quantity"] == 1
     foreign = await scenario["call"]("GET", f"/openapi/v1/orders/{scenario['order_id']}", "foreign")
     assert foreign.status_code == 404
+    # 来源更新到零时，列表与详情都应显示最新可售后数量。
+    synchronized = await scenario["call"](
+        "PUT", f"/internal/v1/orders/{scenario['order_id']}", "connector",
+        {**scenario["source"], "source_version": 2, "available_quantity": 0},
+    )
+    assert synchronized.status_code == 200, synchronized.text
+    listed = await scenario["call"]("GET", orders_path)
+    order = await scenario["call"]("GET", f"/openapi/v1/orders/{scenario['order_id']}")
+    assert listed.json()["items"][0]["available_quantity"] == 0
+    assert order.json()["available_quantity"] == 0
 
 
 async def test_handoff_transaction_retry_and_takeover(scenario):  # noqa: F811