aps-agent/poc/pi-fallback/smoke/smoke_e2e_diag.py

# -*- coding: utf-8 -*-
# P1 冒烟 · 诊断轮（Agent-G）：与 smoke_e2e.py 完全同链，仅两点差异并如实标注：
#   1) 预播种租户世界（桌面 mock 身份下聊天读写的是租户世界，首轮为空导致快照空）；
#      种子 = server/data/world.json 演示数据的副本，全部落在临时 APS_HOME 内。
#   2) server 进程内对 _GUARD_TS_TEMPLATE 做内存级单行修正（注释行的 { block:... } 转义），
#      用于回答「该 bug 是否为唯一拦路点」。产品代码文件零改动；本轮结论不作为产品路径真相，
#      仅作修复验证的预判证据。shim 见 shim_server.py。
import hashlib
import json
import os
import shutil
import subprocess
import sys
import time
import urllib.request
from http.cookiejar import CookieJar
from pathlib import Path

HERE = Path(__file__).resolve().parent
REPO = HERE.parents[2]
SITE = HERE / "site2"
APS_HOME = SITE / "aps-home"
DATA = APS_HOME / "data"
PORT = 8392
BASE = f"http://127.0.0.1:{PORT}"
DEVICE_ID = "smoke-g-device-0001"
MOCK_TENANT = hashlib.sha256(f"{SITE.resolve()}|tenant".encode()).hexdigest()[:32]
MOCK_SECRET = hashlib.sha256(f"{SITE.resolve()}|license".encode()).hexdigest()
NODE_BIN = (
    (os.environ.get("APS_FALLBACK_NODE") or "").strip()
    or shutil.which("node.exe")
    or shutil.which("node")
    or "node"
)

QUERY = ("琢磨个冷门的角度：把每张待交付订单的交货日期换算成一周里的第几天，"
         "交货压力集中在一周的开头还是末尾？再数数物料种类，凭数据给个备料松紧的"
         "判断，别改任何东西，只给结论。")

R: dict = {"query": QUERY, "mode": "diagnostic-shim", "startedAt": time.strftime("%Y-%m-%d %H:%M:%S")}


def _device_user_id(device_id: str) -> int:
    return int(hashlib.sha256(device_id.encode()).hexdigest()[:15], 16)


def http(opener, method, path, body=None, headers=None, timeout=30):
    req = urllib.request.Request(BASE + path, method=method)
    for k, v in (headers or {}).items():
        req.add_header(k, v)
    data = None
    if body is not None:
        data = json.dumps(body, ensure_ascii=False).encode("utf-8")
        req.add_header("Content-Type", "application/json")
    return opener.open(req, data=data, timeout=timeout)


def desktop_headers():
    return {"x-aps-client": "desktop", "x-aps-device-id": DEVICE_ID}


