# P3 S7 真实冒烟（Agent-N）：运维只读诊断（P3-DESIGN §9.2 验收点 A–D）。 # A ops 身份对话 → inbox/ops/ 四文件 + 行级脱敏生效（预埋敏感串断言） # B P1 草稿语义 + 世界零变更 # C 非 ops 对照（desktop planner 身份）→ 物理不可见：走原话术、run 目录零新增 # D config.apply 双人审批链 —— 已由黄金 H-20 确定性全覆盖（双身份 + # PARTIALLY_APPROVED/SOD/grant/原子替换/备份/双人审计），冒烟层不重复， # 此处如实登记覆盖关系。 # # 身份手法（P3-DESIGN §4.1：角色 = 纯配置零代码）： # - ops 通道：APS_AUTH_ENABLED=0 → bypass_identity 角色含 admin ∈ S7.roles； # - 非 ops 对照：desktop 许可身份（角色 desktop/planner，∉ S7.roles）。 # 分阶段执行（单命令 300s 预算）：setup / ops / control / finalize。 # LLM key 只经 server 子进程环境从仓库根 .env 流入，本脚本不读不打印不落盘。 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 / "site5" APS_HOME = SITE / "aps-home" DATA = APS_HOME / "data" PORT = 8395 BASE = f"http://127.0.0.1:{PORT}" DEVICE_ID = "smoke-n-device-s7-0002" RESULTS = HERE / "s7_results.json" MOCK_TENANT = hashlib.sha256(f"{SITE.resolve()}|tenant".encode()).hexdigest()[:32] MOCK_SECRET = hashlib.sha256(f"{SITE.resolve()}|license".encode()).hexdigest() QUERY_OPS = "帮我看下服务日志和配置有没有异常" # 预埋敏感行（脱敏断言用）：键值形态 + 长 key 形态各一 SECRET_VALUE = "abc123secretTOKEN99" KEYISH_VALUE = "A1b2C3d4E5f6G7h8I9j0K1l2" PLANTED_LOG_LINES = [ "INFO startup ok", f"ERROR auth failed token={SECRET_VALUE}", f"WARN upstream api_key={KEYISH_VALUE} retrying", "INFO done", ] def load_results() -> dict: if RESULTS.is_file(): return json.loads(RESULTS.read_text(encoding="utf-8")) return {"startedAt": time.strftime("%Y-%m-%d %H:%M:%S"), "checks": {}} R = load_results() def check(name: str, ok: bool, detail: str = "") -> bool: R["checks"][name] = {"ok": bool(ok), "detail": detail} print(f"[{'PASS' if ok else 'FAIL'}] {name}" + (f" — {detail}" if detail else "")) return ok def dump_results() -> None: RESULTS.write_text(json.dumps(R, ensure_ascii=False, indent=2, default=str), encoding="utf-8") 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 bypass_world_path() -> Path: """APS_AUTH_ENABLED=0 的 bypass 身份（user_id=1, tenant=platform）世界路径。""" return DATA / "tenants" / "platform" / "projects" / "personal-1" / "world.json" def world_fingerprint_of(path: Path) -> str: sys.path.insert(0, str(REPO)) from server.agent_core.harness import world_fingerprint return world_fingerprint(json.loads(path.read_text(encoding="utf-8"))) def start_server(env_extra: dict, log_name: str): env = dict(os.environ) env.update({ "APS_HOME": str(APS_HOME), "APS_FEATURES_PATH": str(SITE / "features.json"), "APS_FALLBACK_HIGHRISK_PATH": str(SITE / "fallback-highrisk.json"), "APS_WORLD_PATH": str(DATA / "world.json"), "APS_DB_PATH": str(DATA / "master.db"), "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_TIMEOUT_SEC": "170", "NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost", "PYTHONIOENCODING": "utf-8", }) env.pop("APS_FALLBACK_DIR", None) env.update(env_extra) server_log = open(HERE / log_name, "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) return proc, server_log def wait_health(proc, seconds=90) -> dict | None: for _ in range(seconds): if proc.poll() is not None: return None try: with urllib.request.urlopen(BASE + "/api/health", timeout=3) as r: return json.loads(r.read().decode("utf-8")) except Exception: time.sleep(1) return None def kill_server(proc, server_log) -> None: if proc.poll() is None: subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"], capture_output=True, timeout=30, check=False) try: proc.wait(timeout=20) except Exception: pass server_log.close() def make_opener(): cj = CookieJar() return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj)) def parse_sse(raw: str) -> list[dict]: 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 return events def _chat(opener, query: str, timeout: int = 190) -> tuple[str, list[dict]]: resp = http(opener, "POST", "/api/chat", {"text": query}, headers=desktop_headers(), timeout=timeout) raw = resp.read().decode("utf-8", errors="replace") return raw, parse_sse(raw) def _robocopy_purge(target: Path) -> None: empty = target.parent / "_empty_purge" empty.mkdir(exist_ok=True) subprocess.run(["robocopy", str(empty), str(target), "/MIR", "/NFL", "/NDL", "/NJH", "/NJS", "/NP"], capture_output=True, timeout=120, check=False) try: empty.rmdir() target.rmdir() except OSError: pass # --------------------------------------------------------------------------- # phase: setup # --------------------------------------------------------------------------- def phase_setup() -> int: if SITE.exists(): for attempt in range(30): try: shutil.rmtree(SITE) break except OSError: if attempt == 29: _robocopy_purge(SITE) if SITE.exists(): raise time.sleep(1.0) DATA.mkdir(parents=True) shutil.copy(REPO / "server" / "data" / "world.json", DATA / "world.json") bworld = bypass_world_path() bworld.parent.mkdir(parents=True) shutil.copy(REPO / "server" / "data" / "world.json", bworld) (SITE / "features.json").write_text( json.dumps({"version": 1, "features": {"fallback": True}}, ensure_ascii=False), encoding="utf-8") whitelist = { "whitelistVersion": 1, "updatedAt": "2026-09-05T09:00:00", "updatedBy": "smoke-s7", "scenarios": { "S4": {"enabled": True, "intents": ["flex.simulate_due", "flex.compare", "scenario.compare", "scenario.sensitivity"], "roles": ["planner", "admin"]}, "S6": {"enabled": True, "intents": ["mes.report"], "roles": ["planner", "admin"], "maxItemsPerRun": 20}, "S7": {"enabled": True, "intents": ["agent.fallback.ops.config.apply", "agent.fallback.policy.update"], "roles": ["ops", "admin"]}}} (SITE / "fallback-highrisk.json").write_text( json.dumps(whitelist, ensure_ascii=False, indent=2), encoding="utf-8") # 预埋含敏感串的日志（APS_HOME/logs，诊断注入的读取源） log_dir = APS_HOME / "logs" log_dir.mkdir(parents=True) (log_dir / "server.log").write_text( "\n".join(PLANTED_LOG_LINES) + "\n", encoding="utf-8") R.clear() R.update({"startedAt": time.strftime("%Y-%m-%d %H:%M:%S"), "checks": {}, "site": str(SITE), "bypassWorld": str(bworld), "query": QUERY_OPS}) check("setup.site_ready", bworld.is_file() and (log_dir / "server.log").is_file()) dump_results() return 0 # --------------------------------------------------------------------------- # phase: ops —— 运维身份只读诊断（验收点 A/B） # --------------------------------------------------------------------------- def phase_ops() -> int: bworld = bypass_world_path() fp_before = world_fingerprint_of(bworld) if bworld.is_file() else None proc, server_log = start_server({"APS_AUTH_ENABLED": "0"}, "server_s7_ops.log") try: health = wait_health(proc) if not health: check("A0.ops_server_health", False, "server 90s 内未就绪（见 server_s7_ops.log）") dump_results() return 1 check("A0.ops_server_health", True) opener = make_opener() fb_root = DATA / "fallback" runs_before = {p.name for p in fb_root.glob("fb-*")} if fb_root.is_dir() else set() t0 = time.monotonic() raw, events = _chat(opener, QUERY_OPS) R["chatElapsedSec"] = round(time.monotonic() - t0, 1) (HERE / "chat_s7_ops.sse").write_text(raw, encoding="utf-8") reply_text = "".join(e.get("text", "") for e in events if e.get("type") == "token") (HERE / "reply_s7_ops.txt").write_text(reply_text, encoding="utf-8") R["replyHead"] = reply_text[:600] intent_ev = next((e for e in events if e.get("type") == "intent"), None) intent_name = ((intent_ev or {}).get("intent") or {}).get("intent") \ if isinstance((intent_ev or {}).get("intent"), dict) \ else (intent_ev or {}).get("intent") check("A1.intent_lands_assistant_reply", intent_name == "assistant.reply", f"intent={intent_name}") runs_after = {p.name for p in fb_root.glob("fb-*")} if fb_root.is_dir() else set() new_runs = sorted(runs_after - runs_before) run_dir = fb_root / new_runs[-1] if new_runs else None R["opsRunDir"] = str(run_dir) if run_dir else None # 验收点 A：inbox/ops/ 四文件 + 行级脱敏 ops_dir = run_dir / "inbox" / "ops" if run_dir else None four = ("logs-tail.md", "config-snapshot.md", "health.md", "integrations.md") files_ok = bool(ops_dir) and all((ops_dir / n).is_file() for n in four) logs_tail = (ops_dir / "logs-tail.md").read_text(encoding="utf-8") \ if ops_dir and (ops_dir / "logs-tail.md").is_file() else "" snapshot = (ops_dir / "config-snapshot.md").read_text(encoding="utf-8") \ if ops_dir and (ops_dir / "config-snapshot.md").is_file() else "" R["logsTailExcerpt"] = logs_tail[:500] check("A2.ops_four_files", files_ok, f"ops_dir={ops_dir} files={[n for n in four if ops_dir and (ops_dir / n).is_file()]}") check("A3.redaction_effective", "***REDACTED***" in logs_tail and SECRET_VALUE not in logs_tail and KEYISH_VALUE not in logs_tail, f"REDACTED 出现={'***REDACTED***' in logs_tail}；" f"敏感串泄漏={SECRET_VALUE in logs_tail or KEYISH_VALUE in logs_tail}") check("A4.config_snapshot_projection", "features.json" in snapshot and "fallback-highrisk.json" in snapshot, "features 全文 + 白名单裁决投影均在") # 验收点 B：P1 草稿语义 + 世界零变更 check("B1.draft_semantics", "草稿" in reply_text, reply_text.replace("\n", " ")[:150]) fp_after = world_fingerprint_of(bworld) if bworld.is_file() else None check("B2.world_unchanged", fp_before is not None and fp_before == fp_after, f"fp {str(fp_before)[:12]} == {str(fp_after)[:12]}") # 诊断报告引用校验（calls.jsonl 凭证链存在即 P1 围墙内读取发生） calls = run_dir / "calls.jsonl" if run_dir else None n_calls = 0 if calls and calls.is_file(): n_calls = len([x for x in calls.read_text(encoding="utf-8").splitlines() if x.strip()]) R["callsCount"] = n_calls check("B3.pi_read_in_wall", n_calls > 0, f"calls.jsonl {n_calls} 行（fs_read 围墙内读取）") finally: kill_server(proc, server_log) dump_results() failed = [k for k, v in R["checks"].items() if not v["ok"]] return 0 if not failed else 1 # --------------------------------------------------------------------------- # phase: control —— 非 ops 对照（desktop planner 身份 → 物理不可见，验收点 C） # --------------------------------------------------------------------------- def phase_control() -> int: proc, server_log = start_server({"APS_LICENSE_PROVIDER": "mock"}, "server_s7_control.log") try: health = wait_health(proc) if not health: check("C0.control_server_health", False, "server 90s 内未就绪（见 server_s7_control.log）") dump_results() return 1 opener = make_opener() resp = http(opener, "POST", "/api/auth/license/activate", {"code": "APS-DAY-DEMO", "deviceId": DEVICE_ID}, headers=desktop_headers()) resp.read() fb_root = DATA / "fallback" runs_before = {p.name for p in fb_root.glob("fb-*")} if fb_root.is_dir() else set() raw, events = _chat(opener, QUERY_OPS, timeout=120) (HERE / "chat_s7_control.sse").write_text(raw, encoding="utf-8") reply_text = "".join(e.get("text", "") for e in events if e.get("type") == "token") (HERE / "reply_s7_control.txt").write_text(reply_text, encoding="utf-8") R["controlReplyHead"] = reply_text[:400] runs_after = {p.name for p in fb_root.glob("fb-*")} if fb_root.is_dir() else set() new_runs = sorted(runs_after - runs_before) # 物理不可见：零新增 run 目录 + 回复不含 S7 兜底标记（走原话术路径） check("C.non_ops_invisible", not new_runs and "[智能兜底" not in reply_text, f"新增 run={new_runs}；兜底标记={'[智能兜底' in reply_text}") finally: kill_server(proc, server_log) dump_results() failed = [k for k, v in R["checks"].items() if not v["ok"]] return 0 if not failed else 1 # --------------------------------------------------------------------------- # phase: finalize —— 收尾校验 + 汇总 # --------------------------------------------------------------------------- def phase_finalize() -> int: time.sleep(2) port_free = False try: urllib.request.urlopen(BASE + "/api/health", timeout=3) except Exception: port_free = True wmi = "" ps_exe = shutil.which("powershell.exe") or shutil.which("powershell") or "powershell.exe" try: wmi = subprocess.run( [ps_exe, "-NoProfile", "-Command", ("Get-CimInstance Win32_Process | Where-Object " "{$_.CommandLine -like '*pi-coding-agent*cli.js*' " "-and $_.CommandLine -notlike '*Get-CimInstance*'} " "| Select-Object -ExpandProperty ProcessId")], capture_output=True, text=True, timeout=60, check=False).stdout.strip() except Exception as exc: R["piResidualCheckError"] = str(exc) R["piResidualPids"] = [x for x in wmi.splitlines() if x.strip()] git = subprocess.run(["git", "status", "--short", "server/data"], cwd=str(REPO), capture_output=True, text=True, timeout=30, check=False).stdout.strip() R["gitServerData"] = git or "(clean)" check("H.cleanup", port_free and not R["piResidualPids"] and not git, f"端口释放={port_free} pi 残留={R['piResidualPids']} git={git or 'clean'}") # 验收点 D 覆盖关系登记（黄金 H-20 确定性全覆盖，冒烟层不重复） R["acceptD"] = {"coveredBy": "tests/golden/test_fallback_highrisk.py::" "test_s7_config_apply_p3_dual_approval_flow", "note": "双人审批链（PARTIALLY_APPROVED→SOD_DENIED→异人二批→" "executionGrant→原子替换+备份+双人审计）确定性锁定"} check("D.covered_by_golden_H20", True, "见 acceptD 登记") dump_results() failed = [k for k, v in R["checks"].items() if not v["ok"]] total = len(R["checks"]) print(json.dumps({"passed": total - len(failed), "total": total, "failed": failed}, ensure_ascii=False, indent=2)) return 0 if not failed else 1 PHASES = {"setup": phase_setup, "ops": phase_ops, "control": phase_control, "finalize": phase_finalize} if __name__ == "__main__": if len(sys.argv) != 2 or sys.argv[1] not in PHASES: print(f"usage: {sys.argv[0]} <{'|'.join(PHASES)}>") sys.exit(2) sys.exit(PHASES[sys.argv[1]]())