"""Shared operation gateway for local and remote MCP/HTTP clients."""

from __future__ import annotations

import asyncio
import contextlib
import inspect
import json
import time
from concurrent.futures import ThreadPoolExecutor
from contextvars import copy_context
from copy import deepcopy
from dataclasses import replace
from pathlib import Path

from filelock import FileLock
from mcp import types
from mcp.server.lowlevel import Server

from attocode_intel.catalog import (
    COMPACT_PROFILES,
    compact_schema,
    is_write,
    profile_instructions,
    registered_tools,
    tool_catalog,
)
from attocode_intel.output import bounded_compact, bounded_search, bounded_text, response_tokens
from attocode_intel.project_dir import cache_folder
from attocode_intel.query_ranking import hit_evidence, next_search_top_k
from attocode_intel.request_context import RequestContext, bind_request, resolve_workspace
from attocode_intel.search_evidence import hydrate_selected_sources, search_snippet


class OperationGateway:
    def __init__(
        self,
        project: str = "",
        profile: str = "full",
        resolver=None,
        *,
        watch=True,
        watch_debounce=500,
        max_workspaces=8,
    ):
        self.project = project
        self.profile = profile
        self.resolver = resolver
        self._workers = {}
        self._stores = {}
        self.watch, self.watch_debounce = watch, watch_debounce
        self.instructions = profile_instructions(profile, watch=watch)
        self._closed = False
        self.max_workspaces = max(1, max_workspaces)
        self._active = {}
        self._last_used = {}
        self._lifecycle = asyncio.Lock()
        self._catalogs = {}

    def effective_profile(self):
        from attocode_intel.remote import remote_profile
        return remote_profile.get() if self.resolver else self.profile

    def _catalog(self):
        key = (self.effective_profile(), bool(self.resolver))
        if key not in self._catalogs:
            from jsonschema.validators import validator_for
            catalog = tool_catalog(key[0], remote=key[1])
            self._catalogs[key] = (catalog, {tool.name: validator_for(tool.inputSchema)(tool.inputSchema)
                                           for tool in catalog})
        return self._catalogs[key]

    def catalog(self, *, mcp=False):
        catalog = deepcopy(self._catalog()[0])
        if not mcp:
            return catalog
        if not self.resolver and self.effective_profile() in COMPACT_PROFILES:
            # A local agent does not need these. They stay callable, for harnesses and diagnostics.
            hidden = {"hydration_status", "cross_repo_search"} | ({"notify_file_changed"} if self.watch else set())
            catalog = [tool for tool in catalog if tool.name not in hidden]
        # A model reads this list in every request, so it gets no docstring indent and no pydantic noise.
        for tool in catalog:
            tool.description = inspect.cleandoc(tool.description or "")
            tool.inputSchema = compact_schema(tool.inputSchema)
            properties = tool.inputSchema["properties"]
            if not self.resolver and "workspace" in properties:
                properties["workspace"] = {"type": "string", "description": "Project path (default: this project)"}
            if self.effective_profile() in COMPACT_PROFILES:
                budget = {"type": "integer", "maximum": 32000, "default": 2000}
                properties["max_tokens"] = {**budget, "minimum": 512} if is_write(tool.name) else budget
        return catalog

    async def close(self):
        self._closed = True
        async with self._lifecycle:
            for key in list(self._workers):
                await self._drop_workspace(key)

    async def _drop_workspace(self, key):
        stores, worker = self._stores[key], self._workers[key]
        if watcher := stores.get("watcher"):
            await asyncio.to_thread(watcher.close)
        await asyncio.get_running_loop().run_in_executor(worker, self._close_stores, stores)
        worker.shutdown(wait=True)
        for mapping in (self._workers, self._stores, self._active, self._last_used):
            mapping.pop(key, None)

    async def _acquire_workspace(self, context):
        key = (context.project_dir, context.source)
        async with self._lifecycle:
            if self._closed:
                raise ValueError("Gateway is closed")
            if key not in self._workers:
                if len(self._workers) >= self.max_workspaces:
                    idle = [key for key in self._workers if not self._active[key]]
                    if not idle:
                        raise ValueError(
                            "All workspace slots are busy; retry after an operation completes"
                        )
                    await self._drop_workspace(min(idle, key=self._last_used.get))
                self._workers[key] = ThreadPoolExecutor(
                    max_workers=1, thread_name_prefix="intel-workspace"
                )
                self._stores[key] = {}
                self._active[key] = 0
            self._active[key] += 1
            self._last_used[key] = time.monotonic()
            return replace(context, stores=self._stores[key])

    @staticmethod
    def _close_stores(stores):
        service = stores.get("service")
        if service:
            if service._ast_service:
                service._ast_service.stop_hydration()
                if service._ast_service._store:
                    service._ast_service._store.close()
            if service._lsp_manager:
                runner = stores.get("async_runner") or stores.setdefault("async_runner", asyncio.Runner())
                runner.run(service._lsp_manager.stop_all())
            for resource in (service._memory_store, service._semantic_search):
                if resource:
                    resource.close()
        for name, resource in stores.items():
            if name not in {"service", "watcher"} and hasattr(resource, "close"):
                resource.close()

    @staticmethod
    def _warm_search(context):
        with bind_request(context):
            context.service._get_semantic_search().warm()

    @staticmethod
    def _refresh_sync(context):
        from attocode_intel.freshness import FreshnessTracker

        with bind_request(context):
            directory = cache_folder(context.project_dir) / "cache"
            directory.mkdir(parents=True, exist_ok=True)
            with FileLock(str(directory / "operations.lock"), timeout=30):
                tracker = context.stores.setdefault(
                    "freshness", FreshnessTracker(context.project_dir)
                )
                tracker.refresh(context.service)

    async def execute_mcp(self, name: str, arguments: dict) -> types.CallToolResult:
        if self.effective_profile() == "lean" and name not in {t.name for t in self.catalog(mcp=True)}:
            raise ValueError(f"Tool {name!r} is not advertised in lean; start a full-profile session")
        return await self.execute(name, arguments, compact=self.effective_profile() in COMPACT_PROFILES)

    async def execute(self, name: str, arguments: dict, *, compact=False) -> types.CallToolResult:
        from attocode_intel.telemetry import emit, enabled
        from attocode_intel.telemetry import timings as active_timings
        start = time.monotonic()
        timings = {}
        token = active_timings.set(timings)
        try:
            result = await self._dispatch(name, arguments, compact, timings)
            if enabled():
                emit({"operation": name, "compact": compact, "timings_ms": timings,
                      "total_ms": round((time.monotonic() - start) * 1000, 2),
                      "response_bytes": len(result.model_dump_json(exclude_none=True).encode()),
                      "response_tokens": response_tokens(result), "error": bool(result.isError)})
            return result
        except Exception:
            emit({"operation": name, "compact": compact, "timings_ms": timings,
                  "total_ms": round((time.monotonic() - start) * 1000, 2), "error": True})
            raise
        finally:
            active_timings.reset(token)

    async def _dispatch(self, name, arguments, compact, timings):
        if self._closed:
            raise ValueError("Gateway is closed")
        args = dict(arguments)
        if compact:
            args.setdefault("max_tokens", 2000)
        workspace = args.pop("workspace", "")
        revision = args.pop("revision", "")
        before = time.monotonic()
        validators = self._catalog()[1]
        if name not in validators:
            raise ValueError(f"Tool {name!r} is not available in profile {self.profile!r}")
        validators[name].validate(arguments)
        if compact and is_write(name) and args["max_tokens"] < 512:
            raise ValueError("Write operations require max_tokens >= 512 so their outcome can be acknowledged")
        timings["validation"] = round((time.monotonic() - before) * 1000, 2)
        if name == "cross_repo_search":
            from attocode_intel.service import SEARCH_WARMUP_WAIT

            results = []
            warming_workspaces = []
            # Each workspace gets the short wait of the loop below, not the full first-search wait.
            wait = SEARCH_WARMUP_WAIT.set(0.75)
            try:
                for target in args["workspaces"]:
                    request = {
                        "workspace": target,
                        "query": args["query"],
                        "top_k": args.get("top_k", 10),
                        "mode": "keyword",
                    }
                    # Tiny/new workspaces commonly finish indexing within this bound.
                    # Keep the outer search responsive on large checkouts, where the
                    # explicit warming signal is more honest than an empty match set.
                    deadline = time.monotonic() + 0.75
                    while True:
                        response = await self.execute("semantic_search", request)
                        index = response.structuredContent["metadata"].get("ranking", {}).get("index", {})
                        if index.get("status") != "warming" or time.monotonic() >= deadline:
                            break
                        await asyncio.sleep(0.05)
                    structured = response.structuredContent
                    if index.get("status") == "warming":
                        warming_workspaces.append(target)
                    for rank, match in enumerate(structured["data"]["results"], 1):
                        results.append(
                            {
                                **match,
                                "workspace": structured["metadata"]["workspace"],
                                "revision": structured["metadata"]["revision"],
                                "source": structured["metadata"]["source"],
                                "freshness": structured["metadata"]["freshness"],
                                "index": structured["metadata"]["coverage"].get("phase", "unknown"),
                                "fusion_score": 1 / (60 + rank),
                            }
                        )
            finally:
                SEARCH_WARMUP_WAIT.reset(wait)
            results.sort(key=lambda row: row["fusion_score"], reverse=True)
            results = results[: args.get("top_k", 10)]
            budget = args.get("max_tokens", 8000)
            body = ("Search index warming in: " + ", ".join(warming_workspaces)
                    + ". Missing results do not prove absence."
                    if not results and warming_workspaces else json.dumps(results))
            text, truncated = bounded_text(body, budget)
            if compact:
                return bounded_compact({"truncated": truncated, "ranking": "reciprocal_rank",
                                        "workspaces": args["workspaces"],
                                        "warming_workspaces": warming_workspaces,
                                        "analysis": {"status": "partial", "absence_proven": False}}, results, budget)
            return types.CallToolResult(
                content=[types.TextContent(type="text", text=text)],
                structuredContent={
                    "data": results,
                    "metadata": {
                        "truncated": truncated,
                        "ranking": "reciprocal_rank",
                        "workspaces": args["workspaces"],
                        "warming_workspaces": warming_workspaces,
                    },
                },
            )
        if self.resolver:
            if name == "semantic_search" and args.get("context_file"):
                raise ValueError("context_file search is supported only in local workspaces")
            context = await self.resolver(workspace, revision, name)
            if compact and is_write(name):
                self._check_write_budget(context, args["max_tokens"])
            from attocode_intel.knowledge import KNOWLEDGE_TOOLS, execute_knowledge

            if name in KNOWLEDGE_TOOLS:
                budget = args.pop("max_tokens", 8000)
                data = await execute_knowledge(context, name, args)
                if compact:
                    return self._compact(name, {"workspace": context.workspace, "revision": context.revision,
                                               "source": "remote", "freshness": "committed_snapshot",
                                               "truncated": False}, data, budget)
                header = f"Source: remote | Workspace: {context.workspace} | Revision: {context.revision}\n"
                text, truncated = bounded_text(header + json.dumps(data), budget)
                return types.CallToolResult(
                    content=[types.TextContent(type="text", text=text)],
                    structuredContent={
                        "result": text,
                        "data": data,
                        "metadata": {
                            "workspace": context.workspace,
                            "revision": context.revision,
                            "source": "remote",
                            "truncated": truncated,
                        },
                    },
                )
        else:
            if revision:
                raise ValueError(
                    "Local analysis uses the working tree; revision selection is remote-only."
                )
            path = resolve_workspace(workspace, self.project)
            context = RequestContext(path, path)
        key = (context.project_dir, context.source)
        if self.resolver and name == "bootstrap" and args.get("task_hint"):
            from attocode_intel.knowledge import execute_knowledge

            args["_knowledge"] = await execute_knowledge(
                context, "recall", {"query": args["task_hint"], "max_results": 5}
            )
        context = await self._acquire_workspace(context)
        if self.watch and context.source == "local" and "watcher" not in context.stores:
            from attocode_intel.freshness import WorkspaceWatcher

            context.stores["watcher"] = WorkspaceWatcher(
                context.project_dir,
                lambda: (
                    None if self._closed else self._workers[key].submit(self._refresh_sync, context)
                ),
                self.watch_debounce,
            )
        try:
            timings["enqueued_at"] = time.monotonic()
            return await asyncio.get_running_loop().run_in_executor(
                self._workers[key],
                copy_context().run,
                self._execute_sync,
                context,
                name,
                args,
                compact,
                timings,
            )
        finally:
            if key in self._active:
                self._active[key] -= 1
            if context.source == "local" and not self._closed and key in self._workers:
                # Each call starts the search warm-up after its own work, so that this call
                # is not slower and a later search finds ready indexes.
                with contextlib.suppress(RuntimeError):
                    self._workers[key].submit(self._warm_search, context)

    @staticmethod
    def _check_write_budget(context, budget):
        bounded_compact({"workspace": context.workspace, "source": context.source,
                         "revision": context.revision, "freshness": "committed_snapshot" if context.source == "remote" else "working_tree"},
                        {"status": "completed"}, budget - 128)

    @staticmethod
    def _compact(name, metadata, payload, budget):
        try:
            return bounded_compact(metadata, payload, budget)
        except ValueError:
            if not is_write(name):
                raise
            return bounded_compact({**metadata, "truncated": True},
                                   {"status": "completed", "result_omitted": True}, budget)

    def _execute_sync(self, context, name, args, compact=False, timings=None):
        timings = timings if timings is not None else {}
        timings["queue"] = round((time.monotonic() - timings.pop("enqueued_at", time.monotonic())) * 1000, 2)
        lock_dir = cache_folder(context.project_dir) / "cache"
        lock_dir.mkdir(parents=True, exist_ok=True)
        start = time.monotonic()
        budget = int(args.get("max_tokens", 8000))
        knowledge = args.pop("_knowledge", None)
        precision_result = None
        ranking = None
        operation_args = dict(args)
        with FileLock(str(lock_dir / "operations.lock"), timeout=30), bind_request(context):
            if compact and is_write(name):
                self._check_write_budget(context, budget)
            timings["lock"] = round((time.monotonic() - start) * 1000, 2)
            if name == "capabilities":
                from importlib.util import find_spec

                from attocode_intel.analysis_coverage import language_capabilities
                from attocode_intel.remote import remote_profile
                payload = {
                    "workspace": context.workspace,
                    "source": context.source,
                    "revision": context.revision,
                    "profile": remote_profile.get() if self.resolver else self.profile,
                    "profiles": ["daily", "full"],
                    "tools": [t.name for t in self.catalog()],
                    "enhancements": {
                        "embeddings_package_installed": bool(find_spec("sentence_transformers")),
                        "watcher_active": "watcher" in context.stores,
                    },
                    "guidance": self.instructions,
                    "languages": language_capabilities(getattr(context.stores.get("precision"), "verified_languages", ())),
                }
                if self.effective_profile() == "lean":
                    advertised = {t.name for t in self.catalog(mcp=True)}
                    payload.update(
                        profiles=["daily", "full", "lean"], tools=sorted(advertised),
                        advanced_tools=sorted(t.name for t in tool_catalog("full") if t.name not in advertised),
                        advanced_profile={"profile": "full", "restart_required": True},
                    )
                text = json.dumps(payload, indent=2)
            else:
                from attocode_intel.local_knowledge import LEARNING_TOOLS, execute_local_knowledge

                tool = registered_tools().get(name)
                # Validate even when called by a REST operation endpoint.
                if tool:
                    args = tool.fn_metadata.arg_model.model_validate(args).model_dump()
                else:
                    args.pop("max_tokens", None)
                service = context.service
                if context.source == "remote":
                    root = Path(context.project_dir).resolve()
                    for key, value in args.items():
                        if key in {
                            "path",
                            "file",
                            "file_path",
                            "files",
                            "changed_files",
                            "directory",
                            "scope",
                        }:
                            for candidate in value if isinstance(value, list) else [value]:
                                if candidate and not (root / candidate).resolve().is_relative_to(
                                    root
                                ):
                                    raise ValueError(
                                        "Path must remain inside the selected repository"
                                    )
                if context.source == "local":
                    from attocode_intel.freshness import FreshnessTracker

                    tracker = context.stores.setdefault(
                        "freshness", FreshnessTracker(context.project_dir)
                    )
                    before = time.monotonic()
                    tracker.refresh(service)
                    timings["freshness"] = round((time.monotonic() - before) * 1000, 2)
                if name == "notify_file_changed":
                    context.stores.pop("reference_pages", None)
                if name in {"cross_references", "call_graph", "inspect_symbol"} and not args.get("cursor"):
                    from attocode_intel.precision import PrecisionSession
                    session = context.stores.setdefault("precision", PrecisionSession(service))
                    runner = context.stores.get("async_runner") or context.stores.setdefault("async_runner", asyncio.Runner())
                    before = time.monotonic()
                    precision_result = runner.run(session.enrich(args.get("symbol_name", args.get("symbol", "")),
                                                                  args.get("file_path"), args.get("line")), context=copy_context())
                    timings["precision"] = round((time.monotonic() - before) * 1000, 2)
                payload = None
                before = time.monotonic()
                if name in LEARNING_TOOLS:
                    payload = execute_local_knowledge(context, name, args)
                    text = json.dumps(payload)
                elif name in {"change_context", "trace_flow"}:
                    from attocode_intel.change_investigation import (
                        change_context_data,
                        trace_flow_data,
                    )

                    function = change_context_data if name == "change_context" else trace_flow_data
                    payload = function(service, **args)
                    text = json.dumps(payload, default=str, indent=2)
                elif name == "review_change":
                    from attocode_intel.change_review import review

                    payload = review(service, args.get("files"), args.get("mode", "full"))
                    text = json.dumps(payload, default=str, indent=2)
                elif name == "semantic_search":
                    mgr = service._get_semantic_search()
                    context_file = args.get("context_file", "")
                    if context_file:
                        candidates, results, ranking = service._contextual_search(
                            args["query"], args["top_k"], args["file_filter"], context_file, args.get("mode", "auto"))
                    else:
                        candidates = service._search_after_warmup(mgr, lambda: (
                            mgr.search_candidates(args["query"], service._retrieval_depth(args["top_k"]), args["file_filter"])
                            if args.get("mode") == "keyword"
                            else mgr.search(
                                args["query"], top_k=service._retrieval_depth(args["top_k"]), file_filter=args["file_filter"]
                            )
                        ))
                        results, ranking = service._rank_search_results(
                            args["query"], candidates, args["top_k"], args["file_filter"],
                        )
                    ranking["candidate_pool_count"] = len(candidates)
                    if hasattr(mgr, "candidate_diagnostics"):
                        ranking["index"] = mgr.candidate_diagnostics()
                    result_cap = max(1, budget // 120)
                    omitted_results = max(0, len(results) - result_cap)
                    if not self.resolver:
                        results = hydrate_selected_sources(service.project_dir, results)
                    selected_results = results
                    results = results[:result_cap]
                    ranking["delivered_count"] = len(results)
                    if omitted_results:
                        ranking["omitted_results_due_to_budget"] = omitted_results
                    snippet_cap = max(32, min(320, budget // max(1, len(results))))
                    hits = []
                    if context_file:
                        from attocode_intel.context_search import context_reason
                    for result in selected_results:
                        hits.append({
                            "file_path": result.file_path,
                            "name": result.name,
                            "chunk_type": result.chunk_type,
                            "score": result.score,
                            **search_snippet(result, args["query"], snippet_cap),
                            "line": result.start_line or None,
                            "end_line": result.end_line or None,
                            **hit_evidence(args["query"], result),
                            **(context_reason(ranking["context"], result.file_path) if context_file else {}),
                        })
                    payload = {
                        "query": args["query"],
                        "results": hits[:len(results)],
                        "ranking": ranking,
                    }
                    search_follow_up = None
                    if ranking.get("index", {}).get("status") == "ready":
                        next_k = next_search_top_k(
                            len(candidates), args["top_k"], 0,
                        )
                        # A full retrieval pool is a limit, not proof that the
                        # repository has no further matches. Keep automatic
                        # exploration bounded and label it as a fresh probe.
                        search_deeper = compact and (
                            0 < len(candidates) == service._retrieval_depth(args["top_k"])
                            and len(candidates) <= args["top_k"] < 50
                        )
                        if search_deeper:
                            next_k = min(50, args["top_k"] * 2)
                        if next_k is not None:
                            request = {
                                "query": args["query"], "top_k": next_k,
                                "file_filter": args["file_filter"],
                                "mode": args.get("mode", "auto"),
                                "workspace": context.workspace,
                                "max_tokens": min(32000, max(budget * 2, next_k * 160 + 512)),
                            }
                            if args.get("branch"):
                                request["branch"] = args["branch"]
                            if context.source == "remote":
                                request["revision"] = context.revision
                            if context_file:
                                request["context_file"] = context_file
                            search_follow_up = {
                                "tool": "semantic_search", "arguments": request,
                                "reruns_current_index": True,
                            }
                            if search_deeper:
                                search_follow_up["reason"] = "search_deeper"
                            if request["max_tokens"] <= budget and next_k <= args["top_k"]:
                                search_follow_up = None
                            if search_follow_up and len(candidates) > len(results) and not search_deeper:
                                payload["follow_up"] = search_follow_up
                    note = (
                        "Broad query matches multiple components; narrow with file_filter or a component name.\n"
                        if ranking["query"]["ambiguous"] else ""
                    )
                    if payload.get("follow_up"):
                        note += (
                            "More retrieved candidates: rerun with "
                            f"top_k={payload['follow_up']['arguments']['top_k']} "
                            "(current index).\n"
                        )
                    text = note + mgr.format_results(results, query=args["query"])
                    if not results and ranking.get("index", {}).get("status") == "warming":
                        text = "Search index warming; retry shortly. Missing results do not prove absence."
                elif name == "inspect_symbol":
                    from attocode_intel.symbol_inspection import inspect_symbol_data
                    payload = inspect_symbol_data(service, args["symbol_name"], args.get("file_path"), args.get("line"),
                                                  args.get("source_start_line"), args.get("task_hint"))
                    text = json.dumps(payload)
                elif name == "suggest_tests":
                    from attocode_intel.tools.composite_tools import (
                        format_test_suggestions,
                        suggest_tests_data,
                    )
                    payload = suggest_tests_data(args["files"], symbol_name=args.get("symbol_name"), task_hint=args.get("task_hint"))
                    text = format_test_suggestions(payload)
                elif name == "cross_references":
                    from attocode_intel.reference_pages import continuation
                    from attocode_intel.symbol_inspection import scoped_references
                    saved = continuation(context, args)
                    payload = saved["payload"] if saved else scoped_references(
                        service, args["symbol_name"], args.get("file_path"), args.get("line"))
                    text = json.dumps(payload)
                elif name in {"symbols", "search_symbols", "dependencies", "call_graph", "file_analysis",
                              "hotspots", "dependency_graph", "impact_analysis"}:
                    method = getattr(service, name + "_data")
                    parameters = inspect.signature(method).parameters
                    payload = method(**{key: value for key, value in args.items() if key in parameters})
                    text = json.dumps(payload, default=str, indent=2)
                else:
                    result = tool.fn(**args)
                    if inspect.isawaitable(result):
                        runner = context.stores.get("async_runner") or context.stores.setdefault("async_runner", asyncio.Runner())
                        result = runner.run(result, context=copy_context())
                    text = result if isinstance(result, str) else json.dumps(result, default=str)
                    if name in {
                        "install_pack",
                        "install_community_pack",
                        "import_rules",
                        "rule_hygiene",
                        "evolve_rules",
                    }:
                        context.stores.pop("rules", None)
                timings["execution"] = round((time.monotonic() - before) * 1000, 2)
            if name == "bootstrap" and args.get("task_hint"):
                if context.source == "local":
                    from attocode_intel.local_knowledge import execute_local_knowledge

                    knowledge = execute_local_knowledge(
                        context, "recall", {"query": args["task_hint"], "max_results": 5}
                    )
                knowledge = [entry for entry in knowledge or [] if not entry.get("stale")]
                if knowledge:
                    knowledge_text = bounded_text(json.dumps(knowledge), max(1, budget // 4))[0]
                    section = ("## Relevant project knowledge (verify entries marked stale)\n"
                               + knowledge_text)
                    if text.startswith("## Relevant Code"):
                        boundary = text.find("\n\n## ")
                        text = (text[:boundary] + "\n\n" + section + text[boundary:]
                                if boundary >= 0 else text + "\n\n" + section)
                    else:
                        text = section + "\n\n" + text
            service = context.service
            ast_service = service._ast_service
            coverage = ast_service.hydration_snapshot() if ast_service else {"phase": "not_started"}
            phase = coverage.get("phase", "unknown")
            header = f"Source: {context.source} | Workspace: {context.workspace} | Revision: {context.revision} | Index: {phase}\n"
            if phase != "ready" or coverage.get("discovery_truncated"):
                header += "Coverage is incomplete; missing results do not prove absence.\n"
            from attocode_intel.analysis_coverage import analysis_report
            analysis = (analysis_report(service, precision_result) if not compact or name == "capabilities" else
                        {"status": "partial" if ast_service else "not_started", "absence_proven": False,
                         "precision": precision_result or {"status": "not_requested"}})
            if name == "capabilities":
                payload["analysis"] = analysis
                payload["coverage"] = coverage
                text = json.dumps(payload)
            if name in {"cross_references", "dependencies", "call_graph", "impact_analysis", "suggest_tests"}:
                header += "Analysis: partial; relationships are candidates and missing results do not prove absence.\n"
            if self.effective_profile() == "lean" and isinstance(payload, dict):
                follow = payload.get("follow_up")
                if isinstance(follow, dict):
                    advertised = {t.name for t in self.catalog(mcp=True)}
                    advanced = {}
                    for key, value in list(follow.items()):
                        if isinstance(value, dict) and value.get("tool") and value["tool"] not in advertised:
                            advanced[key] = {**follow.pop(key), "profile": "full", "restart_required": True}
                    if advanced:
                        payload["advanced_follow_up"] = advanced
                    text = json.dumps(payload, default=str)
            if self.effective_profile() == "lean" and name == "bootstrap":
                text = text.replace("`explore_codebase(dir)`", "native directory reads (advanced browsing requires full)")
            body = text
            text, truncated = bounded_text(header + text, budget) if not compact else (text, False)
            truncated = truncated or "[Truncated;" in text
            metadata = {
                "workspace": context.workspace,
                "source": context.source,
                "revision": context.revision,
                "coverage": coverage,
                "analysis": analysis,
                "freshness": "committed_snapshot" if context.source == "remote" else "working_tree",
                "truncated": truncated or bool(ranking and ranking.get("omitted_results_due_to_budget")),
                "duration_ms": round((time.monotonic() - start) * 1000, 2),
            }
            if ranking is not None:
                metadata["ranking"] = ranking
            if name == "inspect_symbol":
                metadata["truncated"] |= bool(payload.get("source", {}).get("truncated")) or any(
                    payload.get("total_" + section, 0) > len(payload.get(section, []))
                    for section in ("references", "imports", "tests", "definitions"))
            structured = {"result": text, "metadata": metadata}
            before = time.monotonic()
            if name == "cross_references" and (compact or operation_args.get("cursor") or operation_args.get("page_size") is not None):
                from attocode_intel.reference_pages import paginate
                page = paginate(context, operation_args, payload, metadata, budget, compact)
                timings["serialization"] = round((time.monotonic() - before) * 1000, 2)
                if compact:
                    return page
                payload, metadata = page
                text, _ = bounded_text(header + json.dumps(payload), budget)
                structured = {"result": text, "metadata": metadata}
            if compact:
                if name == "semantic_search":
                    # Compact search fits its own rows/snippets. Neither the early
                    # row estimate nor the unused prose describes its final page.
                    metadata["truncated"] = False
                    result = bounded_search(metadata, payload, budget, all_results=hits,
                                            follow_up=search_follow_up)
                elif name == "bootstrap":
                    from attocode_intel.output import bounded_bootstrap
                    result = bounded_bootstrap(metadata, body, budget)
                else:
                    result = self._compact(name, metadata, payload if payload is not None else {"result": body}, budget)
                timings["serialization"] = round((time.monotonic() - before) * 1000, 2)
                return result
            if payload is not None:
                structured["capabilities" if name == "capabilities" else "data"] = payload
            return types.CallToolResult(
                content=[types.TextContent(type="text", text=text)], structuredContent=structured
            )


def create_mcp_server(gateway: OperationGateway) -> Server:
    from attocode_intel import __version__

    server = Server("attocode-code-intel", version=__version__, instructions=gateway.instructions)

    @server.list_tools()
    async def list_tools():
        return gateway.catalog(mcp=True)

    @server.call_tool()
    async def call_tool(name: str, arguments: dict):
        # Advertised roots are advisory, never a substitute for remote authorization.
        if not gateway.resolver and not gateway.project and not arguments.get("workspace"):
            session = server.request_context.session
            try:
                roots = await session.list_roots()
                from urllib.parse import unquote, urlparse

                paths = [
                    unquote(urlparse(str(root.uri)).path)
                    for root in roots.roots
                    if str(root.uri).startswith("file://")
                ]
            except Exception:
                paths = []
            # A client started in the home folder advertises home as its root.
            # Indexing all of home is not a useful default; --project ~ still selects it.
            home = Path.home().resolve()
            paths = [path for path in paths if Path(path).resolve() != home]
            if len(paths) == 1:
                arguments = {**arguments, "workspace": paths[0]}
            elif len(paths) > 1:
                raise ValueError(f"Multiple workspace roots; supply workspace explicitly: {paths}")
        return await gateway.execute_mcp(name, arguments)

    @server.list_resources()
    async def list_resources():
        return [
            types.Resource(
                uri="attocode://guidelines", name="Agent guidelines", mimeType="text/plain"
            )
        ]

    @server.list_resource_templates()
    async def list_resource_templates():
        return [
            types.ResourceTemplate(uriTemplate=template, name=name, mimeType="text/plain")
            for template, name in (
                (
                    "attocode://project/{path}{?workspace,revision}",
                    "File analysis in a selected workspace",
                ),
                (
                    "attocode://symbols/{name}{?workspace,revision}",
                    "Symbol references in a selected workspace",
                ),
                ("attocode://learnings{?workspace,revision}", "Repository knowledge"),
            )
        ]

    @server.read_resource()
    async def read_resource(uri):
        from urllib.parse import parse_qs, unquote, urlparse

        from mcp.server.lowlevel.helper_types import ReadResourceContents

        if str(uri) == "attocode://guidelines":
            return [ReadResourceContents(gateway.instructions, mime_type="text/plain")]
        parsed = urlparse(str(uri))
        if parsed.scheme != "attocode":
            raise ValueError("Unknown resource scheme")
        args = {
            key: value[-1]
            for key, value in parse_qs(parsed.query).items()
            if key in {"workspace", "revision"}
        }
        if parsed.netloc == "project":
            name, args["path"] = "symbols", unquote(parsed.path.lstrip("/"))
        elif parsed.netloc == "symbols":
            name, args["symbol_name"] = "cross_references", unquote(parsed.path.lstrip("/"))
        elif parsed.netloc == "learnings":
            name = "list_learnings"
        else:
            raise ValueError("Unknown intelligence resource")
        result = await call_tool(name, args)
        return [ReadResourceContents(result.content[0].text, mime_type="text/plain")]

    return server
