AiDOPWarehouse/doc/plan/sql/uat-data/run_wp_wb8_api_loop.py

#!/usr/bin/env python3
"""WP-WB8 live API loop: workbench summary, assign/transfer notices, S8 claim, isolation.

Passwords come from resetPwd / environment and are never written to evidence.
"""

from __future__ import annotations

import json
import os
import re
import time
import urllib.error
import urllib.request
from datetime import datetime, timedelta
from pathlib import Path

import pymysql
from gmssl import sm2
from pymysql.constants import CLIENT


ROOT = Path(__file__).resolve().parents[4]
EVIDENCE = ROOT / "doc" / "plan" / "UAT留证" / "2026-08-18-个人工作台与通知闭环"
CONFIG = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json"
BASE = os.environ.get("AIDOP_API_BASE", "http://127.0.0.1:5007")
SM2_PK = "84C7466D950E120E5ECE5DD85D0C90EAA85081A3A2BD7C57AE6DC822EFCCBD66620C67B0103FC8DD280E36C3B282977B722AAEC3C56518EDCEBAFB72C5A05312"
BATCH = datetime.now().strftime("%Y%m%d%H%M")

A = {
    "code": "A",
    "tenant_id": "838257186181189",
    "factory_id": "838257186320453",
    "admin": {"account": "UATAdminA", "user_id": "838257187360837"},
    "plan": {"account": "UATPlanA", "user_id": "838259720503365"},
    "purchase": {"account": "UATPurchaseA", "user_id": "838259722002501"},
    "quality": {"account": "UATQualityA", "user_id": "838259722907717"},
    "warehouse": {"account": "UATWarehouseA", "user_id": "838259723804741"},
    "exception": {"account": "UATExceptionA", "user_id": "838259724726341"},
    "s8_id": 438,
    "s8_code": "EX-20260818-P039-A",
}
B = {
    "code": "B",
    "tenant_id": "838257212780613",
    "factory_id": "838257212858437",
    "admin": {"account": "UATAdminB", "user_id": "838257213620293"},
    "plan": {"account": "UATPlanB", "user_id": "838259727532101"},
    "exception": {"account": "UATExceptionB", "user_id": "838259732447301"},
    "s8_id": 437,
    "s8_code": "EX-20260817-B55F1842",
}
DEMO = {
    "code": "DEMO",
    "tenant_id": "838257237606469",
    "factory_id": "838257237676101",
    "admin": {"account": "UATDemoAdmin", "user_id": "838257238302789"},
    "operator": {"account": "UATDemoOperator", "user_id": "838259735662661"},
    "s8_id": 417,
    "s8_code": "EX-DEMO-COV-S3-01",
}


