"""selftest.py — contract tests for the ai-os plugin's identity + privacy guard.

Run:  python3 core/plugin/ai-os/selftest.py
Loads identity.py + guard.py under a synthetic package so relative imports work
even though the plugin dir name ("ai-os") isn't a valid Python identifier.
Builds a throwaway bundle in a temp dir and asserts the guard blocks cross-user
access while allowing own-user and non-user paths.
"""
import os
import sys
import json
import types
import tempfile
import importlib.util

# UTF-8 stdout (Windows mặc định cp1252 → lỗi khi in nhãn tiếng Việt; Linux đã UTF-8)
for _s in (sys.stdout, sys.stderr):
    try:
        _s.reconfigure(encoding="utf-8", errors="replace")
    except Exception:
        pass

HERE = os.path.dirname(os.path.abspath(__file__))


def _load_pkg():
    pkg = types.ModuleType("aiosplugin")
    pkg.__path__ = [HERE]
    sys.modules["aiosplugin"] = pkg

    def _load(name):
        spec = importlib.util.spec_from_file_location("aiosplugin." + name, os.path.join(HERE, name + ".py"))
        m = importlib.util.module_from_spec(spec)
        sys.modules["aiosplugin." + name] = m
        spec.loader.exec_module(m)
        return m

    mods = {n: _load(n) for n in ("identity", "guard", "state", "personas", "team")}
    return mods


class _MockCtx:
    """Captures register_* calls so we can assert wiring without a live Hermes."""
    def __init__(self):
        self.tools, self.hooks, self.commands = {}, [], {}

    def register_tool(self, name, toolset, schema, handler, description="", **kw):
        self.tools[name] = handler

    def register_hook(self, event, cb):
        self.hooks.append(event)

    def register_command(self, name, handler, description="", args_hint="", **kw):
        self.commands[name] = handler


