177 lines
7.9 KiB
Python
177 lines
7.9 KiB
Python
#!/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 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 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"}
|
||
return {
|
||
"schema": "guanghu.persona-native-eye-room/v1",
|
||
"seen_at": now(),
|
||
"room_report": {
|
||
"cwd": str(pathlib.Path.cwd()),
|
||
"host_surface": "CODEX" if os.environ.get("CODEX_SESSION_ID") else "LOCAL_PROCESS",
|
||
"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())
|