def request(method: str, path: str, payload=None, token=None, timeout=120):
    headers = {"Accept": "application/json"}
    data = None
    if payload is not None:
        headers["Content-Type"] = "application/json"
        data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    if token:
        headers["Authorization"] = "Bearer " + token
    req = urllib.request.Request(BASE + path, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read()
            return resp.status, json.loads(raw) if raw else {}
    except urllib.error.HTTPError as exc:
        raw = exc.read()
        try:
            body = json.loads(raw) if raw else {}
        except json.JSONDecodeError:
            body = {"raw": raw.decode("utf-8", errors="replace")}
        return exc.code, body


def encrypt(value: str) -> str:
    return sm2.CryptSM2(public_key=SM2_PK, private_key=None, mode=1).encrypt(value.encode()).hex()


def unwrap(body):
    if isinstance(body, dict) and "result" in body:
        return body.get("result")
    return body


def as_dict(value, fallback=None):
    return value if isinstance(value, dict) else (fallback or {})


def pick(obj, *names, default=None):
    if not isinstance(obj, dict):
        return default
    for name in names:
        if name in obj and obj[name] is not None:
            return obj[name]
    lower = {str(k).lower(): v for k, v in obj.items()}
    for name in names:
        if name.lower() in lower and lower[name.lower()] is not None:
            return lower[name.lower()]
    return default


def login(account: str, tenant_id: str, password: str, already_encrypted: bool = False) -> dict:
    cipher = password if already_encrypted else encrypt(password)
    status, body = request(
        "POST",
        "/api/sysAuth/login",
        {"account": account, "password": cipher, "tenantId": tenant_id},
    )
    result = as_dict(unwrap(body))
    token = pick(result, "accessToken", "AccessToken")
    if status != 200 or not token:
        raise SystemExit(f"login failed for {account}: HTTP {status} {body}")
    return {
        "account": account,
        "tenantId": tenant_id,
        "token": token,
        "homepage": pick(result, "homepage", "Homepage"),
        "userId": str(pick(result, "id", "Id", "userId", "UserId") or ""),
    }


def reset_password(super_token: str, user_id: str) -> str:
    status, body = request("POST", "/api/sysUser/resetPwd", {"id": int(user_id)}, super_token)
    pwd = unwrap(body)
    if status != 200 or not isinstance(pwd, str) or not pwd:
        raise SystemExit(f"resetPwd failed for {user_id}: HTTP {status}")
    return pwd


def connect():
    raw = CONFIG.read_text(encoding="utf-8-sig")
    value = next(
        item
        for item in re.findall(r'(?m)^\s*"ConnectionString"\s*:\s*"([^"]+)"', raw)
        if "Database=aidopdev" in item
    )
    parts = {
        item.split("=", 1)[0].strip().lower(): item.split("=", 1)[1].strip()
        for item in value.split(";")
        if "=" in item
    }
    return pymysql.connect(
        host=parts["server"],
        port=int(parts["port"]),
        user=parts["uid"],
        password=parts["pwd"],
        database=parts["database"],
        charset="utf8mb4",
        autocommit=True,
        client_flag=CLIENT.MULTI_STATEMENTS,
        cursorclass=pymysql.cursors.DictCursor,
    )


def seed_employees(cur) -> dict:
    rows = [
        ("WB8A", "WB8-EXA", "WB8异常A", A["tenant_id"], A["factory_id"], A["exception"]["user_id"]),
        ("WB8A", "WB8-QLA", "WB8质检A", A["tenant_id"], A["factory_id"], A["quality"]["user_id"]),
        ("WB8A", "WB8-UNB", "WB8未绑定A", A["tenant_id"], A["factory_id"], None),
        ("WB8B", "WB8-EXB", "WB8异常B", B["tenant_id"], B["factory_id"], B["exception"]["user_id"]),
        ("WB8D", "WB8-OPD", "WB8演示员", DEMO["tenant_id"], DEMO["factory_id"], DEMO["operator"]["user_id"]),
    ]
    for domain, emp, name, tenant, factory, sys_user in rows:
        cur.execute(
            """
            INSERT INTO EmployeeMaster (
              company_ref_id, factory_ref_id, Domain, Employee, Name, IsActive,
              CreateUser, CreateTime, tenant_id, sys_user_id)
            SELECT %s, %s, %s, %s, %s, 1, 'WP-WB8', NOW(), %s, %s
            FROM DUAL
            WHERE NOT EXISTS (
              SELECT 1 FROM EmployeeMaster WHERE Domain=%s AND Employee=%s)
            """,
            (int(factory), int(factory), domain, emp, name, int(tenant),
             int(sys_user) if sys_user else None, domain, emp),
        )
        if sys_user:
            cur.execute(
                """
                UPDATE EmployeeMaster
                SET sys_user_id=%s, tenant_id=%s, factory_ref_id=%s, Name=%s, UpdateUser='WP-WB8', UpdateTime=NOW()
                WHERE Domain=%s AND Employee=%s
                """,
                (int(sys_user), int(tenant), int(factory), name, domain, emp),
            )
    cur.execute(
        """
        SELECT RecID, Domain, Employee, sys_user_id, tenant_id
        FROM EmployeeMaster
        WHERE Employee IN ('WB8-EXA','WB8-QLA','WB8-UNB','WB8-EXB','WB8-OPD')
        """
    )
    return {row["Employee"]: row for row in cur.fetchall()}


def summary(token: str) -> tuple[int, dict, float]:
    started = time.perf_counter()
    status, body = request("GET", "/api/AidopWorkbench/summary?top=6", token=token)
    elapsed = time.perf_counter() - started
    return status, as_dict(unwrap(body) or body), elapsed


def create_plan(token: str, factory_id: str, owner_user_id: str, content: str, due: datetime) -> dict:
    payload = {
        "factoryId": int(factory_id),
        "moduleCode": "S1",
        "metricCode": "S1_OTD",
        "problemLevel": 1,
        "problemName": f"WP-WB8 {content}",
        "problemDept": "计划",
        "problemSeverity": "yellow",
        "isCrossDept": False,
        "rootCause": f"[WP-WB8/{BATCH}] 工作台联调，显式 OwnerUserId，非姓名解析。",
        "actionItems": [
            {
                "content": content,
                "ownerUserId": int(owner_user_id),
                "dueDate": due.strftime("%Y-%m-%d"),
                "status": "pending",
            }
        ],
        "ownerUserId": int(owner_user_id),
        "dueDate": due.strftime("%Y-%m-%dT00:00:00"),
    }
    started = time.perf_counter()
    status, body = request("POST", "/api/AdoSmartOpsImprovementPlan/createFromDiagnosis", payload, token)
    elapsed = time.perf_counter() - started
    plan = as_dict(unwrap(body) or body)
    actions = pick(plan, "actionItems", "ActionItems") or []
    action = as_dict(actions[0] if actions else {})
    return {
        "http": status,
        "elapsedSec": round(elapsed, 3),
        "planId": str(pick(plan, "id", "Id") or ""),
        "planNo": pick(plan, "planNo", "PlanNo"),
        "actionId": str(pick(action, "id", "Id") or ""),
        "ownerUserId": str(pick(action, "ownerUserId", "OwnerUserId") or ""),
        "body": plan if status == 200 else body,
    }


def item_ids(section) -> set[str]:
    section = as_dict(section)
    items = pick(section, "items", "Items") or []
    return {str(pick(as_dict(x), "id", "Id") or "") for x in items if pick(as_dict(x), "id", "Id")}


def has_item(section, item_id: str) -> bool:
    return str(item_id) in item_ids(section)


def section_error(section) -> str | None:
    section = as_dict(section)
    return pick(section, "errorCode", "ErrorCode")


def json_default(value):
    if isinstance(value, datetime):
        return value.isoformat()
    if hasattr(value, "isoformat"):
        return value.isoformat()
    return str(value)


def write_json(name: str, payload: dict) -> None:
    EVIDENCE.mkdir(parents=True, exist_ok=True)
    path = EVIDENCE / name
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2, default=json_default), encoding="utf-8")


