"""Portable, read-only graph CLI tests; no application-store imports."""

import contextlib
import io
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest

sys.dont_write_bytecode = True
SCRIPTS = Path(__file__).resolve().parents[2] / "skills" / "learnharness" / "scripts"
sys.path.insert(0, str(SCRIPTS))
import graph

GOLDEN = Path(__file__).resolve().parents[1] / "fixtures" / "golden-data"


def snapshot(root):
    """Notice both changed bytes/mtimes and newly created files/directories."""
    return {
        str(path.relative_to(root)): (path.read_bytes(), path.stat().st_mtime_ns) if path.is_file() else None
        for path in root.rglob("*")
    }


def point(chat, message=""):
    return {"chatId": chat, "messageId": message}


def node(identifier, session=None, **updates):
    value = {
        "id": identifier, "name": f"Name {identifier}",
        "statement": f"Full statement {identifier}", "aliases": [],
        "createdAt": "2026-01-01T00:00:00Z",
        "provenance": [] if session is None else [point(session)],
    }
    value.update(updates)
    return value


def chain(identifier, premises, conclusion, session, **updates):
    value = {
        "id": identifier, "premises": premises, "conclusion": conclusion,
        "createdAt": "2026-01-02T00:00:00Z", "provenance": [point(session)],
    }
    value.update(updates)
    return value


class GraphTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)

    def write_json(self, relative, value):
        path = self.root / relative
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")

    def write_ledger(self, name, records):
        path = self.root / "graph" / f"{name}.jsonl"
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text("".join(json.dumps(record, ensure_ascii=False) + "\n" for record in records), encoding="utf-8")

    def invoke(self, *arguments, format="json", root=None):
        out, err = io.StringIO(), io.StringIO()
        with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
            status = graph.main(["--data-dir", str(root or self.root), "--format", format, *arguments])
        return status, out.getvalue(), err.getvalue()

    def result(self, *arguments, root=None):
        status, output, diagnostics = self.invoke(*arguments, root=root)
        self.assertEqual(status, 0, diagnostics)
        self.assertEqual(diagnostics, "")
        payload = json.loads(output)
        self.assertEqual(payload["data_dir"], str((root or self.root).resolve()))
        return payload["records"]

    def fixture(self):
        self.write_ledger("nodes", [
            node("n_b", "chat_old", name="Identity stands", statement="First precise statement", aliases=["Original wording"], provenance=[point("chat_old", "m1")], extra="preserved"),
            node("n_a", "src_notes", name="Café", statement="A separate proposition.", aliases=["Literal [brackets]?", "two\nlines"]),
            node("n_b", "chat_new", name="Name must not replace", statement="Later restatement", aliases=["Original wording", "Another phrasing", ""], createdAt="2099-01-01", provenance=[point("chat_old", "m1"), point("chat_new", "m2"), point("chat_new", "m2")]),
            node("n_empty"),
            node("n_surviving", "chat_new"),
            node("n_unknown", "missing_record"),
        ])
        self.write_ledger("chains", [
            chain("arg_first", ["n_b", "n_a"], "n_surviving", "chat_old", extra="preserved"),
            chain("arg_other_id", ["n_a", "n_b"], "n_surviving", "chat_new", createdAt="2099-01-01"),
            chain("arg_single", ["n_surviving"], "n_a", "src_notes"),
            chain("arg_dangling", ["n_missing"], "n_a", "chat_old"),
        ])
        for identifier, title, updated in [
            ("chat_old", "Old conversation", "2026-02-01T00:00:00Z"),
            ("chat_new", "Recent conversation", "2026-03-01T00:00:00Z"),
            ("chat_without_nodes", "Not a birth session", "2099-01-01"),
        ]:
            self.write_json(f"chats/{identifier}.json", {
                "id": identifier, "title": title, "createdAt": "2026-01-01T00:00:00Z",
                "updatedAt": updated, "messages": [],
            })
        self.write_json("graph/sources.json", {
            "src_notes": {"id": "src_notes", "filePath": "/never/open/these notes.md", "importedAt": "2026-02-15T00:00:00Z"},
        })

    def test_merge_nodes_first_identity_and_union(self):
        self.fixture()
        before = snapshot(self.root)
        merged = self.result("propositions")
        self.assertEqual([item["id"] for item in merged], ["n_a", "n_b", "n_empty", "n_surviving", "n_unknown"])
        value = next(item for item in merged if item["id"] == "n_b")
        self.assertEqual(value["name"], "Identity stands")
        self.assertEqual(value["statement"], "First precise statement")
        self.assertEqual(value["createdAt"], "2026-01-01T00:00:00Z")
        self.assertEqual(value["aliases"], ["Original wording", "Later restatement", "Another phrasing"])
        self.assertEqual(value["provenance"], [point("chat_old", "m1"), point("chat_new", "m2")])
        self.assertEqual(value["extra"], "preserved")
        self.assertEqual(snapshot(self.root), before)

    def test_chain_key_identity_provenance_and_endpoint_filter(self):
        self.fixture()
        merged = self.result("chains")
        self.assertEqual([item["id"] for item in merged], ["arg_first", "arg_single", "arg_dangling"])
        self.assertEqual(merged[0]["premises"], ["n_b", "n_a"])
        self.assertEqual(merged[0]["createdAt"], "2026-01-02T00:00:00Z")
        self.assertEqual(merged[0]["extra"], "preserved")
        self.assertEqual(merged[0]["provenance"], [point("chat_old"), point("chat_new")])
        self.assertEqual(self.result("chains", "--node", "n_b"), merged[:1])
        self.assertEqual(len(self.result("chains", "--node", "n_a")), 3)
        self.assertEqual(self.result("chains", "--node", "no-such-node"), [])

    def test_birth_index_metadata_and_filter(self):
        self.fixture()
        sessions = self.result("sessions")
        self.assertEqual([item["id"] for item in sessions], ["chat_new", "src_notes", "chat_old", "missing_record"])
        self.assertEqual([item["propositions"] for item in sessions], [1, 1, 1, 1])
        self.assertEqual(sessions[0]["kind"], "chat")
        self.assertEqual(sessions[0]["name"], "Recent conversation")
        self.assertEqual(sessions[1]["name"], "/never/open/these notes.md")
        self.assertEqual(sessions[-1], {"id": "missing_record", "name": "missing_record", "kind": "file", "createdAt": None, "updatedAt": None, "propositions": 1})
        self.assertEqual([item["id"] for item in self.result("propositions", "--session", "chat_new")], ["n_surviving"])
        self.assertEqual([item["id"] for item in self.result("propositions", "--session", "chat_old")], ["n_b"])
        self.assertEqual(self.result("propositions", "--session", "nonexistent"), [])

    def test_bad_chat_metadata_is_isolated_and_diagnosed(self):
        self.fixture()
        corrupted = self.root / "chats" / "chat_old.json"
        corrupted.write_text("{torn", encoding="utf-8")
        self.write_json("chats/chat_future.json", {
            "id": "chat_future", "title": "Future metadata", "createdAt": "2026-01-01",
            "updatedAt": "2026-01-01", "version": 2, "messages": [],
        })
        self.write_json("chats/src_notes.json", {
            "id": "src_notes", "title": "Older parts metadata", "createdAt": "2026-01-01",
            "updatedAt": "2026-01-01", "messages": [{"role": "assistant", "parts": []}],
        })
        self.write_json("chats/chat_mismatch.json", {
            "id": "another_id", "title": "Wrong filename", "createdAt": "2026-01-01",
            "updatedAt": "2026-01-01", "messages": [],
        })
        before = snapshot(self.root)
        for arguments, format in [(("sessions",), "json"), (("propositions",), "markdown")]:
            with self.subTest(arguments=arguments):
                status, output, diagnostics = self.invoke(*arguments, format=format)
                self.assertEqual(status, 0, diagnostics)
                for name in ("chat_old.json", "chat_future.json", "src_notes.json", "chat_mismatch.json"):
                    self.assertIn(name, diagnostics)
                for reason in ("format 2", "older UIMessage/parts", "filename/ID mismatch", "skipped"):
                    self.assertIn(reason, diagnostics)
                if format == "json":
                    sessions = {item["id"]: item for item in json.loads(output)["records"]}
                    self.assertEqual(sessions["chat_old"]["name"], "chat_old")
                    self.assertIsNone(sessions["chat_old"]["createdAt"])
                    self.assertEqual(sessions["chat_new"]["name"], "Recent conversation")
                    self.assertEqual(sessions["src_notes"]["name"], "/never/open/these notes.md")
                else:
                    self.assertIn("First precise statement", output)
                    self.assertIn("Recent conversation", output)
                    self.assertIn("/never/open/these notes.md", output)
        self.assertEqual(snapshot(self.root), before)

    def test_explicit_empty_query_and_ids_rejected_but_whitespace_accepted(self):
        for arguments in (("find", ""), ("propositions", "--session", ""), ("chains", "--node", "")):
            with self.subTest(arguments=arguments):
                out, err = io.StringIO(), io.StringIO()
                with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
                    with self.assertRaises(SystemExit) as raised:
                        graph.main(["--data-dir", str(self.root), *arguments])
                self.assertEqual(raised.exception.code, 2)
                self.assertEqual(out.getvalue(), "")
                self.assertIn("must not be empty", err.getvalue())
        for arguments in (("find", " "), ("propositions", "--session", " "), ("chains", "--node", " ")):
            self.assertEqual(self.result(*arguments), [])
        self.assertEqual(list(self.root.iterdir()), [])

    def test_empty_first_provenance_acquires_birth_from_surviving_link(self):
        self.write_ledger("nodes", [node("n_x"), node("n_x", "survivor")])
        self.assertEqual([item["id"] for item in self.result("propositions", "--session", "survivor")], ["n_x"])
        self.assertEqual(self.result("sessions")[0]["id"], "survivor")

    def test_session_order_timestamp_ties_then_stable_id(self):
        self.write_ledger("nodes", [node("n_b", "b"), node("n_a", "a"), node("n_c", "c")])
        self.assertEqual([item["id"] for item in self.result("sessions")], ["a", "b", "c"])

    def test_find_matches_names_statements_aliases_and_literal_punctuation(self):
        self.fixture()
        for query in ("IDENTITY", "precise", "later RESTATEMENT", "another phrasing"):
            self.assertEqual([item["id"] for item in self.result("find", query)], ["n_b"])
        self.assertEqual([item["id"] for item in self.result("find", "[brackets]?")], ["n_a"])
        self.assertEqual([item["id"] for item in self.result("find", "Café")], ["n_a"])
        self.assertEqual(self.result("find", "n_b"), [])  # IDs are not searchable fields.
        self.assertEqual(self.result("find", "Name must not replace"), [])
        self.assertEqual(self.result("find", " no such text "), [])

    def test_find_case_matching_agrees_with_app_js_regex(self):
        self.write_ledger("nodes", [node("n_case", name="Exact label", statement="Exact statement", aliases=["K", "ſ", "ı", "İ", "ς", "𐐨", "Café"])])
        for query in ("k", "s", "i", "𐐀"):
            # Explicit stars avoid unrelated substring matches in the name.
            self.assertEqual(self.result("find", query + "*"), [], query)
        for query in ("K*", "ſ*", "ı*", "İ*", "σ*", "𐐨*", "CAFÉ*"):
            self.assertEqual([item["id"] for item in self.result("find", query)], ["n_case"], query)

    def test_find_star_glob_newlines_and_ledger_order(self):
        self.fixture()
        self.assertEqual([item["id"] for item in self.result("find", "*statement*")], ["n_b", "n_empty", "n_surviving", "n_unknown"])
        self.assertEqual([item["id"] for item in self.result("find", "two*lines")], ["n_a"])
        self.assertEqual(self.result("find", "precise*"), [])
        self.assertEqual(len(self.result("find", "*")), 5)

    def test_missing_ledgers_empty_graph_and_no_directory_creation(self):
        for command in ("sessions", "propositions", "chains"):
            self.assertEqual(self.result(command), [])
        self.assertEqual(self.result("find", "anything"), [])
        self.assertEqual(list(self.root.iterdir()), [])
        status, output, diagnostics = self.invoke("propositions", format="markdown")
        self.assertEqual(status, 0, diagnostics)
        self.assertIn("No matching records", output)
        self.assertEqual(list(self.root.iterdir()), [])

    def test_malformed_and_wrong_shape_lines_warn_instead_of_silent_loss(self):
        self.write_ledger("nodes", [node("n_ok", "source")])
        path = self.root / "graph" / "nodes.jsonl"
        with path.open("a", encoding="utf-8") as handle:
            handle.write('\n{"torn":\n{"id":"bad"}\nnull\n')
        self.write_ledger("chains", [chain("arg_bad", [], "n_ok", "source")])
        before = snapshot(self.root)
        status, output, diagnostics = self.invoke("propositions")
        self.assertEqual(status, 0)
        self.assertEqual([item["id"] for item in json.loads(output)["records"]], ["n_ok"])
        self.assertIn("nodes.jsonl:3: malformed JSON", diagnostics)
        self.assertIn("nodes.jsonl:4: invalid record shape", diagnostics)
        self.assertIn("nodes.jsonl:5: invalid record shape", diagnostics)
        self.assertIn("chains.jsonl:1: invalid record shape", diagnostics)
        self.assertIn("incomplete", diagnostics)
        self.assertEqual(snapshot(self.root), before)

    def test_future_format_refuses_without_partial_output_or_writes(self):
        self.fixture()
        self.write_json("graph/format.json", {"version": 2})
        before = snapshot(self.root)
        for command in ("sessions", "propositions", "chains"):
            status, output, diagnostics = self.invoke(command)
            self.assertNotEqual(status, 0)
            self.assertEqual(output, "")
            self.assertIn("newer", diagnostics)
        self.assertEqual(snapshot(self.root), before)

    def test_current_format_works_and_malformed_json_refuses(self):
        self.write_json("graph/format.json", {"version": 1})
        self.assertEqual(self.result("propositions"), [])
        (self.root / "graph" / "format.json").write_text("{torn", encoding="utf-8")
        status, output, diagnostics = self.invoke("propositions")
        self.assertNotEqual(status, 0)
        self.assertEqual(output, "")
        self.assertIn("format.json", diagnostics)

    def test_source_paths_are_only_references_and_markdown_is_full_text(self):
        self.fixture()
        before = snapshot(self.root)
        status, output, diagnostics = self.invoke("propositions", format="markdown")
        self.assertEqual(status, 0, diagnostics)
        for text in ("First precise statement", "Original wording", "Later restatement", "Another phrasing", "chat_old", "m1", "Recent conversation", "Café", "/never/open/these notes.md"):
            self.assertIn(text, output)
        self.assertIn(f"Data directory: {self.root.resolve()}", output)
        self.assertIn("not mastery or guaranteed truth", output)
        status, output, diagnostics = self.invoke("chains", "--node", "n_a", format="markdown")
        self.assertEqual(status, 0, diagnostics)
        self.assertIn("First precise statement", output)
        self.assertIn("missing proposition", output)
        self.assertEqual(snapshot(self.root), before)

    @unittest.skipUnless(hasattr(os, "symlink"), "symlinks unavailable")
    def test_ledger_symlink_escape_is_rejected(self):
        with tempfile.TemporaryDirectory() as external:
            path = Path(external) / "nodes.jsonl"
            path.write_text(json.dumps(node("n_external")), encoding="utf-8")
            (self.root / "graph").mkdir()
            try:
                (self.root / "graph" / "nodes.jsonl").symlink_to(path)
            except OSError as exc:
                self.skipTest(f"symlink permission unavailable: {exc}")
            status, output, diagnostics = self.invoke("propositions")
            self.assertNotEqual(status, 0)
            self.assertEqual(output, "")
            self.assertTrue(diagnostics)
            self.assertEqual(path.read_text(encoding="utf-8"), json.dumps(node("n_external")))

    def test_cli_root_options_order_missing_root_and_no_bytecode(self):
        env = {**os.environ, "LH_DATA_DIR": str(GOLDEN)}
        env.pop("PYTHONDONTWRITEBYTECODE", None)
        before = snapshot(self.root)
        caches_before = set(SCRIPTS.rglob("*.pyc"))
        command = [sys.executable, str(SCRIPTS / "graph.py")]
        completed = subprocess.run([*command, "--data-dir", str(self.root), "--format", "json", "sessions"], env=env, capture_output=True, text=True, check=False)
        self.assertEqual(completed.returncode, 0, completed.stderr)
        self.assertEqual(json.loads(completed.stdout), {"data_dir": str(self.root.resolve()), "records": []})  # Never falls back to repo/LH_DATA_DIR.
        bad = subprocess.run([*command, "sessions", "--data-dir", str(self.root)], env=env, capture_output=True, text=True, check=False)
        self.assertNotEqual(bad.returncode, 0)
        absent = subprocess.run([*command, "--data-dir", str(self.root / "missing"), "sessions"], env=env, capture_output=True, text=True, check=False)
        self.assertNotEqual(absent.returncode, 0)
        self.assertEqual(absent.stdout, "")
        self.assertEqual(snapshot(self.root), before)
        self.assertEqual(set(SCRIPTS.rglob("*.pyc")), caches_before)

    def test_golden_data_read_only_broad_graph(self):
        before = snapshot(GOLDEN)
        nodes = self.result("propositions", root=GOLDEN)
        chains = self.result("chains", root=GOLDEN)
        sessions = self.result("sessions", root=GOLDEN)
        self.assertGreaterEqual(len(nodes), 250)
        self.assertGreaterEqual(len(chains), 150)
        self.assertEqual(len({item["id"] for item in nodes}), len(nodes))
        self.assertEqual(sum(item["propositions"] for item in sessions), sum(bool(item["provenance"]) for item in nodes))
        for session in sessions:
            born = self.result("propositions", "--session", session["id"], root=GOLDEN)
            self.assertEqual(len(born), session["propositions"])
            self.assertTrue(all(item["provenance"][0]["chatId"] == session["id"] for item in born))
        self.assertTrue(any(item["kind"] == "file" for item in sessions))
        # Golden chat files are not the original graph birth sessions; their
        # missing historical metadata must remain honestly undated.
        self.assertTrue(any(item["createdAt"] is None for item in sessions))
        self.assertEqual(snapshot(GOLDEN), before)


if __name__ == "__main__":
    unittest.main()