def main() -> int:
    if SITE.exists():
        shutil.rmtree(SITE)
    DATA.mkdir(parents=True)
    # 预播种：平台默认世界 + 桌面租户个人项目世界（均为演示数据副本，限临时目录）
    demo_world = (REPO / "server" / "data" / "world.json").read_bytes()
    (DATA / "world.json").write_bytes(demo_world)
    tenant_world = (DATA / "tenants" / MOCK_TENANT / "projects"
                    / f"personal-{_device_user_id(DEVICE_ID)}" / "world.json")
    tenant_world.parent.mkdir(parents=True, exist_ok=True)
    tenant_world.write_bytes(demo_world)
    features = SITE / "features.json"
    features.write_text(
        json.dumps({"version": 1, "features": {"fallback": True}}, ensure_ascii=False),
        encoding="utf-8")
    R["site"] = str(SITE)

    env = dict(os.environ)
    env.update({
        "APS_HOME": str(APS_HOME),
        "APS_FEATURES_PATH": str(features),
        "APS_WORLD_PATH": str(DATA / "world.json"),
        "APS_DB_PATH": str(DATA / "master.db"),
        "APS_LICENSE_PROVIDER": "mock",
        "APS_MOCK_LICENSE_SECRET": MOCK_SECRET,
        "APS_MOCK_LICENSE_TENANT_UUID": MOCK_TENANT,
        "APS_PORT": str(PORT),
        "APS_FALLBACK_MODEL": "aps-fallback/kimi-k2.6",
        "APS_FALLBACK_NODE": NODE_BIN,
        "PYTHONIOENCODING": "utf-8",
    })
    env.pop("APS_FALLBACK_DIR", None)

    server_log = open(HERE / "server2.log", "w", encoding="utf-8")
    proc = subprocess.Popen(
        [sys.executable, str(HERE / "shim_server.py")],
        cwd=str(REPO), env=env, stdout=server_log, stderr=subprocess.STDOUT)
    R["serverPid"] = proc.pid
    try:
        health = None
        for _ in range(90):
            if proc.poll() is not None:
                break
            try:
                with urllib.request.urlopen(BASE + "/api/health", timeout=3) as r:
                    health = json.loads(r.read().decode("utf-8"))
                break
            except Exception:
                time.sleep(1)
        R["health"] = health
        if not health:
            R["fatal"] = "server 90s 内未就绪（见 server2.log）"
            return 1

        cj = CookieJar()
        opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
        resp = http(opener, "POST", "/api/auth/license/activate",
                    {"code": "APS-DAY-DEMO", "deviceId": DEVICE_ID},
                    headers=desktop_headers())
        act = json.loads(resp.read().decode("utf-8"))
        R["licenseActivate"] = {"status": resp.status, "provider": act.get("provider"),
                                "username": (act.get("user") or {}).get("username")}

        resp = http(opener, "GET", "/api/features", headers=desktop_headers())
        feats = json.loads(resp.read().decode("utf-8"))
        R["features"] = {"fallback": feats["features"].get("fallback"),
                         "defaultOff": feats.get("defaultOff"), "source": feats.get("source")}

        t0 = time.monotonic()
        resp = http(opener, "POST", "/api/chat", {"text": QUERY},
                    headers=desktop_headers(), timeout=280)
        raw = resp.read().decode("utf-8", errors="replace")
        R["chatElapsedSec"] = round(time.monotonic() - t0, 1)
        (HERE / "chat2.sse").write_text(raw, encoding="utf-8")

        events = []
        for line in raw.splitlines():
            line = line.strip()
            if line.startswith("data:"):
                try:
                    events.append(json.loads(line[5:].strip()))
                except json.JSONDecodeError:
                    pass
        R["sseEventTypes"] = {}
        for e in events:
            R["sseEventTypes"][e.get("type", "?")] = R["sseEventTypes"].get(e.get("type", "?"), 0) + 1
        R["intentEvent"] = next((e for e in events if e.get("type") == "intent"), None)
        reply_text = "".join(e.get("text", "") for e in events if e.get("type") == "token")
        R["replyLen"] = len(reply_text)
        R["replyContainsFallbackMarker"] = "[智能兜底 · 草稿]" in reply_text
        (HERE / "reply2.txt").write_text(reply_text, encoding="utf-8")
    finally:
        if proc.poll() is None:
            subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"],
                           capture_output=True, timeout=30)
        try:
            proc.wait(timeout=20)
        except Exception:
            pass
        server_log.close()

    # 离线校验
    fb_root = DATA / "fallback"
    runs = sorted([p for p in fb_root.glob("fb-*") if p.is_dir()]) if fb_root.is_dir() else []
    R["runDirs"] = [str(p) for p in runs]
    if runs:
        run = runs[-1]
        R["runTree"] = sorted(str(p.relative_to(run)) for p in run.rglob("*"))
        rj = run / "result.json"
        if rj.is_file():
            outcome = json.loads(rj.read_text(encoding="utf-8"))
            R["resultJson"] = {k: outcome.get(k) for k in (
                "run_id", "ok", "stop_reason", "error_message", "steps",
                "output_bytes", "elapsed_sec", "citation_check")}
        calls = run / "calls.jsonl"
        if calls.is_file():
            recs = [json.loads(x) for x in calls.read_text(encoding="utf-8").splitlines() if x.strip()]
            R["callsJsonl"] = {"count": len(recs),
                               "tools": sorted({r.get("tool", "?") for r in recs}),
                               "statuses": sorted({r.get("status", "?") for r in recs})}
        R["reportMdExists"] = (run / "outbox" / "report.md").is_file()
        guard = next(run.glob("guard-*.ts"), None)
        if guard:
            g = guard.read_text(encoding="utf-8")
            R["guardSanity"] = {"hasReturnBlock": "return { block: true, reason };" in g}
        log = run / "orchestrator.log"
        if log.is_file():
            R["orchestratorLogTail"] = log.read_text(encoding="utf-8").splitlines()[-8:]
        snap = run / "inbox" / "snapshot.md"
        if snap.is_file():
            R["snapshotMd"] = snap.read_text(encoding="utf-8")
        ocsv = run / "inbox" / "orders.csv"
        if ocsv.is_file():
            R["ordersCsv"] = ocsv.read_text(encoding="utf-8")

    # 审计：聊天身份的世界（租户个人项目）
    audits = []
    for w in DATA.glob("tenants/*/projects/*/world.json"):
        for a in (json.loads(w.read_text(encoding="utf-8")).get("auditEvents") or []):
            if a.get("action") == "agent.fallback.propose":
                audits.append({"worldFile": str(w), "actor": a.get("actor"),
                               "result": a.get("result"), "power": a.get("power"),
                               "rationale": a.get("rationale")})
    R["fallbackAuditEvents"] = audits

    (HERE / "results2.json").write_text(
        json.dumps(R, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
    print(json.dumps({k: R[k] for k in (
        "health", "licenseActivate", "features", "chatElapsedSec", "sseEventTypes",
        "intentEvent", "replyLen", "replyContainsFallbackMarker", "runDirs", "resultJson",
        "callsJsonl", "reportMdExists", "guardSanity", "fallbackAuditEvents") if k in R},
        ensure_ascii=False, indent=2, default=str))
    return 0


if __name__ == "__main__":
    sys.exit(main())