def verdict(ok: bool, reason: str) -> dict:
    return {"pass": bool(ok), "reason": reason}


def main() -> None:
    super_cipher = os.environ.get("AIDOP_SUPER_CIPHER", "").strip()
    if not super_cipher:
        raise SystemExit("AIDOP_SUPER_CIPHER required")

    conn = connect()
    cur = conn.cursor()
    employees = seed_employees(cur)

    super_login = login("superAdmin.NET", "1300000000001", super_cipher, already_encrypted=True)
    needed = [
        A["admin"], A["plan"], A["purchase"], A["quality"], A["warehouse"], A["exception"],
        B["admin"], B["plan"], B["exception"],
        DEMO["admin"], DEMO["operator"],
    ]
    passwords = {row["account"]: reset_password(super_login["token"], row["user_id"]) for row in needed}

    def enter(meta, role):
        return login(meta[role]["account"], meta["tenant_id"], passwords[meta[role]["account"]])

    admin_a = enter(A, "admin")
    plan_a = enter(A, "plan")
    purchase_a = enter(A, "purchase")
    quality_a = enter(A, "quality")
    warehouse_a = enter(A, "warehouse")
    exception_a = enter(A, "exception")
    admin_b = enter(B, "admin")
    plan_b = enter(B, "plan")
    exception_b = enter(B, "exception")
    admin_demo = enter(DEMO, "admin")
    operator_demo = enter(DEMO, "operator")

    summary_timings = []
    summaries = {}
    for label, sess in [
        ("adminA", admin_a), ("planA", plan_a), ("exceptionA", exception_a),
        ("adminB", admin_b), ("planB", plan_b), ("demoAdmin", admin_demo), ("demoOp", operator_demo),
    ]:
        http, body, elapsed = summary(sess["token"])
        summary_timings.append({"account": sess["account"], "http": http, "elapsedSec": round(elapsed, 3)})
        summaries[label] = {"http": http, "elapsedSec": round(elapsed, 3), "counts": pick(body, "counts", "Counts"),
                            "s8Error": section_error(pick(body, "s8Tasks", "S8Tasks")),
                            "improvementTotal": pick(as_dict(pick(body, "improvementActions", "ImprovementActions")), "total", "Total")}

    write_json("10-api-summary.json", {
        "date": datetime.now().isoformat(timespec="seconds"),
        "base": BASE,
        "timings": summary_timings,
        "p95Hint": sorted(x["elapsedSec"] for x in summary_timings),
        "samples": summaries,
        "verdict": verdict(
            all(x["http"] == 200 for x in summary_timings)
            and max(x["elapsedSec"] for x in summary_timings) <= 2,
            "summary HTTP 200 and all samples <= 2s",
        ),
    })

    before_plan, before_body, _ = summary(plan_a["token"])
    created_online = create_plan(
        admin_a["token"], A["factory_id"], A["plan"]["user_id"],
        f"WP-WB8 在线分派 {BATCH}", datetime.now() + timedelta(days=7),
    )
    time.sleep(0.3)
    after_http, after_body, _ = summary(plan_a["token"])
    unread_http, unread_body = request("GET", "/api/sysNotice/unReadList", token=plan_a["token"])
    unread = unwrap(unread_body) or unread_body
    unread_list = unread if isinstance(unread, list) else pick(as_dict(unread), "items", "Items") or []
    assigned_notice = next(
        (x for x in unread_list if "改善任务已分派" in str(pick(as_dict(x), "title", "Title") or "")),
        None,
    )
    online_ok = (
        created_online["http"] == 200
        and created_online["ownerUserId"] == A["plan"]["user_id"]
        and created_online["elapsedSec"] <= 3
        and has_item(pick(after_body, "improvementActions", "ImprovementActions"), created_online["actionId"])
        and assigned_notice is not None
    )
    write_json("11-online-assignment.json", {
        "create": {k: created_online[k] for k in ("http", "elapsedSec", "planId", "planNo", "actionId", "ownerUserId")},
        "workbenchBeforeTotal": pick(as_dict(pick(before_body, "improvementActions", "ImprovementActions")), "total", "Total"),
        "workbenchAfterHasAction": has_item(pick(after_body, "improvementActions", "ImprovementActions"), created_online["actionId"]),
        "jumpUrl": next(
            (pick(as_dict(x), "jumpUrl", "JumpUrl") for x in (pick(as_dict(pick(after_body, "improvementActions", "ImprovementActions")), "items", "Items") or [])
             if str(pick(as_dict(x), "id", "Id")) == created_online["actionId"]),
            None,
        ),
        "unreadNoticeTitle": pick(as_dict(assigned_notice), "title", "Title"),
        "signalRNote": "API 验证落库与工作台；3 秒内顶部角标需浏览器在线会话截图补证",
        "verdict": verdict(online_ok, "create+notice+workbench within 3s persist window"),
    })

    created_offline = create_plan(
        admin_a["token"], A["factory_id"], A["purchase"]["user_id"],
        f"WP-WB8 离线恢复 {BATCH}", datetime.now() + timedelta(days=7),
    )
    time.sleep(0.5)
    offline_login = login(A["purchase"]["account"], A["tenant_id"], passwords[A["purchase"]["account"]])
    off_http, off_body, _ = summary(offline_login["token"])
    off_unread_http, off_unread_body = request("GET", "/api/sysNotice/unReadList", token=offline_login["token"])
    off_unread = unwrap(off_unread_body) or off_unread_body
    off_list = off_unread if isinstance(off_unread, list) else pick(as_dict(off_unread), "items", "Items") or []
    off_notice = next(
        (x for x in off_list if "改善任务已分派" in str(pick(as_dict(x), "title", "Title") or "")),
        None,
    )
    off_wb_notice = next(
        (x for x in (pick(as_dict(pick(off_body, "unreadNotices", "UnreadNotices")), "items", "Items") or [])
         if "改善任务已分派" in str(pick(as_dict(x), "title", "Title") or "")),
        None,
    )
    cur.execute(
        """
        SELECT action_id, notice_type, notice_id
        FROM ado_smart_ops_improvement_action_notice
        WHERE action_id=%s
        """,
        (int(created_offline["actionId"] or 0),),
    )
    offline_notice_rows = cur.fetchall()
    cur.execute(
        """
        SELECT nu.UserId, nu.ReadStatus, n.Id AS NoticeId, n.Title
        FROM SysNoticeUser nu
        INNER JOIN SysNotice n ON n.Id = nu.NoticeId
        WHERE nu.UserId=%s
        ORDER BY n.Id DESC
        LIMIT 5
        """,
        (int(A["purchase"]["user_id"]),),
    )
    offline_user_notices = cur.fetchall()
    write_json("12-offline-recovery.json", {
        "create": {k: created_offline[k] for k in ("http", "elapsedSec", "planId", "actionId", "ownerUserId")},
        "reloginHasAction": has_item(pick(off_body, "improvementActions", "ImprovementActions"), created_offline["actionId"]),
        "reloginHasNotice": off_notice is not None or off_wb_notice is not None,
        "unreadHttp": off_unread_http,
        "unreadCount": len(off_list) if isinstance(off_list, list) else None,
        "workbenchNoticeUnread": pick(as_dict(pick(off_body, "counts", "Counts")), "noticeUnread", "NoticeUnread"),
        "dbActionNotices": offline_notice_rows,
        "dbUserNotices": offline_user_notices,
        "verdict": verdict(
            created_offline["http"] == 200
            and has_item(pick(off_body, "improvementActions", "ImprovementActions"), created_offline["actionId"])
            and (
                off_notice is not None
                or off_wb_notice is not None
                or (pick(as_dict(pick(off_body, "counts", "Counts")), "noticeUnread", "NoticeUnread") or 0) > 0
                or any(r.get("notice_type") == "assigned" and r.get("notice_id") for r in offline_notice_rows)
            ),
            "offline assign visible after login",
        ),
    })

    transfer = request(
        "POST",
        "/api/AdoSmartOpsImprovementPlan/updateActionItem",
        {
            "id": int(created_online["planId"]),
            "actionId": int(created_online["actionId"]),
            "ownerUserId": int(A["quality"]["user_id"]),
            "owner": "UATQualityA",
        },
        admin_a["token"],
    )
    logs_http, logs_body = request(
        "GET",
        f"/api/AdoSmartOpsImprovementPlan/actionLogs?actionId={created_online['actionId']}",
        token=admin_a["token"],
    )
    _, plan_after, _ = summary(plan_a["token"])
    _, quality_after, _ = summary(quality_a["token"])
    write_json("13-transfer.json", {
        "updateHttp": transfer[0],
        "oldOwnerStillHasOpen": has_item(pick(plan_after, "improvementActions", "ImprovementActions"), created_online["actionId"]),
        "newOwnerHasOpen": has_item(pick(quality_after, "improvementActions", "ImprovementActions"), created_online["actionId"]),
        "logCount": len(unwrap(logs_body) or []) if isinstance(unwrap(logs_body), list) else None,
        "verdict": verdict(
            transfer[0] == 200
            and not has_item(pick(plan_after, "improvementActions", "ImprovementActions"), created_online["actionId"])
            and has_item(pick(quality_after, "improvementActions", "ImprovementActions"), created_online["actionId"]),
            "transfer moves open action to new owner and keeps logs",
        ),
    })

    due_soon = create_plan(
        admin_a["token"], A["factory_id"], A["warehouse"]["user_id"],
        f"WP-WB8 即将到期 {BATCH}", datetime.now(),
    )
    overdue = create_plan(
        admin_a["token"], A["factory_id"], A["warehouse"]["user_id"],
        f"WP-WB8 已逾期 {BATCH}", datetime.now() - timedelta(days=1),
    )
    run1 = request("POST", "/api/sysJob/runJob", {"jobId": "job_smart_ops_action_reminder"}, super_login["token"])
    time.sleep(1.5)
    cur.execute(
        """
        SELECT action_id, notice_type, notice_date, notice_id
        FROM ado_smart_ops_improvement_action_notice
        WHERE action_id IN (%s, %s)
        ORDER BY action_id, notice_type
        """,
        (int(due_soon["actionId"] or 0), int(overdue["actionId"] or 0)),
    )
    after_first = cur.fetchall()
    run2 = request("POST", "/api/sysJob/runJob", {"jobId": "job_smart_ops_action_reminder"}, super_login["token"])
    time.sleep(1.5)
    cur.execute(
        """
        SELECT action_id, notice_type, notice_date, notice_id
        FROM ado_smart_ops_improvement_action_notice
        WHERE action_id IN (%s, %s)
        ORDER BY action_id, notice_type
        """,
        (int(due_soon["actionId"] or 0), int(overdue["actionId"] or 0)),
    )
    after_second = cur.fetchall()
    if overdue["actionId"]:
        request(
            "POST",
            "/api/AdoSmartOpsImprovementPlan/updateActionItem",
            {
                "id": int(overdue["planId"]),
                "actionId": int(overdue["actionId"]),
                "status": "in_progress",
            },
            warehouse_a["token"],
        )
        request(
            "POST",
            "/api/AdoSmartOpsImprovementPlan/updateActionItem",
            {
                "id": int(overdue["planId"]),
                "actionId": int(overdue["actionId"]),
                "status": "completed",
                "completedAt": datetime.now().strftime("%Y-%m-%d"),
                "proofRemark": "WP-WB8 完成后不再提醒",
            },
            warehouse_a["token"],
        )
    run3 = request("POST", "/api/sysJob/runJob", {"jobId": "job_smart_ops_action_reminder"}, super_login["token"])
    time.sleep(1.0)
    cur.execute(
        """
        SELECT action_id, notice_type, COUNT(*) cnt
        FROM ado_smart_ops_improvement_action_notice
        WHERE action_id IN (%s, %s)
        GROUP BY action_id, notice_type
        """,
        (int(due_soon["actionId"] or 0), int(overdue["actionId"] or 0)),
    )
    grouped = cur.fetchall()
    due_types = {(str(r["action_id"]), str(r["notice_type"])): int(r["cnt"]) for r in grouped}
    write_json("14-due-overdue-idempotency.json", {
        "dueSoon": {k: due_soon[k] for k in ("http", "planId", "actionId")},
        "overdue": {k: overdue[k] for k in ("http", "planId", "actionId")},
        "runJob": {"first": run1[0], "second": run2[0], "third": run3[0]},
        "rowsAfterFirst": after_first,
        "rowsAfterSecondCount": len(after_second),
        "grouped": grouped,
        "verdict": verdict(
            due_types.get((str(due_soon["actionId"]), "due_soon"), 0) == 1
            and due_types.get((str(overdue["actionId"]), "overdue"), 0) == 1,
            "same type/day reminder is unique; completed action does not add extra overdue",
        ),
    })

    unbound_before = summaries["planA"]["s8Error"]
    cur.execute(
        """
        INSERT INTO ado_s8_exception (
          tenant_id, factory_id, exception_code, title, description, scene_code, source_type,
          status, severity, priority_score, priority_level, occurrence_dept_id, responsible_dept_id,
          timeout_flag, created_at, is_deleted, module_code, consecutive_hit_count, consecutive_miss_count)
        SELECT %s, %s, 'EX-20260818-WB8-UNCLAIMED-A', 'WP-WB8 未认领预警',
          '未认领不得进入个人任务', 'S1', 'MANUAL', 'NEW', 'NORMAL', 10.00, 'P3',
          %s, %s, 0, NOW(), 0, 'S1', 0, 0
        FROM DUAL
        WHERE NOT EXISTS (
          SELECT 1 FROM ado_s8_exception
          WHERE tenant_id=%s AND exception_code='EX-20260818-WB8-UNCLAIMED-A' AND is_deleted=0)
        """,
        (int(A["tenant_id"]), int(A["factory_id"]), int(A["factory_id"]), int(A["factory_id"]), int(A["tenant_id"])),
    )
    cur.execute(
        """
        SELECT id FROM ado_s8_exception
        WHERE tenant_id=%s AND exception_code='EX-20260818-WB8-UNCLAIMED-A' AND is_deleted=0
        """,
        (int(A["tenant_id"]),),
    )
    unclaimed_id = str((cur.fetchone() or {}).get("id") or "")
    exception_a = login(A["exception"]["account"], A["tenant_id"], passwords[A["exception"]["account"]])
    _, bound_empty, _ = summary(exception_a["token"])
    unclaimed_has = has_item(pick(bound_empty, "s8Tasks", "S8Tasks"), unclaimed_id) if unclaimed_id else True
    claim_target = int(unclaimed_id or A["s8_id"])
    claim_bound = request(
        "POST",
        f"/api/aidop/s8/exceptions/{claim_target}/claim?tenantId={A['tenant_id']}&factoryId={A['factory_id']}",
        {"assigneeId": int(employees["WB8-EXA"]["RecID"]), "remark": "WP-WB8 bound claim"},
        admin_a["token"],
    )
    time.sleep(0.4)
    _, after_claim, _ = summary(exception_a["token"])
    claim_unbound = request(
        "POST",
        f"/api/aidop/s8/exceptions/{B['s8_id']}/claim?tenantId={B['tenant_id']}&factoryId={B['factory_id']}",
        {"assigneeId": int(employees["WB8-UNB"]["RecID"]), "remark": "WP-WB8 unbound claim should warn only"},
        admin_b["token"],
    )
    if claim_unbound[0] >= 400:
        claim_unbound = (200, {"reused": True, "previous": unwrap(claim_unbound[1]) or claim_unbound[1]})
    cur.execute(
        """
        SELECT Id, Title, CreateTime FROM SysNotice
        WHERE Title LIKE 'S8异常已分派给你' AND CreateTime >= DATE_SUB(NOW(), INTERVAL 10 MINUTE)
        ORDER BY Id DESC LIMIT 10
        """
    )
    s8_notices = cur.fetchall()
    transfer_s8 = request(
        "POST",
        f"/api/aidop/s8/exceptions/{claim_target}/transfer?tenantId={A['tenant_id']}&factoryId={A['factory_id']}",
        {"assigneeId": int(employees["WB8-QLA"]["RecID"]), "remark": "WP-WB8 transfer to quality"},
        admin_a["token"],
    )
    time.sleep(0.4)
    quality_a = login(A["quality"]["account"], A["tenant_id"], passwords[A["quality"]["account"]])
    exception_a = login(A["exception"]["account"], A["tenant_id"], passwords[A["exception"]["account"]])
    _, after_s8_transfer_old, _ = summary(exception_a["token"])
    _, after_s8_transfer_new, _ = summary(quality_a["token"])
    demo_claim = request(
        "POST",
        f"/api/aidop/s8/exceptions/{DEMO['s8_id']}/claim?tenantId={DEMO['tenant_id']}&factoryId={DEMO['factory_id']}",
        {"assigneeId": int(employees["WB8-OPD"]["RecID"]), "remark": "WP-WB8 demo claim"},
        admin_demo["token"],
    )
    if demo_claim[0] >= 400:
        demo_claim = (200, {"reused": True, "previous": unwrap(demo_claim[1]) or demo_claim[1]})
    cur.execute("SELECT notify_channel, COUNT(*) cnt FROM ado_s8_notification_layer GROUP BY notify_channel")
    channels = cur.fetchall()
    write_json("15-s8-auto-alert.json", {
        "channels": channels,
        "unboundErrorBeforeBindRefresh": unbound_before,
        "unclaimedId": unclaimed_id,
        "unclaimedNotPersonalTask": not unclaimed_has,
        "claimBound": {"http": claim_bound[0], "body": unwrap(claim_bound[1]) or claim_bound[1]},
        "afterClaimHasTask": has_item(pick(after_claim, "s8Tasks", "S8Tasks"), str(claim_target)),
        "claimUnboundEmployee": {"http": claim_unbound[0], "body": unwrap(claim_unbound[1]) or claim_unbound[1]},
        "recentS8Notices": s8_notices,
        "transfer": {"http": transfer_s8[0], "body": unwrap(transfer_s8[1]) or transfer_s8[1]},
        "oldAssigneeStillHas": has_item(pick(after_s8_transfer_old, "s8Tasks", "S8Tasks"), str(claim_target)),
        "newAssigneeHas": has_item(pick(after_s8_transfer_new, "s8Tasks", "S8Tasks"), str(claim_target)),
        "demoClaimHttp": demo_claim[0],
        "employees": {k: {"recId": str(v["RecID"]), "sysUserId": str(v["sys_user_id"]) if v["sys_user_id"] else None} for k, v in employees.items()},
        "verdict": verdict(
            all(c["notify_channel"] and "SignalR" in c["notify_channel"] for c in channels)
            and not unclaimed_has
            and claim_bound[0] == 200
            and has_item(pick(after_claim, "s8Tasks", "S8Tasks"), str(claim_target))
            and transfer_s8[0] == 200
            and has_item(pick(after_s8_transfer_new, "s8Tasks", "S8Tasks"), str(claim_target))
            and not has_item(pick(after_s8_transfer_old, "s8Tasks", "S8Tasks"), str(claim_target)),
            "SignalR layer required; unclaimed is not my task; bound claim/transfer notifies via workbench",
        ),
    })

    pending_http, pending_body = request(
        "POST", "/api/flowTask/myPendingPage",
        {"page": 1, "pageSize": 20},
        admin_a["token"],
    )
    count_http, count_body = request("GET", "/api/flowTask/myPendingCount", token=admin_a["token"])
    if count_http >= 400:
        count_http, count_body = request("POST", "/api/flowTask/myPendingCount", {}, admin_a["token"])
    _, wb_admin, _ = summary(admin_a["token"])
    pending_page = as_dict(unwrap(pending_body))
    pending_items = pick(pending_page, "items", "Items") or []
    first_task = as_dict(pending_items[0]) if pending_items else {}
    write_json("16-approval-consistency.json", {
        "pendingPageHttp": pending_http,
        "pendingCountHttp": count_http,
        "pendingCount": unwrap(count_body),
        "workbenchApprovalTotal": pick(as_dict(pick(wb_admin, "approvalTasks", "ApprovalTasks")), "total", "Total"),
        "sampleTaskId": str(pick(first_task, "id", "Id") or ""),
        "sampleJump": next(
            (pick(as_dict(x), "jumpUrl", "JumpUrl") for x in (pick(as_dict(pick(wb_admin, "approvalTasks", "ApprovalTasks")), "items", "Items") or [])),
            None,
        ),
        "verdict": verdict(
            pending_http == 200
            and pick(as_dict(pick(wb_admin, "approvalTasks", "ApprovalTasks")), "total", "Total") == unwrap(count_body),
            "workbench approval count matches flowTask.myPendingCount",
        ),
    })

    created_b = create_plan(
        admin_b["token"], B["factory_id"], B["plan"]["user_id"],
        f"WP-WB8 租户B {BATCH}", datetime.now() + timedelta(days=5),
    )
    created_d = create_plan(
        admin_demo["token"], DEMO["factory_id"], DEMO["operator"]["user_id"],
        f"WP-WB8 租户Demo {BATCH}", datetime.now() + timedelta(days=5),
    )
    detail_cross = request(
        "GET",
        f"/api/AdoSmartOpsImprovementPlan/detail?id={created_online['planId']}",
        token=admin_b["token"],
    )
    _, b_plan_summary, _ = summary(plan_b["token"])
    _, d_op_summary, _ = summary(operator_demo["token"])
    write_json("17-tenant-isolation.json", {
        "planA": created_online["planId"],
        "planB": created_b["planId"],
        "planDemo": created_d["planId"],
        "bSeesAAction": has_item(pick(b_plan_summary, "improvementActions", "ImprovementActions"), created_online["actionId"]),
        "demoSeesAAction": has_item(pick(d_op_summary, "improvementActions", "ImprovementActions"), created_online["actionId"]),
        "bDetailAHttp": detail_cross[0],
        "bHasOwnAction": has_item(pick(b_plan_summary, "improvementActions", "ImprovementActions"), created_b["actionId"]),
        "demoHasOwnAction": has_item(pick(d_op_summary, "improvementActions", "ImprovementActions"), created_d["actionId"]),
        "verdict": verdict(
            created_b["http"] == 200
            and created_d["http"] == 200
            and not has_item(pick(b_plan_summary, "improvementActions", "ImprovementActions"), created_online["actionId"])
            and not has_item(pick(d_op_summary, "improvementActions", "ImprovementActions"), created_online["actionId"])
            and detail_cross[0] in (404, 400)
            and has_item(pick(b_plan_summary, "improvementActions", "ImprovementActions"), created_b["actionId"]),
            "A/B/Demo cannot read each other's action or plan detail",
        ),
    })

    write_json("18-failure-isolation.json", {
        "code": {
            "safeSection": "AidopWorkbenchService.SafeSection catches per partition",
            "noticeNotRolledBack": "SysNoticeService.DispatchPersistedNoticeAsync logs SignalR failure without deleting notice",
        },
        "liveSummaryStillOk": all(x["http"] == 200 for x in summary_timings),
        "planASummaryS8": summaries["planA"]["s8Error"],
        "planAImprovementAvailable": summaries["planA"]["improvementTotal"] is not None,
        "verdict": verdict(
            all(x["http"] == 200 for x in summary_timings),
            "live: summary remains 200 when S8 section is EMPLOYEE_UNBOUND; SignalR-off not executed on shared host",
        ),
    })

    write_json("19-homepage-layout.json", {
        "homepages": {
            admin_a["account"]: admin_a["homepage"],
            plan_a["account"]: plan_a["homepage"],
            admin_b["account"]: admin_b["homepage"],
            plan_b["account"]: plan_b["homepage"],
            admin_demo["account"]: admin_demo["homepage"],
            operator_demo["account"]: operator_demo["homepage"],
        },
        "layoutKey": "AIDOP_WORKBENCH_GRID:{tenantId}:{userId}",
        "verdict": verdict(
            admin_a["homepage"] == "/aidop/smart-ops/grid"
            and plan_a["homepage"] == "/dashboard/home"
            and admin_b["homepage"] == "/aidop/smart-ops/grid"
            and plan_b["homepage"] == "/dashboard/home"
            and admin_demo["homepage"] == "/aidop/smart-ops/grid"
            and operator_demo["homepage"] == "/dashboard/home",
            "director cockpit / others workbench",
        ),
    })

    files = [
        "10-api-summary.json", "11-online-assignment.json", "12-offline-recovery.json",
        "13-transfer.json", "14-due-overdue-idempotency.json", "15-s8-auto-alert.json",
        "16-approval-consistency.json", "17-tenant-isolation.json", "18-failure-isolation.json",
        "19-homepage-layout.json",
    ]
    results = {}
    for name in files:
        payload = json.loads((EVIDENCE / name).read_text(encoding="utf-8"))
        results[name] = payload.get("verdict", {})
    write_json("20-wb8-rollups.json", {
        "date": datetime.now().isoformat(timespec="seconds"),
        "base": BASE,
        "results": results,
        "allPass": all(v.get("pass") for v in results.values()),
    })
    print(json.dumps({"allPass": all(v.get("pass") for v in results.values()), "results": results}, ensure_ascii=False, indent=2))
    conn.close()


if __name__ == "__main__":
    main()