def main() -> int:
    mods = _load_pkg()
    ident, guard, state, personas, team = (
        mods["identity"], mods["guard"], mods["state"], mods["personas"], mods["team"])
    ok = True

    def check(label, cond):
        nonlocal ok
        print(("PASS" if cond else "FAIL") + ": " + label)
        ok = ok and cond

    tmp = tempfile.mkdtemp(prefix="aios-selftest-")
    users = os.path.join(tmp, "users")
    for sub in ("_template/ai_mentor", "tg_1/ai_mentor", "tg_2/ai_mentor"):
        os.makedirs(os.path.join(users, sub), exist_ok=True)
    open(os.path.join(users, "tg_2/ai_mentor/progress.md"), "w").close()

    # Pin bundle root + identity to tg_1 (telegram).
    os.environ["AI_OS_BUNDLE_ROOT"] = tmp
    os.environ["HERMES_SESSION_PLATFORM"] = "telegram"
    os.environ["HERMES_SESSION_USER_ID"] = "1"
    os.environ.pop("HERMES_SESSION_CHAT_ID", None)

    check("resolve_user -> tg_1", ident.resolve_user() == "tg_1")

    # Identity bridge: stash_identity must make resolve_user work via os.environ
    # (simulates the pre_gateway_dispatch bridge for slash-command context).
    for k in ("HERMES_SESSION_PLATFORM", "HERMES_SESSION_USER_ID"):
        os.environ.pop(k, None)
    check("resolve_user None without env", ident.resolve_user() is None)
    ident.stash_identity(platform="telegram", user_id="1", chat_id="1")
    check("stash_identity -> resolve tg_1", ident.resolve_user() == "tg_1")

    # Guardrail: path escape rejected.
    try:
        ident.user_path("tg_1", "../tg_2/x")
        check("user_path escape blocked", False)
    except ValueError:
        check("user_path escape blocked", True)

    # Guard: write into another user's folder -> BLOCK.
    r = guard.check("write_file", {"path": os.path.join(users, "tg_2/ai_mentor/x.md")})
    check("write to tg_2 blocked", isinstance(r, dict) and r.get("action") == "block")

    # Guard: write into own folder -> ALLOW.
    r = guard.check("write_file", {"path": os.path.join(users, "tg_1/ai_mentor/x.md")})
    check("write to own tg_1 allowed", r is None)

    # Guard: read shared content (outside users/) -> ALLOW.
    r = guard.check("read_file", {"path": os.path.join(tmp, "content/ai-bos/x.md")})
    check("read content/ allowed", r is None)

    # Guard: _template is shared scaffolding -> ALLOW.
    r = guard.check("read_file", {"path": os.path.join(users, "_template/ai_mentor/profile.md")})
    check("read _template allowed", r is None)

    # Guard: terminal command touching another user's folder -> BLOCK.
    r = guard.check("terminal", {"command": "cat users/tg_2/ai_mentor/progress.md"})
    check("terminal cat tg_2 blocked", isinstance(r, dict) and r.get("action") == "block")

    # Guard: terminal command in own folder -> ALLOW.
    r = guard.check("terminal", {"command": "ls users/tg_1/ai_mentor"})
    check("terminal ls tg_1 allowed", r is None)

    # --- P3: state tools (engine-scoped) ---
    w = json.loads(state.write_state({"persona": "ai_mentor", "file": "progress.md", "content": "lesson_01: done"}))
    check("state write own ok", w.get("ok") is True)
    rd = json.loads(state.read_state({"persona": "ai_mentor", "file": "progress.md"}))
    check("state read own ok", rd.get("ok") and rd.get("content") == "lesson_01: done")
    bad = json.loads(state.write_state({"persona": "_template", "file": "x.md", "content": "y"}))
    check("state rejects _template persona", bad.get("ok") is False)

    # --- P4: persona registry ---
    pdir = os.path.join(tmp, "personas")
    os.makedirs(pdir, exist_ok=True)
    with open(os.path.join(pdir, "ai_mentor.yaml"), "w", encoding="utf-8") as f:
        f.write("id: ai_mentor\ndisplay_name: AI Mentor\npersonality: ai_mentor\nstatus: ready\n"
                "tiers: [starter, pro]\naccess:\n  org_roles: [ceo, manager, staff]\n")
    with open(os.path.join(pdir, "business_mentor.yaml"), "w", encoding="utf-8") as f:
        f.write("id: business_mentor\ndisplay_name: Business Mentor\npersonality: business_mentor\nstatus: stub\n"
                "tiers: [pro]\naccess:\n  org_roles: [ceo, manager]\n")
    with open(os.path.join(pdir, "_template.yaml"), "w", encoding="utf-8") as f:
        f.write("id: <persona_id>\n")  # must be skipped
    loaded = personas.load_personas()
    check("personas loaded (2, _template skipped)", set(loaded) == {"ai_mentor", "business_mentor"})
    ident.upsert_index("tg_1", org_role="staff")
    staff_list = {p["id"] for p in personas.list_for_role("staff")}
    check("staff sees only ai_mentor", staff_list == {"ai_mentor"})
    out = personas.cmd_mentor("business_mentor")
    check("staff blocked from business_mentor", "chưa được mở" in out)
    ident.upsert_index("tg_1", org_role="ceo")
    out = personas.cmd_mentor("business_mentor")
    check("ceo selects business_mentor", "Business Mentor" in out and personas.active_persona("tg_1") == "business_mentor")
    check("unknown id handled", "Không thấy" in personas.cmd_mentor("nope"))

    # --- P5: team progress (admin-gated) ---
    ident.upsert_index("tg_1", org_role="ceo")
    rep = json.loads(team.team_progress({}))
    check("ceo team_progress ok", rep.get("ok") is True and rep.get("count", 0) >= 1)
    ident.upsert_index("tg_1", org_role="staff")
    rep = json.loads(team.team_progress({}))
    check("staff team_progress forbidden", rep.get("ok") is False and "forbidden" in rep.get("error", ""))

    # --- register() wiring ---
    init_spec = importlib.util.spec_from_file_location("aiosplugin.__plugin__", os.path.join(HERE, "__init__.py"))
    init_mod = importlib.util.module_from_spec(init_spec)
    sys.modules["aiosplugin.__plugin__"] = init_mod
    init_spec.loader.exec_module(init_mod)
    ctx = _MockCtx()
    init_mod.register(ctx)
    check("hooks wired", set(ctx.hooks) == {"pre_tool_call", "on_session_start", "pre_gateway_dispatch"})
    check("state+admin tools wired",
          {"mentor_state_read", "mentor_state_write", "mentor_state_list", "team_progress_read"} <= set(ctx.tools))
    check("commands wired", {"aiwhoami", "experts", "mentor"} <= set(ctx.commands))

    # --- tier gating (gap fix): không hiện "ghost" expert ở tier thấp ---
    tdir = os.path.join(tmp, "tiers")
    os.makedirs(tdir, exist_ok=True)
    open(os.path.join(tdir, "starter.yaml"), "w").write("tier: starter\nextends: null\n")
    open(os.path.join(tdir, "starter_plus.yaml"), "w").write("tier: starter_plus\nextends: starter\n")
    open(os.path.join(tdir, "pro.yaml"), "w").write("tier: pro\nextends: starter_plus\n")
    ident.upsert_index("tg_1", org_role="ceo")
    # instance=starter → business_mentor (tiers[pro]) BỊ ẨN; ai_mentor (tiers[starter,pro]) HIỆN
    open(os.path.join(tmp, "BUILD.txt"), "w").write("tier: starter\n")
    check("tier_chain(starter)={starter}", personas.tier_chain("starter") == {"starter"})
    ceo_starter = {p["id"] for p in personas.list_for_role("ceo")}
    check("starter ẩn ghost business_mentor", "business_mentor" not in ceo_starter and "ai_mentor" in ceo_starter)
    check("mentor business_mentor chặn ở starter", "gói cao hơn" in personas.cmd_mentor("business_mentor"))
    # instance=pro → business_mentor hiện lại (chain pro⊇starter_plus⊇starter)
    open(os.path.join(tmp, "BUILD.txt"), "w").write("tier: pro\n")
    check("tier_chain(pro) gồm starter_plus", "starter_plus" in personas.tier_chain("pro"))
    check("pro hiện business_mentor", "business_mentor" in {p["id"] for p in personas.list_for_role("ceo")})
    # /mentor parse: NBSP + câu hỏi kèm trên cùng dòng → vẫn lấy đúng id, mời gửi lại câu hỏi
    r = personas.cmd_mentor("business_mentor\xa0\xa0\"câu hỏi test\"")
    check("mentor parse NBSP+question", "Đã chuyển sang" in r and "gửi lại câu hỏi" in r)
    # Nhận diện thông minh: tên 1 phần → đúng; mơ hồ → hỏi lại; không khớp → báo
    check("mentor fuzzy 'business' -> chọn", "Đã chuyển sang" in personas.cmd_mentor("business"))
    check("mentor 'BUSINESS' (hoa) -> chọn", "Đã chuyển sang" in personas.cmd_mentor("BUSINESS"))
    check("mentor mơ hồ 'mentor' -> hỏi lại", "ý anh/chị là ai" in personas.cmd_mentor("mentor"))
    check("mentor không khớp", "Không thấy" in personas.cmd_mentor("xyzkhong"))

    print("SELFTEST_OK" if ok else "SELFTEST_FAIL")
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(main())