221 lines
9.6 KiB
Python
221 lines
9.6 KiB
Python
# -*- coding: utf-8 -*-
|
||
# P1 真实端到端冒烟（Agent-G）：真实 server + 真实 Pi + 真实 LLM。
|
||
# 单次执行内完成：起 server → 健康/开关检查 → 桌面 mock 授权 → 真实 /api/chat →
|
||
# 校验 run 目录/凭证/审计 → 杀进程树。LLM key 只进请求头/子进程 env，绝不打印落盘。
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import shutil
|
||
import socket
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from http.cookiejar import CookieJar
|
||
from pathlib import Path
|
||
|
||
HERE = Path(__file__).resolve().parent # poc/pi-fallback/smoke
|
||
REPO = HERE.parents[2] # 仓库根
|
||
SITE = HERE / "site" # 临时现场（保留供查验）
|
||
APS_HOME = SITE / "aps-home"
|
||
DATA = APS_HOME / "data"
|
||
PORT = 8391
|
||
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()
|
||
|
||
# 口令：快路规则全部避让（无 分析+数据/看看+订单/交期/看板/排产祈使/星期几 等），
|
||
# 落到 assistant.reply/unknown 分支；只读分析类，无任何写意图。
|
||
QUERY = ("琢磨个冷门的角度：把每张待交付订单的交货日期换算成一周里的第几天，"
|
||
"交货压力集中在一周的开头还是末尾？再数数物料种类，凭数据给个备料松紧的"
|
||
"判断，别改任何东西，只给结论。")
|
||
|
||
R: dict = {"query": QUERY, "startedAt": time.strftime("%Y-%m-%d %H:%M:%S")}
|
||
|
||
|
||
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:
|
||
# ---- 1) 搭临时现场（隔离：不动 server/data 真实数据）----
|
||
if SITE.exists():
|
||
shutil.rmtree(SITE)
|
||
DATA.mkdir(parents=True)
|
||
shutil.copy(REPO / "server" / "data" / "world.json", DATA / "world.json")
|
||
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),
|
||
# P0 已实测：.env 的 kimi-k2-0711-preview 该 key 无权限(404)，显式锁定 kimi-k2.6
|
||
"APS_FALLBACK_MODEL": "aps-fallback/kimi-k2.6",
|
||
"PYTHONIOENCODING": "utf-8",
|
||
})
|
||
env.pop("APS_FALLBACK_DIR", None) # run 目录走默认 path_under_data → 仍在临时 APS_HOME 内
|
||
R["envOverrides"] = {k: ("<set>" if "KEY" not in k else "<hidden>") for k in (
|
||
"APS_HOME", "APS_FEATURES_PATH", "APS_WORLD_PATH", "APS_DB_PATH",
|
||
"APS_LICENSE_PROVIDER", "APS_PORT", "APS_FALLBACK_MODEL")}
|
||
|
||
# ---- 2) 起真实 server（uvicorn 子进程，finally 杀树）----
|
||
server_log = open(HERE / "server.log", "w", encoding="utf-8")
|
||
proc = subprocess.Popen(
|
||
[sys.executable, "-m", "uvicorn", "server.main:app",
|
||
"--host", "127.0.0.1", "--port", str(PORT)],
|
||
cwd=str(REPO), env=env, stdout=server_log, stderr=subprocess.STDOUT)
|
||
R["serverPid"] = proc.pid
|
||
try:
|
||
# 判据 1：健康
|
||
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 内未就绪（见 server.log）"
|
||
return 1
|
||
|
||
cj = CookieJar()
|
||
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
|
||
|
||
# 开发态认证：桌面模式 + mock 授权码激活（cookie aps_desktop_session）
|
||
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")}
|
||
|
||
# 判据 2：/api/features
|
||
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"),
|
||
}
|
||
|
||
# 判据 3：真实 /api/chat（SSE），读到底
|
||
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 / "chat1.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
|
||
intent_ev = next((e for e in events if e.get("type") == "intent"), None)
|
||
R["intentEvent"] = intent_ev
|
||
reply_text = "".join(e.get("text", "") for e in events if e.get("type") == "token")
|
||
R["replyLen"] = len(reply_text)
|
||
R["replyHead"] = reply_text[:600]
|
||
R["replyTail"] = reply_text[-300:]
|
||
R["replyContainsFallbackMarker"] = "[智能兜底 · 草稿]" in reply_text
|
||
(HERE / "reply1.txt").write_text(reply_text, encoding="utf-8")
|
||
finally:
|
||
# 杀 server 进程树（pi/node 子进程随树回收）
|
||
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()
|
||
|
||
# ---- 3) 收尾后离线校验（server 已停，纯读文件）----
|
||
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")}
|
||
R["reportTextHead"] = (outcome.get("report_text") or "")[:600]
|
||
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}),
|
||
"sampleCallId": recs[0]["call_id"] if recs else None,
|
||
}
|
||
rep = run / "outbox" / "report.md"
|
||
R["reportMdExists"] = rep.is_file()
|
||
log = run / "orchestrator.log"
|
||
if log.is_file():
|
||
R["orchestratorLogTail"] = log.read_text(encoding="utf-8").splitlines()[-6:]
|
||
|
||
# 判据 5：审计链 agent.fallback.propose
|
||
world = json.loads((DATA / "world.json").read_text(encoding="utf-8"))
|
||
audits = [a for a in (world.get("auditEvents") or [])
|
||
if a.get("action") == "agent.fallback.propose"]
|
||
R["fallbackAuditEvents"] = [
|
||
{"actor": a.get("actor"), "result": a.get("result"),
|
||
"power": a.get("power"), "rationale": a.get("rationale")}
|
||
for a in audits]
|
||
|
||
# 真实数据零污染校验：server/data/world.json 改动前后字节级一致由外层 git/校验负责，
|
||
# 这里确认 run 目录与审计均落在临时 APS_HOME 内。
|
||
R["dataIsolated"] = str(fb_root).startswith(str(SITE))
|
||
|
||
(HERE / "results.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", "fallbackAuditEvents",
|
||
"dataIsolated") if k in R}, ensure_ascii=False, indent=2, default=str))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|