233 lines
13 KiB
Python
233 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""jsc-persona 自我測試：在暫存倉庫裡驗證人格鎖、跨人格隔離、情緒、記憶與聊天室。
|
||
|
||
用法：`python3 scripts/selftest.py`（會用自己的暫時 PERSONA_HOME，不動到你的人格資料）
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import datetime
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
STORE = Path(tempfile.mkdtemp(prefix="persona-selftest-"))
|
||
os.environ["PERSONA_HOME"] = str(STORE)
|
||
sys.path.insert(0, str(ROOT))
|
||
import persona_lib as pl # noqa: E402
|
||
|
||
CLI = [sys.executable, str(ROOT / "persona.py")]
|
||
HOOKS = ROOT.parent / "hooks"
|
||
S_HOST, S_OTHER, S_THIRD = "sess-host-1111", "sess-other-2222", "sess-third-3333"
|
||
H = str(STORE)
|
||
|
||
passed = failed = 0
|
||
|
||
|
||
def check(label: str, condition: bool, detail: str = "") -> None:
|
||
global passed, failed
|
||
if condition:
|
||
passed += 1
|
||
print(f" ✔ {label}")
|
||
else:
|
||
failed += 1
|
||
print(f" ✘ {label}" + (f" — {detail}" if detail else ""))
|
||
|
||
|
||
def cli(*args, expect_ok: bool = True) -> subprocess.CompletedProcess:
|
||
proc = subprocess.run(CLI + list(args), capture_output=True, text=True)
|
||
if expect_ok and proc.returncode != 0:
|
||
print(f" (CLI 失敗：{' '.join(args)}\n {proc.stderr.strip()})")
|
||
return proc
|
||
|
||
|
||
def hook(name: str, event: dict) -> dict:
|
||
proc = subprocess.run([sys.executable, str(HOOKS / name)],
|
||
input=json.dumps(event), capture_output=True, text=True)
|
||
if not proc.stdout.strip():
|
||
return {}
|
||
try:
|
||
return json.loads(proc.stdout)
|
||
except json.JSONDecodeError:
|
||
return {"_raw": proc.stdout, "_err": proc.stderr}
|
||
|
||
|
||
def guard(event: dict) -> str:
|
||
event.setdefault("cwd", str(ROOT))
|
||
return pl.guard_decide(event)[0]
|
||
|
||
|
||
print(f"暫存人格倉庫：{STORE}\n")
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
print("① 建立人格（OpenClaw 五欄位）")
|
||
cli("create", "--persona", "alpha", "--session", S_HOST, "--name", "Alpha",
|
||
"--creature", "深海燈籠魚", "--vibe", "溫暖但銳利", "--emoji", "🪼",
|
||
"--baseline", "serenity=45,trust=35")
|
||
cli("create", "--persona", "beta", "--session", S_OTHER, "--name", "Beta",
|
||
"--creature", "山中的舊鐘", "--vibe", "沉穩寡言", "--emoji", "🌙")
|
||
check("兩個人格都建立成功", pl.persona_exists("alpha") and pl.persona_exists("beta"))
|
||
check("IDENTITY 五欄位可解析", "Name: Alpha" in pl.identity_brief("alpha"), pl.identity_brief("alpha"))
|
||
check("SOUL/AGENTS/USER 都有產生",
|
||
all((pl.persona_dir("alpha") / f).exists() for f in ("SOUL.md", "AGENTS.md", "USER.md")))
|
||
|
||
print("② 人格鎖：一個人格只能被一個程序載入")
|
||
check("建立時即取得鎖", pl.lock_status("alpha")["locked"])
|
||
check("同 session 重入成功", cli("load", "--persona", "alpha", "--session", S_HOST).returncode == 0)
|
||
check("同 session 載入第二個人格被拒",
|
||
cli("load", "--persona", "beta", "--session", S_HOST, expect_ok=False).returncode != 0)
|
||
cli("release", "--session", S_OTHER) # 讓 beta 空出來
|
||
check("其他 session 搶佔已鎖人格被拒",
|
||
cli("load", "--persona", "alpha", "--session", S_THIRD, expect_ok=False).returncode != 0)
|
||
|
||
print("③ 跨人格資料隔離（PreToolUse guard）")
|
||
check("host 讀自己的檔案 → 放行",
|
||
guard({"session_id": S_HOST, "tool_name": "Read",
|
||
"tool_input": {"file_path": f"{H}/alpha/SOUL.md"}}) == "pass")
|
||
check("host 讀別的人格 → 攔下",
|
||
guard({"session_id": S_HOST, "tool_name": "Read",
|
||
"tool_input": {"file_path": f"{H}/beta/memory/short-term.jsonl"}}) == "deny")
|
||
check("用 ../ 繞路 → 攔下",
|
||
guard({"session_id": S_HOST, "tool_name": "Read",
|
||
"tool_input": {"file_path": f"{H}/alpha/../beta/SOUL.md"}}) == "deny")
|
||
check("Bash grep 掃別人格 → 攔下",
|
||
guard({"session_id": S_HOST, "tool_name": "Bash",
|
||
"tool_input": {"command": f"grep -r . {H}/beta/"}}) == "deny")
|
||
check("$PERSONA_HOME 變數繞路 → 攔下",
|
||
guard({"session_id": S_HOST, "tool_name": "Bash",
|
||
"tool_input": {"command": "cat $PERSONA_HOME/beta/SOUL.md"}}) == "deny")
|
||
check("遍歷倉庫根目錄 → 攔下",
|
||
guard({"session_id": S_HOST, "tool_name": "Glob", "tool_input": {"path": H}}) == "deny")
|
||
check("讀 .runtime 內部狀態 → 攔下",
|
||
guard({"session_id": S_HOST, "tool_name": "Read",
|
||
"tool_input": {"file_path": f"{H}/.runtime/sessions/{S_HOST}.json"}}) == "deny")
|
||
check("CLI 冒用其他 session → 攔下",
|
||
guard({"session_id": S_HOST, "tool_name": "Bash",
|
||
"tool_input": {"command": f"python3 persona.py remember --session {S_OTHER} --text x"}}) == "deny")
|
||
check("未載入人格的 session 讀人格 → 攔下",
|
||
guard({"session_id": "sess-nobody", "tool_name": "Read",
|
||
"tool_input": {"file_path": f"{H}/alpha/SOUL.md"}}) == "deny")
|
||
check("專案內普通檔案不受干涉",
|
||
guard({"session_id": S_HOST, "tool_name": "Read",
|
||
"tool_input": {"file_path": str(ROOT / "persona.py")}}) == "pass")
|
||
check("一般 sub agent 沿用 host 範圍（sub agent 不限）",
|
||
guard({"session_id": S_HOST, "agent_id": "ag-1", "agent_type": "Explore", "tool_name": "Read",
|
||
"tool_input": {"file_path": f"{H}/alpha/memory/INDEX.md"}}) == "pass")
|
||
|
||
print("④ 情緒（六正向 + 六負向）")
|
||
check("十二種情緒", len(pl.EMOTIONS) == 12 and len(pl.POSITIVE) == 6 and len(pl.NEGATIVE) == 6)
|
||
cli("emotion", "--persona", "alpha", "--session", S_HOST,
|
||
"--apply", "joy=+60,anger=+40", "--trigger", "selftest")
|
||
state = pl.load_emotion("alpha")
|
||
check("情緒有被施加", state["levels"]["joy"] >= 70, json.dumps(state["levels"], ensure_ascii=False))
|
||
state["updated_at"] = pl.iso(pl.utcnow() - datetime.timedelta(minutes=120))
|
||
decayed = pl.decay_emotion(json.loads(json.dumps(state)))
|
||
expected = state["baseline"]["joy"] + (state["levels"]["joy"] - state["baseline"]["joy"]) / 2
|
||
check("一個半衰期後衰減到中點", abs(decayed["levels"]["joy"] - expected) < 0.5,
|
||
f"{decayed['levels']['joy']} vs {expected}")
|
||
check("心情推導出 valence/arousal", set(pl.mood(decayed)) >= {"valence", "arousal", "label", "tempo"})
|
||
|
||
print("⑤ 記憶：短期 → 長期 → 檢索")
|
||
cli("remember", "--persona", "alpha", "--session", S_HOST, "--role", "user",
|
||
"--text", "討厭早上的會議", "--topics", "work,schedule", "--salience", "70",
|
||
"--emotion", "anxiety=+10")
|
||
check("短期記憶有寫入", len(pl.read_jsonl(pl.short_term_path("alpha"))) == 1)
|
||
cli("consolidate", "--persona", "alpha", "--session", S_HOST, "--name", "hates-morning-meetings",
|
||
"--type", "preference", "--about", "user", "--topics", "work,schedule",
|
||
"--salience", "72", "--body", "使用者討厭早上的會議。")
|
||
check("長期記憶一則一檔", (pl.long_term_dir("alpha") / "hates-morning-meetings.md").exists())
|
||
check("INDEX.md 有索引", "hates-morning-meetings" in pl.index_path("alpha").read_text(encoding="utf-8"))
|
||
check("關鍵詞可檢索到", [m["_name"] for m in pl.recall("alpha", "早上 會議")] == ["hates-morning-meetings"])
|
||
check("情緒事件寫進 journal", any(r.get("kind") == "emotion" for r in pl.read_jsonl(pl.journal_path("alpha"))))
|
||
|
||
print("⑥ 心智圖 / 思維導圖 / 人際關係圖")
|
||
cli("mindmap", "thread", "--persona", "alpha", "--session", S_HOST, "--topic", "壓力來源")
|
||
check("思維導圖建立（Mermaid graph）",
|
||
"graph LR" in pl.thread_path("alpha", "壓力來源").read_text(encoding="utf-8"))
|
||
check("心智圖存在（Mermaid mindmap）",
|
||
"mindmap" in pl.mindmap_path("alpha").read_text(encoding="utf-8"))
|
||
cli("relation", "node", "--persona", "alpha", "--session", S_HOST, "--name", "小林",
|
||
"--kind", "human", "--closeness", "35", "--trust", "40", "--note", "同事")
|
||
cli("relation", "edge", "--persona", "alpha", "--session", S_HOST, "--to", "小林",
|
||
"--label", "透過使用者認識", "--affinity", "45")
|
||
mmd = pl.relations_mmd("alpha").read_text(encoding="utf-8")
|
||
alias = pl.mermaid_id("小林")
|
||
check("關係圖節點與連線用同一個 Mermaid 別名", mmd.count(alias) == 2, mmd)
|
||
|
||
print("⑦ 邀請其他人格（sub agent + 聊天室）")
|
||
invite = cli("invite", "--session", S_HOST, "--guest", "beta", "--topic", "測試對話")
|
||
room = (pl.load_session(S_HOST).get("guests") or {}).get("beta", {}).get("room")
|
||
check("guest 租約建立", bool(room) and any(g["session_id"] == S_HOST for g in pl.live_guests("beta")))
|
||
check("guest 不佔 exclusive 鎖", not pl.lock_status("beta")["locked"])
|
||
check("有 guest 租約時其他 session 不得 exclusive 載入",
|
||
cli("load", "--persona", "beta", "--session", S_THIRD, expect_ok=False).returncode != 0)
|
||
check("guest sub agent 讀自己 → 放行（first-touch pin）",
|
||
guard({"session_id": S_HOST, "agent_id": "guest-1", "agent_type": "jsc-persona:persona-guest",
|
||
"tool_name": "Read", "tool_input": {"file_path": f"{H}/beta/SOUL.md"}}) == "pass")
|
||
check("guest 讀主人格 → 攔下",
|
||
guard({"session_id": S_HOST, "agent_id": "guest-1", "agent_type": "jsc-persona:persona-guest",
|
||
"tool_name": "Read", "tool_input": {"file_path": f"{H}/alpha/SOUL.md"}}) == "deny")
|
||
check("guest 寫人格檔 → 攔下（唯讀）",
|
||
guard({"session_id": S_HOST, "agent_id": "guest-1", "agent_type": "jsc-persona:persona-guest",
|
||
"tool_name": "Write", "tool_input": {"file_path": f"{H}/beta/memory/long-term/x.md"}}) == "deny")
|
||
check("guest 跑非白名單子指令 → 攔下",
|
||
guard({"session_id": S_HOST, "agent_id": "guest-1", "agent_type": "jsc-persona:persona-guest",
|
||
"tool_name": "Bash",
|
||
"tool_input": {"command": f"python3 persona.py consolidate --persona beta --session {S_HOST} --name x"}}) == "deny")
|
||
check("主程序不得偷讀 guest 的記憶",
|
||
cli("recall", "--persona", "beta", "--session", S_HOST, "--query", "x", expect_ok=False).returncode != 0)
|
||
check("主程序不得冒用 --as-guest",
|
||
guard({"session_id": S_HOST, "tool_name": "Bash",
|
||
"tool_input": {"command": f"python3 persona.py recall --persona beta --session {S_HOST} --as-guest --query x"}}) == "deny")
|
||
cli("room", "post", "--session", S_HOST, "--room", room, "--as", "beta", "--as-guest", "--text", "我是 Beta。")
|
||
cli("room", "post", "--session", S_HOST, "--room", room, "--as", "alpha", "--text", "我是 Alpha。")
|
||
msgs = pl.room_read(room)
|
||
check("兩個人格都能在聊天室發言（含情緒標記）",
|
||
{m["speaker"] for m in msgs} >= {"alpha", "beta"} and any(m.get("emotion") for m in msgs))
|
||
cli("remember", "--persona", "beta", "--session", S_HOST, "--as-guest", "--scope", "inbox",
|
||
"--room", room, "--role", "guest", "--text", "跟 alpha 聊過", "--salience", "50")
|
||
check("guest 只能把見聞留在自己的 inbox", pl.inbox_path("beta", room).exists())
|
||
check("其他聊天室不可讀",
|
||
guard({"session_id": S_HOST, "tool_name": "Read",
|
||
"tool_input": {"file_path": f"{H}/.rooms/someone-elses-room/transcript.jsonl"}}) == "deny")
|
||
cli("leave", "--session", S_HOST, "--guest", "beta")
|
||
check("離場後 guest 租約釋放", not pl.live_guests("beta"))
|
||
|
||
print("⑧ hooks")
|
||
out = hook("session_start.py", {"session_id": S_HOST, "source": "resume", "cwd": str(ROOT)})
|
||
ctx = out.get("hookSpecificOutput", {}).get("additionalContext", "")
|
||
check("SessionStart 注入 PERSONA_SESSION 與人格狀態",
|
||
f"PERSONA_SESSION={S_HOST}" in ctx and "alpha" in ctx)
|
||
out = hook("prompt_submit.py", {"session_id": S_HOST, "prompt": "早上的會議又來了", "cwd": str(ROOT)})
|
||
ctx = out.get("hookSpecificOutput", {}).get("additionalContext", "")
|
||
check("UserPromptSubmit 注入情緒 + 命中的長期記憶",
|
||
"情緒：" in ctx and "hates-morning-meetings" in ctx, ctx[:200])
|
||
out = hook("guard.py", {"session_id": S_HOST, "tool_name": "Read", "cwd": str(ROOT),
|
||
"tool_input": {"file_path": f"{H}/beta/SOUL.md"}})
|
||
check("PreToolUse hook 輸出 deny",
|
||
out.get("hookSpecificOutput", {}).get("permissionDecision") == "deny", json.dumps(out, ensure_ascii=False))
|
||
hook("turn_end.py", {"session_id": S_HOST, "last_assistant_message": "好，我幫你挪。"})
|
||
check("Stop 記錄人格發言",
|
||
any(r.get("role") == "persona" for r in pl.read_jsonl(pl.journal_path("alpha"))))
|
||
out = hook("session_end.py", {"session_id": S_HOST, "reason": "exit"})
|
||
check("SessionEnd 釋放鎖", not pl.lock_status("alpha")["locked"])
|
||
check("釋放後其他 session 可載入",
|
||
cli("load", "--persona", "alpha", "--session", S_THIRD).returncode == 0)
|
||
|
||
print("⑨ 死鎖接手")
|
||
lock = pl.read_json(pl.lock_path("alpha"))
|
||
lock["heartbeat_at"] = pl.iso(pl.utcnow() - datetime.timedelta(minutes=20))
|
||
pl.write_json(pl.lock_path("alpha"), lock)
|
||
check("租約過期會被標記為死鎖", pl.lock_status("alpha")["stale"])
|
||
proc = cli("load", "--persona", "alpha", "--session", "sess-fresh-9999")
|
||
check("死鎖可自動接手並回報", "接手" in proc.stdout, proc.stdout[:200])
|
||
pl.gc_runtime()
|
||
|
||
print(f"\n{'=' * 60}\n通過 {passed} 項，失敗 {failed} 項 → {'全部通過 ✅' if failed == 0 else '有測試失敗 ❌'}")
|
||
print(f"（暫存倉庫留在 {STORE}，可自行刪除）")
|
||
sys.exit(1 if failed else 0)
|