from __future__ import annotations

import asyncio
import hashlib
import re
import shutil
import sys
import tempfile
import unittest
from contextlib import AsyncExitStack
from pathlib import Path

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from docforge.changesets import ChangesetStore
from docforge.context import compile_context
from docforge.index import ProjectIndex
from docforge.project import Project
from docforge.rendering import RenderService

ROOT = Path(__file__).resolve().parents[1]
FIXTURES = ROOT / "tests" / "fixtures"


def copy_fixture(name: str, destination: Path) -> Path:
    root = destination / name
    shutil.copytree(FIXTURES / name, root)
    return root


def canonical_digest(root: Path) -> str:
    paths = [
        root / ".docforge/project.toml",
        *(path for path in root.rglob("*") if path.is_file() and ".docforge" not in path.parts),
    ]
    digest = hashlib.sha256()
    for path in sorted(set(paths), key=lambda item: item.relative_to(root).as_posix()):
        digest.update(path.relative_to(root).as_posix().encode())
        digest.update(path.read_bytes())
    return digest.hexdigest()


def node_hash(project: Project, node_id: str) -> str:
    return next(node.content_hash for node in project.load().nodes if node.node_id == node_id)


class UnrelatedProjectProofTests(unittest.TestCase):
    def test_generic_ski_project_completes_the_full_workflow(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = copy_fixture("awesome-ski-game", Path(directory))
            before = canonical_digest(root)
            project = Project.open(root)
            snapshot = project.load()

            self.assertEqual("generic", project.descriptor.adapter)
            self.assertEqual("awesome-ski-game", snapshot.descriptor.project_id)
            self.assertEqual(
                {"proof", "riding", "safety", "session", "trail"},
                {node.family for node in snapshot.nodes},
            )
            self.assertEqual(5, len(snapshot.nodes))
            self.assertEqual(6, len(snapshot.edges))

            index = ProjectIndex(project)
            built = index.build()
            self.assertEqual(built, index.check())
            self.assertEqual(
                ["proof.carve-playtest", "riding.edge-control"],
                [item["node_id"] for item in index.search("grip carve", limit=5)["results"]],
            )
            self.assertEqual(
                ["safety.avalanche-window"],
                [
                    item["node_id"]
                    for item in index.filter_nodes(family="safety", status="current")["results"]
                ],
            )
            self.assertEqual(
                {
                    "riding.edge-control",
                    "safety.avalanche-window",
                    "trail.glacier-bowl",
                },
                {
                    item["node_id"]
                    for item in index.dependencies("session.first-descent", depth=3)["results"]
                },
            )

            first_context = compile_context(index, "ride-day", budget=300)
            second_context = compile_context(index, "ride-day", budget=300)
            self.assertEqual(first_context, second_context)
            self.assertLessEqual(first_context["estimated_tokens"], 300)
            self.assertEqual("session.first-descent", first_context["entries"][0]["node_id"])
            self.assertTrue(first_context["omissions"])

            changesets = ChangesetStore(project, "ski-editor")
            created = changesets.create("edge-tuning")
            updated = changesets.propose_update(
                changeset_id="edge-tuning",
                expected_changeset_hash=created["changeset_hash"],
                node_id="riding.edge-control",
                expected_content_hash=node_hash(project, "riding.edge-control"),
                metadata={"summary": "Proposes a narrower grip transition for review."},
                content="<script>unsafe()</script>\n\n**Previewed edge-tuning proposal.**",
                relationship_changes=[],
                rationale="Exercise an unrelated project's isolated proposal path.",
            )
            validated = changesets.validate("edge-tuning")
            diff = changesets.diff("edge-tuning")
            self.assertTrue(validated["valid"])
            self.assertEqual(updated["changeset_hash"], validated["changeset_hash"])
            self.assertEqual("update", diff["changes"][0]["operation"])

            rendering = RenderService(project, changesets)
            first_preview = rendering.preview("edge-tuning", "field-guide")
            preview_path = root / ".docforge/previews/edge-tuning/field-guide.html"
            preview_bytes = preview_path.read_bytes()
            second_preview = rendering.preview("edge-tuning", "field-guide")
            self.assertEqual(first_preview["preview_identity"], second_preview["preview_identity"])
            self.assertEqual(preview_bytes, preview_path.read_bytes())
            preview = preview_bytes.decode()
            self.assertIn("&lt;script&gt;unsafe()&lt;/script&gt;", preview)
            self.assertIn("<strong>Previewed edge-tuning proposal.</strong>", preview)
            self.assertFalse((root / ".docforge/rendered/field-guide.html").exists())
            self.assertEqual(before, canonical_digest(root))

    def test_generic_core_contains_no_project_specific_vocabulary(self) -> None:
        banned = re.compile(r"\b(worldforge|assetforge|phase|villager)\b", re.IGNORECASE)
        leaks = []
        for path in sorted((ROOT / "src/docforge").rglob("*.py")):
            for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
                if banned.search(line):
                    leaks.append(f"{path.name}:{line_number}:{line.strip()}")
        self.assertEqual([], leaks)


class UnrelatedProjectIsolationTests(unittest.IsolatedAsyncioTestCase):
    async def test_two_live_servers_remain_project_isolated(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)
            ski_root = copy_fixture("awesome-ski-game", parent / "ski")
            alpha_root = copy_fixture("alpha", parent / "alpha")
            ski_before = canonical_digest(ski_root)
            alpha_before = canonical_digest(alpha_root)
            ski_project = Project.open(ski_root)
            alpha_project = Project.open(alpha_root)
            ProjectIndex(ski_project).build()
            ProjectIndex(alpha_project).build()

            ski_parameters = StdioServerParameters(
                command=sys.executable,
                args=[
                    "-m",
                    "docforge.mcp_server",
                    "--project-root",
                    str(ski_root),
                    "--proposal-writer",
                    "ski-editor",
                ],
            )
            alpha_parameters = StdioServerParameters(
                command=sys.executable,
                args=[
                    "-m",
                    "docforge.mcp_server",
                    "--project-root",
                    str(alpha_root),
                    "--proposal-writer",
                    "alpha-editor",
                ],
            )

            async with AsyncExitStack() as stack:
                ski_read, ski_write = await stack.enter_async_context(stdio_client(ski_parameters))
                alpha_read, alpha_write = await stack.enter_async_context(
                    stdio_client(alpha_parameters)
                )
                ski_session = await stack.enter_async_context(ClientSession(ski_read, ski_write))
                alpha_session = await stack.enter_async_context(
                    ClientSession(alpha_read, alpha_write)
                )
                await asyncio.gather(ski_session.initialize(), alpha_session.initialize())

                ski_info, alpha_info = await asyncio.gather(
                    ski_session.call_tool("docforge_project_info", {}),
                    alpha_session.call_tool("docforge_project_info", {}),
                )
                ski_own, alpha_own, ski_cross, alpha_cross = await asyncio.gather(
                    ski_session.call_tool(
                        "docforge_get_node", {"node_id": "session.first-descent"}
                    ),
                    alpha_session.call_tool("docforge_get_node", {"node_id": "guide.workflow"}),
                    ski_session.call_tool("docforge_get_node", {"node_id": "guide.workflow"}),
                    alpha_session.call_tool(
                        "docforge_get_node", {"node_id": "session.first-descent"}
                    ),
                )
                ski_result, alpha_result = await asyncio.gather(
                    self._propose_and_preview(
                        ski_session,
                        node_id="riding.edge-control",
                        content_hash=node_hash(ski_project, "riding.edge-control"),
                        view_id="field-guide",
                        summary="A ski-only proposal from its bound server.",
                    ),
                    self._propose_and_preview(
                        alpha_session,
                        node_id="guide.workflow",
                        content_hash=node_hash(alpha_project, "guide.workflow"),
                        view_id="manual",
                        summary="An Alpha-only proposal from its bound server.",
                    ),
                )

            self.assertEqual("awesome-ski-game", ski_info.structuredContent["project_id"])
            self.assertEqual("alpha-docs", alpha_info.structuredContent["project_id"])
            self.assertEqual("current", ski_info.structuredContent["index_health"]["state"])
            self.assertEqual("current", alpha_info.structuredContent["index_health"]["state"])
            self.assertNotEqual(
                ski_info.structuredContent["project_root_fingerprint"],
                alpha_info.structuredContent["project_root_fingerprint"],
            )
            self.assertNotEqual(
                ski_info.structuredContent["source_hash"],
                alpha_info.structuredContent["source_hash"],
            )
            self.assertEqual("session.first-descent", ski_own.structuredContent["node"]["node_id"])
            self.assertEqual("guide.workflow", alpha_own.structuredContent["node"]["node_id"])
            self.assertEqual("missing_node", ski_cross.structuredContent["error"]["code"])
            self.assertEqual("awesome-ski-game", ski_cross.structuredContent["project_id"])
            self.assertEqual("missing_node", alpha_cross.structuredContent["error"]["code"])
            self.assertEqual("alpha-docs", alpha_cross.structuredContent["project_id"])
            self.assertEqual("ski-editor", ski_result["creator"])
            self.assertEqual("alpha-editor", alpha_result["creator"])

            ski_changeset = ski_root / ".docforge/changesets/parallel-edit.json"
            alpha_changeset = alpha_root / ".docforge/changesets/parallel-edit.json"
            ski_preview = ski_root / ".docforge/previews/parallel-edit/field-guide.html"
            alpha_preview = alpha_root / ".docforge/previews/parallel-edit/manual.html"
            ski_index = ski_root / ".docforge/cache/index.sqlite3"
            alpha_index = alpha_root / ".docforge/cache/index.sqlite3"
            self.assertTrue(ski_index.is_file())
            self.assertTrue(alpha_index.is_file())
            self.assertTrue(ski_index.resolve().is_relative_to(ski_root))
            self.assertTrue(alpha_index.resolve().is_relative_to(alpha_root))
            self.assertTrue(ski_changeset.is_file())
            self.assertTrue(alpha_changeset.is_file())
            self.assertTrue(ski_preview.is_file())
            self.assertTrue(alpha_preview.is_file())
            ski_html = ski_preview.read_text(encoding="utf-8")
            alpha_html = alpha_preview.read_text(encoding="utf-8")
            self.assertIn('data-project="awesome-ski-game"', ski_html)
            self.assertNotIn('data-project="alpha-docs"', ski_html)
            self.assertIn('data-project="alpha-docs"', alpha_html)
            self.assertNotIn('data-project="awesome-ski-game"', alpha_html)
            self.assertFalse((ski_root / ".docforge/rendered/field-guide.html").exists())
            self.assertFalse((alpha_root / ".docforge/rendered/manual.html").exists())
            self.assertEqual(ski_before, canonical_digest(ski_root))
            self.assertEqual(alpha_before, canonical_digest(alpha_root))

    async def _propose_and_preview(
        self,
        session: ClientSession,
        *,
        node_id: str,
        content_hash: str,
        view_id: str,
        summary: str,
    ) -> dict[str, object]:
        created = await session.call_tool(
            "docforge_create_changeset", {"changeset_id": "parallel-edit"}
        )
        updated = await session.call_tool(
            "docforge_propose_node_update",
            {
                "changeset_id": "parallel-edit",
                "expected_changeset_hash": created.structuredContent["changeset_hash"],
                "node_id": node_id,
                "expected_content_hash": content_hash,
                "metadata": {"summary": summary},
                "content": None,
                "relationship_changes": [],
                "rationale": "Prove simultaneous project-bound proposal isolation.",
            },
        )
        validated = await session.call_tool(
            "docforge_validate_changeset", {"changeset_id": "parallel-edit"}
        )
        preview = await session.call_tool(
            "docforge_preview_changeset",
            {"changeset_id": "parallel-edit", "view_id": view_id},
        )
        self.assertEqual("ok", updated.structuredContent["status"])
        self.assertTrue(validated.structuredContent["valid"])
        self.assertEqual("current", preview.structuredContent["state"])
        return {
            "creator": created.structuredContent["creator"],
            "changeset_hash": updated.structuredContent["changeset_hash"],
            "preview_identity": preview.structuredContent["preview_identity"],
        }


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