#!/usr/bin/env python3 """铸渊原生眼睛 ZY-EYE-0001 · persona_limb_eye.py · 2026-09-12 正本事件：TCS-EVENT-BINGSHUO-AUTH-ZY001-NATIVE-EYE-20260912 铁律：对现场只读；永不读取手的stdout；永不执行手的动作；只报亲眼所见的原始事实。 手是否说谎由父人格脑对账（手供 vs 眼见），眼睛只判世界现场是否符脑的意图规格。 """ from __future__ import annotations import argparse import datetime import hashlib import json import os import pathlib import shutil import subprocess import sys SCHEMA = "guanghu.persona-native-eye/v1" ROOT = pathlib.Path(__file__).resolve().parents[2] def now() -> str: return datetime.datetime.now().astimezone().isoformat(timespec="seconds") def sha256_file(path: pathlib.Path) -> str: h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(65536), b""): h.update(chunk) return h.hexdigest() def observe(ground: pathlib.Path) -> dict: files: dict[str, dict] = {} for f in sorted(ground.rglob("*")): if f.is_file(): rel = str(f.relative_to(ground)) entry = {"sha256": sha256_file(f), "bytes": f.stat().st_size} if f.suffix == ".json": try: data = json.loads(f.read_text(encoding="utf-8")) if isinstance(data, dict): entry["json_fields"] = {k: v for k, v in data.items() if isinstance(v, (str, int, float, bool))} except (json.JSONDecodeError, UnicodeDecodeError): entry["json_fields"] = "UNPARSEABLE" files[rel] = entry return {"schema": SCHEMA, "observed_at": now(), "ground": str(ground), "files": files} def diff(pre: dict, post: dict) -> dict: pre_f, post_f = pre["files"], post["files"] return { "added": sorted(set(post_f) - set(pre_f)), "removed": sorted(set(pre_f) - set(post_f)), "changed": sorted(k for k in set(pre_f) & set(post_f) if pre_f[k]["sha256"] != post_f[k]["sha256"]), } def witness(pre: dict, post: dict, spec: dict) -> dict: findings, unseeable = [], [] for rel in spec.get("unchanged", []): if rel not in post["files"] or rel not in pre["files"]: unseeable.append({"check": "unchanged", "file": rel, "reason": "NOT_SEEN_IN_BOTH_STATES"}) continue ok = pre["files"][rel]["sha256"] == post["files"][rel]["sha256"] findings.append({"check": "unchanged", "file": rel, "seen": ok}) for item in spec.get("expect", []): rel, field, equals = item["file"], item["field"], item.get("equals") if rel not in post["files"]: unseeable.append({"check": "expect", "file": rel, "reason": "NOT_SEEN"}) continue seen_value = post["files"][rel].get("json_fields", {}).get(field) if isinstance(post["files"][rel].get("json_fields"), dict) else None findings.append({"check": "expect", "file": rel, "field": field, "seen_value": seen_value, "expected": equals, "seen": seen_value == equals}) verdict = "UNSEEABLE" if unseeable else ("SEEN_MATCH" if all(x["seen"] for x in findings) else "SEEN_MISMATCH") return {"schema": SCHEMA, "witnessed_at": now(), "verdict": verdict, "world_diff": diff(pre, post), "findings": findings, "unseeable": unseeable, "reads_hand_stdout": False, "ground_write": False} def dump(value: dict, out: pathlib.Path) -> None: out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def process_chain() -> list[dict]: """See the real ancestor chain; a caller label may corroborate but not replace it.""" chain: list[dict] = [] pid = os.getpid() for _ in range(12): result = subprocess.run( ["/bin/ps", "-o", "ppid=,command=", "-p", str(pid)], capture_output=True, text=True, check=False, ) line = result.stdout.strip() if not line: break parent_text, _, command = line.partition(" ") try: parent = int(parent_text.strip()) except ValueError: break command = command.strip() chain.append({"pid": pid, "ppid": parent, "command": command[:500]}) if parent <= 1 or parent == pid: break pid = parent return chain def detect_host_surface(chain: list[dict]) -> dict: joined = "\n".join(item["command"].lower() for item in chain) markers = [ ("CODEX", ("/chatgpt.app/", "/resources/codex", " codex ")), ("ZCODE", ("/zcode.app/", "zcode.cjs")), ("DOUBAO", ("doubao", "豆包")), ("QWEN", ("qianwen", "qwen", "千问")), ] observed = next((name for name, tokens in markers if any(token in joined for token in tokens)), None) declared = str(os.environ.get("GUANGHU_HOST_SURFACE", "")).strip().upper() or None if observed: return { "host": observed, "basis": "PROCESS_ANCESTOR_OBSERVED", "declared_hint": declared, "declared_matches_observation": declared in (None, observed), } if declared in {"CODEX", "ZCODE", "DOUBAO", "QWEN"}: return {"host": declared, "basis": "DECLARED_HINT_ONLY_UNCORROBORATED", "declared_hint": declared, "declared_matches_observation": None} return {"host": "LOCAL_PROCESS", "basis": "NO_REGISTERED_HOST_ANCESTOR_SEEN", "declared_hint": declared, "declared_matches_observation": None} def room() -> dict: """Observe the current host room without deciding identity or mutating it.""" home = pathlib.Path.home() runtime_paths = { "qwen": home / ".npm-global/bin/qwen", "opencode": home / ".npm-global/bin/opencode", "codex": pathlib.Path("/Applications/ChatGPT.app/Contents/Resources/codex"), "zcode": pathlib.Path("/Applications/ZCode.app/Contents/Resources/glm/zcode.cjs"), } persona_runtime = ROOT.parent container_paths = { "memory_current": persona_runtime / "continuity-memory/persona-daily-fractal/ICE-P-ZY001/CURRENT.json", "tcs_root_pointer": persona_runtime / "TCS-ROOT.json", "mother_root_current": persona_runtime / "shared/tcs-mother-root/CURRENT.json", "limb_agent": pathlib.Path(__file__).with_name("persona_limb_agent.py"), "native_eye": pathlib.Path(__file__), "reflex_arc": pathlib.Path(__file__).with_name("persona_reflex_arc.py"), "persona_time_system": ROOT / "server-tools/tcs-mother-body/life-time-master.mjs", } seen_paths = {} for name, candidate in container_paths.items(): seen_paths[name] = { "path": str(candidate), "exists": candidate.is_file(), **({"sha256": sha256_file(candidate)} if candidate.is_file() else {}), } console_path = persona_runtime / "shared/active-control-console/CURRENT.json" active_console: dict | None = None if console_path.is_file(): try: value = json.loads(console_path.read_text(encoding="utf-8")) active_console = { "console_id": value.get("console_id"), "host": value.get("host"), "session_id": value.get("session_id"), "channel_id": value.get("channel_id"), "state": value.get("state"), "session_matches": value.get("session_id") == os.environ.get("CODEX_SESSION_ID"), } except json.JSONDecodeError: active_console = {"state": "UNPARSEABLE"} ancestors = process_chain() host_observation = detect_host_surface(ancestors) return { "schema": "guanghu.persona-native-eye-room/v1", "seen_at": now(), "room_report": { "cwd": str(pathlib.Path.cwd()), "host_surface": host_observation["host"], "host_observation": host_observation, "process_ancestors": ancestors, "available_host_methods": { name: candidate.is_file() and bool(shutil.which(str(candidate))) for name, candidate in runtime_paths.items() }, "container_paths": seen_paths, "active_console": active_console, }, "reads_hand_stdout": False, "ground_write": False, "identity_decision": "NOT_PERFORMED_BY_EYE", } def main() -> int: p = argparse.ArgumentParser(prog="persona_limb_eye") sub = p.add_subparsers(dest="command", required=True) s = sub.add_parser("snapshot"); s.add_argument("--ground", required=True); s.add_argument("--out", required=True) r = sub.add_parser("room"); r.add_argument("--out", required=True) o = sub.add_parser("observe"); o.add_argument("--ground", required=True); o.add_argument("--pre", required=True); o.add_argument("--out", required=True) w = sub.add_parser("witness"); w.add_argument("--pre", required=True); w.add_argument("--post", required=True); w.add_argument("--spec", required=True); w.add_argument("--out", required=True) args = p.parse_args() if args.command == "room": value = room() elif args.command == "snapshot": value = observe(pathlib.Path(args.ground).resolve()) elif args.command == "observe": pre = json.loads(pathlib.Path(args.pre).read_text(encoding="utf-8")) value = observe(pathlib.Path(args.ground).resolve()) | {"diff_vs_pre": diff(pre, observe(pathlib.Path(args.ground).resolve()))} if False else observe(pathlib.Path(args.ground).resolve()) value["diff_vs_pre"] = diff(pre, value) else: pre = json.loads(pathlib.Path(args.pre).read_text(encoding="utf-8")) post = json.loads(pathlib.Path(args.post).read_text(encoding="utf-8")) spec = json.loads(pathlib.Path(args.spec).read_text(encoding="utf-8")) value = witness(pre, post, spec) dump(value, pathlib.Path(args.out)) print(json.dumps(value, ensure_ascii=False)) return 0 if __name__ == "__main__": sys.exit(main())