client.py

"""Glean Client REST API wrapper.

Implements the surfaces documented at https://developers.glean.com for the
Client API. When no token is configured, the client returns realistic mock
responses so you can try every command offline.

Endpoints covered (all POST unless noted):
  /chat                      Chat with the Glean Assistant
  /search                    Search the Glean index
  /autocomplete              Query autocomplete
  /recommendations           Recommended results for a user
  /feedback                  Send feedback on a result or chat turn
  /agents/search             List or search agents
  /agents/runs/wait          Run an agent and wait for the final response
  /agents/runs/stream        Run an agent with streaming output
  /tools/list                List callable tools
  /tools/call                Invoke a tool
  /getdocuments              Fetch documents by id or URL
  /getdocpermissions         Fetch permissions for a document
  /listentities              List entities (people, teams, etc.)
  /people                    Get a person profile
  /createannouncement        Create an announcement
  /deleteannouncement        Delete an announcement
  /listcollections           List collections
  /createcollection          Create a collection
  /listpins                  List pinned results
  /pin                       Create a pinned result
"""
from __future__ import annotations

import json
import time
import uuid
from typing import Any, Dict, List, Optional

try:
    import urllib.request
    import urllib.error
    import urllib.parse
except Exception:  # pragma: no cover
    urllib = None  # type: ignore

from . import flow as _flow
from . import mock_corpus
from . import personal
from . import __version__
from .config import Config

# Sent on every request. Derived from the package version so a version bump
# cannot leave a stale string behind, which is exactly what happened while this
# was hardcoded as "glean-code/0.1".
USER_AGENT = f"glean-code/{__version__}"


class GleanError(Exception):
    pass


class GleanClient:
    def __init__(self, config: Config):
        self.config = config

    # ---------------- low level ----------------

    def _headers(self) -> Dict[str, str]:
        h = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": USER_AGENT,
        }
        if self.config.effective_api_token:
            h["Authorization"] = f"Bearer {self.config.effective_api_token}"
        if self.config.act_as:
            h["X-Glean-ActAs"] = self.config.act_as
        return h

    def _capture(self, path: str, body: Dict[str, Any], resp: Dict[str, Any]) -> None:
        """Hand the exchange to the flow mapper, if capture is on for this mode.

        Every Client API call funnels through _post, so this one hook covers
        the REPL and the MCP server alike. flow.record never raises: a capture
        bug must not fail a user's command.
        """
        setting = getattr(self.config, "flow_capture", _flow.DEFAULT_CAPTURE)
        if _flow.capture_enabled(setting, self.config.effective_mode):
            _flow.record(self.config, path, body, resp)

    def _post(self, path: str, body: Dict[str, Any],
              query: Optional[Dict[str, Any]] = None,
              method: str = "POST") -> Dict[str, Any]:
        """Call a Client API endpoint.

        `query` holds parameters the spec defines as query-string rather than
        body fields; `method` covers the handful of GET endpoints. Mock and
        local dispatch stay keyed on the bare `path`, so neither is affected.
        """
        if self.config.effective_mode == "local":
            resp = _local_response(path, body)
            self._capture(path, body, resp)
            return resp
        if self.config.effective_mode == "mock":
            mock_corpus.use_path(self.config.mock_corpus_path)
            try:
                resp = _mock_response(path, body)
            except mock_corpus.CorpusError as e:
                raise GleanError(str(e)) from None
            self._capture(path, body, resp)
            return resp

        base = self.config.effective_base_url
        if not base:
            raise GleanError("No base URL configured. Run /login or /config set instance <name>.")
        url = f"{base}{path}"
        if query:
            pairs = {k: v for k, v in query.items() if v is not None}
            if pairs:
                url += "?" + urllib.parse.urlencode(pairs, doseq=True)
        # GET endpoints must not carry a body, or the server rejects the request.
        data = None if method == "GET" else json.dumps(body).encode("utf-8")
        req = urllib.request.Request(url, data=data, headers=self._headers(), method=method)
        try:
            with urllib.request.urlopen(req, timeout=60) as resp:
                raw = resp.read().decode("utf-8")
                if not raw:
                    return {}
                parsed = json.loads(raw)
                self._capture(path, body, parsed)
                return parsed
        except urllib.error.HTTPError as e:
            body_text = e.read().decode("utf-8", "replace")
            raise GleanError(f"HTTP {e.code} from {path}: {body_text}") from None
        except urllib.error.URLError as e:
            raise GleanError(f"Network error calling {path}: {e.reason}") from None

    # ---------------- indexing API (low level) ----------------

    def _indexing_headers(self) -> Dict[str, str]:
        return {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": USER_AGENT,
            "Authorization": f"Bearer {self.config.effective_indexing_token}",
        }

    def _indexing_post(self, path: str, body: Dict[str, Any]) -> Dict[str, Any]:
        return self._indexing_request("POST", self.config.effective_indexing_base_url, path, body)

    def _indexing_request(
        self,
        method: str,
        base: Optional[str],
        path: str,
        body: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        if self.config.effective_mode == "local":
            raise GleanError(
                "The Indexing API pushes content into a Glean tenant, which local "
                "mode has no equivalent of. Use /personal index <folder> to add "
                "content to the local index, or /mode live to reach your tenant."
            )
        if not self.config.effective_indexing_token:
            raise GleanError(
                "No indexing token set. Add one with: /config set indexing_token <token-or-secure-ref>"
            )
        if self.config.effective_mode == "mock":
            return _mock_indexing_response(path, body or {}, method=method)
        if not base:
            raise GleanError("No instance configured for indexing API calls.")
        url = f"{base}{path}"
        data = json.dumps(body).encode("utf-8") if body is not None else None
        req = urllib.request.Request(
            url, data=data, headers=self._indexing_headers(), method=method
        )
        try:
            with urllib.request.urlopen(req, timeout=60) as resp:
                raw = resp.read().decode("utf-8")
                return json.loads(raw) if raw else {}
        except urllib.error.HTTPError as e:
            body_text = e.read().decode("utf-8", "replace")
            raise GleanError(f"HTTP {e.code} from indexing{path}: {body_text}") from None
        except urllib.error.URLError as e:
            raise GleanError(f"Network error calling indexing{path}: {e.reason}") from None

    def datasource_status(self, datasource: str) -> Dict[str, Any]:
        """POST /api/index/v1/debug/{datasource}/status — beta endpoint."""
        return self._indexing_post(f"/debug/{datasource}/status", {})

    def rotate_indexing_token(self) -> Dict[str, Any]:
        """POST /api/index/v1/rotatetoken — rotates the indexing token secret."""
        return self._indexing_post("/rotatetoken", {})

    # ---- Tier 1: read-only inspection ----

    def get_datasource_config(self, datasource: str) -> Dict[str, Any]:
        return self._indexing_post("/getdatasourceconfig", {"datasource": datasource})

    def get_document_status(self, datasource: str, object_type: str, doc_id: str) -> Dict[str, Any]:
        return self._indexing_post(
            "/getdocumentstatus",
            {"datasource": datasource, "objectType": object_type, "docId": doc_id},
        )

    def get_document_count(self, datasource: str) -> Dict[str, Any]:
        return self._indexing_post("/getdocumentcount", {"datasource": datasource})

    def get_user_count(self, datasource: str) -> Dict[str, Any]:
        return self._indexing_post("/getusercount", {"datasource": datasource})

    def check_document_access(self, datasource: str, object_type: str,
                              doc_id: str, user_email: str) -> Dict[str, Any]:
        return self._indexing_post("/checkdocumentaccess", {
            "datasource": datasource, "objectType": object_type,
            "docId": doc_id, "userEmail": user_email,
        })

    def debug_document(self, datasource: str, object_type: str, doc_id: str) -> Dict[str, Any]:
        return self._indexing_post(
            f"/debug/{datasource}/document",
            {"objectType": object_type, "docId": doc_id},
        )

    def debug_documents(self, datasource: str, items: List[Dict[str, Any]]) -> Dict[str, Any]:
        return self._indexing_post(
            f"/debug/{datasource}/documents",
            {"debugDocuments": items},
        )

    def debug_user(self, datasource: str, email: str) -> Dict[str, Any]:
        return self._indexing_post(f"/debug/{datasource}/user", {"email": email})

    # ---- Tier 3: single-record CRUD ----

    def index_document(self, document: Dict[str, Any],
                       version: Optional[int] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"document": document}
        if version:
            body["version"] = version
        return self._indexing_post("/indexdocument", body)

    def delete_document(self, datasource: str, object_type: str, doc_id: str,
                        version: Optional[int] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"datasource": datasource, "objectType": object_type, "id": doc_id}
        if version:
            body["version"] = version
        return self._indexing_post("/deletedocument", body)

    def update_permissions(self, body: Dict[str, Any]) -> Dict[str, Any]:
        return self._indexing_post("/updatepermissions", body)

    def index_user(self, datasource: str, user: Dict[str, Any],
                   version: Optional[int] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"datasource": datasource, "user": user}
        if version:
            body["version"] = version
        return self._indexing_post("/indexuser", body)

    def delete_user(self, datasource: str, email: str,
                    version: Optional[int] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"datasource": datasource, "email": email}
        if version:
            body["version"] = version
        return self._indexing_post("/deleteuser", body)

    def index_group(self, datasource: str, group: Dict[str, Any],
                    version: Optional[int] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"datasource": datasource, "group": group}
        if version:
            body["version"] = version
        return self._indexing_post("/indexgroup", body)

    def delete_group(self, datasource: str, group_name: str,
                     version: Optional[int] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"datasource": datasource, "groupName": group_name}
        if version:
            body["version"] = version
        return self._indexing_post("/deletegroup", body)

    def index_membership(self, datasource: str, membership: Dict[str, Any],
                         version: Optional[int] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"datasource": datasource, "membership": membership}
        if version:
            body["version"] = version
        return self._indexing_post("/indexmembership", body)

    def delete_membership(self, datasource: str, membership: Dict[str, Any],
                          version: Optional[int] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"datasource": datasource, "membership": membership}
        if version:
            body["version"] = version
        return self._indexing_post("/deletemembership", body)

    # ---- Tier 5: bulk + process-all ----

    def index_documents(self, body: Dict[str, Any]) -> Dict[str, Any]:
        return self._indexing_post("/indexdocuments", body)

    def bulk_index_documents(self, body: Dict[str, Any]) -> Dict[str, Any]:
        return self._indexing_post("/bulkindexdocuments", body)

    def bulk_index_users(self, body: Dict[str, Any]) -> Dict[str, Any]:
        return self._indexing_post("/bulkindexusers", body)

    def bulk_index_groups(self, body: Dict[str, Any]) -> Dict[str, Any]:
        return self._indexing_post("/bulkindexgroups", body)

    def bulk_index_memberships(self, body: Dict[str, Any]) -> Dict[str, Any]:
        return self._indexing_post("/bulkindexmemberships", body)

    def bulk_index_employees(self, body: Dict[str, Any]) -> Dict[str, Any]:
        return self._indexing_post("/bulkindexemployees", body)

    def bulk_index_teams(self, body: Dict[str, Any]) -> Dict[str, Any]:
        return self._indexing_post("/bulkindexteams", body)

    def bulk_index_shortcuts(self, body: Dict[str, Any]) -> Dict[str, Any]:
        return self._indexing_post("/bulkindexshortcuts", body)

    def upload_shortcuts(self, body: Dict[str, Any]) -> Dict[str, Any]:
        return self._indexing_post("/uploadshortcuts", body)

    def index_employee_list(self, employees: List[Dict[str, Any]]) -> Dict[str, Any]:
        return self._indexing_post("/indexemployeelist", {"employees": employees})

    def process_all_documents(self, datasource: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {}
        if datasource:
            body["datasource"] = datasource
        return self._indexing_post("/processalldocuments", body)

    def process_all_memberships(self, datasource: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {}
        if datasource:
            body["datasource"] = datasource
        return self._indexing_post("/processallmemberships", body)

    def process_all_employees_teams(self) -> Dict[str, Any]:
        return self._indexing_post("/processallemployeesandteams", {})

    # ---------------- custom metadata API ----------------

    def set_metadata_schema(self, group: str,
                            metadata_keys: List[Dict[str, Any]]) -> Dict[str, Any]:
        return self._indexing_request(
            "PUT", self.config.effective_metadata_base_url,
            f"/custom-metadata/schema/{group}",
            {"metadataKeys": metadata_keys},
        )

    def get_metadata_schema(self, group: str) -> Dict[str, Any]:
        return self._indexing_request(
            "GET", self.config.effective_metadata_base_url,
            f"/custom-metadata/schema/{group}",
        )

    def delete_metadata_schema(self, group: str) -> Dict[str, Any]:
        return self._indexing_request(
            "DELETE", self.config.effective_metadata_base_url,
            f"/custom-metadata/schema/{group}",
        )

    def attach_metadata(self, doc_id: str, group: str,
                        custom_metadata: List[Dict[str, Any]]) -> Dict[str, Any]:
        return self._indexing_request(
            "PUT", self.config.effective_metadata_base_url,
            f"/document/{doc_id}/custom-metadata/{group}",
            {"customMetadata": custom_metadata},
        )

    def detach_metadata(self, doc_id: str, group: str) -> Dict[str, Any]:
        return self._indexing_request(
            "DELETE", self.config.effective_metadata_base_url,
            f"/document/{doc_id}/custom-metadata/{group}",
        )

    # ---------------- chat ----------------

    def chat(self, message: str, chat_id: Optional[str] = None,
             agent: Optional[str] = None, stream: bool = False) -> Dict[str, Any]:
        body = {
            "messages": [{"author": "USER", "messageType": "CONTENT",
                          "fragments": [{"text": message}]}],
            "stream": stream,
        }
        if chat_id:
            body["chatId"] = chat_id
        if agent:
            body["agentConfig"] = {"agent": agent}
        return self._post("/chat", body)

    # ---------------- search ----------------

    def search(self, query: str, page_size: int = 10,
               datasource: Optional[str] = None,
               request_options: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"query": query, "pageSize": page_size}
        if datasource:
            body["requestOptions"] = {"datasourceFilter": datasource}
        if request_options:
            body.setdefault("requestOptions", {}).update(request_options)
        return self._post("/search", body)

    def autocomplete(self, query: str) -> Dict[str, Any]:
        return self._post("/autocomplete", {"query": query})

    def list_datasources(self, sample_size: int = 100) -> Dict[str, Any]:
        """Derive the set of datasources the caller can see.

        The Glean Client REST API does not expose a dedicated 'list
        datasources' endpoint. This method runs a broad search and reads
        the datasource field (and facet counts when present) off the
        response, so it respects the caller's permissions.
        """
        body = {
            "query": "*",
            "pageSize": sample_size,
            "requestOptions": {"facetBucketSize": 50,
                                "facetFilters": [],
                                "facets": ["datasource"]},
        }
        resp = self._post("/search", body)

        sources: Dict[str, int] = {}
        # Prefer facet data if the server returned it
        for facet in (resp.get("facetResults") or []):
            if facet.get("sourceName") == "datasource" or facet.get("name") == "datasource":
                for b in facet.get("buckets", []):
                    name = b.get("value") or b.get("name")
                    if name:
                        sources[name] = int(b.get("count", 0))
                break
        # Fallback: count datasources seen in the result list
        if not sources:
            for r in resp.get("results", []):
                ds = r.get("datasource")
                if ds:
                    sources[ds] = sources.get(ds, 0) + 1
        return {"datasources": [{"name": k, "count": v}
                                 for k, v in sorted(sources.items(),
                                                    key=lambda x: -x[1])]}

    def recommendations(self, user: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {}
        if user:
            body["user"] = user
        return self._post("/recommendations", body)

    # Friendly aliases for the two Feedback.event values the CLI exposes.
    _FEEDBACK_EVENTS = {"THUMBS_UP": "UPVOTE", "THUMBS_DOWN": "DOWNVOTE"}

    def feedback(self, tracking_token: str, rating: str,
                 comments: Optional[str] = None,
                 category: Optional[str] = None) -> Dict[str, Any]:
        """Send feedback for a tracking token.

        `rating` accepts THUMBS_UP / THUMBS_DOWN (mapped to the spec's
        UPVOTE / DOWNVOTE) or any Feedback.event value verbatim. `event` and
        `trackingTokens` are both required by the API; `category` (SEARCH,
        CHAT, ...) is a separate enum and stays optional.
        """
        event = self._FEEDBACK_EVENTS.get(rating.upper(), rating.upper())
        body: Dict[str, Any] = {
            "event": event,
            "trackingTokens": [tracking_token],
        }
        if category:
            body["category"] = category.upper()
        if comments:
            body["manualFeedbackInfo"] = {"comments": comments}
        return self._post("/feedback", body)

    # ---------------- agents and tools ----------------

    def agents_search(self, query: Optional[str] = None) -> Dict[str, Any]:
        # The API filters agents by name; `query` is the CLI's flag name.
        return self._post("/agents/search", {"name": query or ""})

    def agent_run(self, agent_id: str, input: str, stream: bool = False) -> Dict[str, Any]:
        path = "/agents/runs/stream" if stream else "/agents/runs/wait"
        return self._post(path, {"agent_id": agent_id, "input": input})

    def tools_list(self, tool_names: Optional[List[str]] = None) -> Dict[str, Any]:
        # GET, per the spec — the optional filter is a query parameter.
        return self._post("/tools/list", {}, method="GET",
                          query={"toolNames": tool_names} if tool_names else None)

    def tools_call(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
        # The wire field is `parameters`; `arguments` is kept as the Python
        # argument name because that is what the /tools.call command calls it.
        return self._post("/tools/call", {"name": name, "parameters": arguments})

    # ---------------- documents and people ----------------

    def get_documents(self, ids: Optional[List[str]] = None,
                      urls: Optional[List[str]] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {}
        if ids:
            body["documentSpecs"] = [{"id": i} for i in ids]
        elif urls:
            body["documentSpecs"] = [{"url": u} for u in urls]
        return self._post("/getdocuments", body)

    def document_permissions(self, doc_id: str) -> Dict[str, Any]:
        return self._post("/getdocpermissions", {"documentId": doc_id})

    def list_entities(self, kind: str = "PEOPLE", page_size: int = 10,
                      query: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"entityType": kind, "pageSize": page_size}
        if query:
            body["query"] = query
        return self._post("/listentities", body)

    def person(self, email: str) -> Dict[str, Any]:
        return self._post("/people", {"emailIds": [email]})

    # ---------------- announcements, collections, pins ----------------

    def announcement_create(self, title: str, body_text: str,
                            audience: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"title": title, "body": {"text": body_text}}
        if audience:
            body["audienceFilters"] = [{"filter": audience}]
        return self._post("/createannouncement", body)

    def announcement_delete(self, ann_id: str) -> Dict[str, Any]:
        return self._post("/deleteannouncement", {"id": ann_id})

    def collections_list(self) -> Dict[str, Any]:
        return self._post("/listcollections", {})

    def collection_create(self, name: str, description: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"name": name}
        if description:
            body["description"] = description
        return self._post("/createcollection", body)

    def pins_list(self) -> Dict[str, Any]:
        return self._post("/listpins", {})

    def pin_create(self, url: Optional[str] = None, query: str = "",
                   doc_id: Optional[str] = None) -> Dict[str, Any]:
        """Pin a document as a result for one or more queries.

        The API pins by document id, while the CLI has always taken a URL, so a
        URL is resolved through /getdocuments first. Pass `doc_id` to skip that
        round trip.
        """
        if not doc_id:
            if not url:
                raise GleanError("pin_create needs either a url or a doc_id.")
            resolved = self.get_documents(urls=[url])
            docs = resolved.get("documents") or resolved.get("results") or []
            if isinstance(docs, dict):          # some tenants key by spec
                docs = list(docs.values())
            doc_id = next((d.get("id") for d in docs if isinstance(d, dict) and d.get("id")), None)
            if not doc_id:
                raise GleanError(
                    f"Could not resolve {url} to a document id, so it cannot be pinned. "
                    "Pass --doc-id instead."
                )
        return self._post("/pin", {"documentId": doc_id, "queries": [query]})

    def pin_delete(self, pin_id: str) -> Dict[str, Any]:
        return self._post("/unpin", {"id": pin_id})

    # ---------------- collections delete ----------------

    def collection_delete(self, ids: List[int]) -> Dict[str, Any]:
        return self._post("/deletecollection", {"ids": ids})

    # ---------------- shortcuts ----------------

    def shortcuts_list(self, query: Optional[str] = None,
                       page_size: int = 20) -> Dict[str, Any]:
        body: Dict[str, Any] = {"pageSize": page_size}
        if query:
            body["query"] = query
        return self._post("/listshortcuts", body)

    def shortcut_get(self, alias: str) -> Dict[str, Any]:
        return self._post("/getshortcut", {"alias": alias})

    def shortcut_create(self, alias: str, url: str,
                        description: Optional[str] = None,
                        unlisted: bool = False) -> Dict[str, Any]:
        data: Dict[str, Any] = {"inputAlias": alias, "destinationUrl": url,
                                 "unlisted": unlisted}
        if description:
            data["description"] = description
        return self._post("/createshortcut", {"data": data})

    def shortcut_update(self, shortcut_id: int,
                        alias: Optional[str] = None,
                        url: Optional[str] = None,
                        description: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"id": shortcut_id}
        if alias:
            body["inputAlias"] = alias
        if url:
            body["destinationUrl"] = url
        if description is not None:
            body["description"] = description
        return self._post("/updateshortcut", body)

    def shortcut_delete(self, shortcut_id: int) -> Dict[str, Any]:
        return self._post("/deleteshortcut", {"id": shortcut_id})

    # ---------------- answers ----------------

    def answers_list(self) -> Dict[str, Any]:
        return self._post("/listanswers", {})

    def answer_get(self, answer_id: int) -> Dict[str, Any]:
        return self._post("/getanswer", {"id": answer_id})

    def answer_create(self, question: str, body_text: str,
                      audience: Optional[str] = None) -> Dict[str, Any]:
        data: Dict[str, Any] = {"question": question, "bodyText": body_text}
        if audience:
            data["audienceFilters"] = [{"filter": audience}]
        return self._post("/createanswer", {"data": data})

    def answer_update(self, answer_id: int,
                      question: Optional[str] = None,
                      body_text: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"id": answer_id}
        if question:
            body["question"] = question
        if body_text:
            body["bodyText"] = body_text
        return self._post("/editanswer", body)

    def answer_delete(self, answer_id: int) -> Dict[str, Any]:
        return self._post("/deleteanswer", {"id": answer_id})

    # ---------------- summarize ----------------

    def summarize(self, url: Optional[str] = None,
                  doc_id: Optional[str] = None,
                  query: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {}
        if doc_id:
            body["documentSpecs"] = [{"id": doc_id}]
        elif url:
            body["documentSpecs"] = [{"url": url}]
        if query:
            body["query"] = query
        return self._post("/summarize", body)

    # ---------------- verification ----------------

    def verification_list(self, count: Optional[int] = None) -> Dict[str, Any]:
        # `count` is a query parameter in the spec, not a body field.
        return self._post("/listverifications", {},
                          query={"count": count} if count else None)

    def verification_verify(self, doc_id: str,
                             action: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"documentId": doc_id}
        if action:
            body["action"] = action
        return self._post("/verify", body)

    def verification_remind(self, doc_id: str,
                             remind_in_days: Optional[int] = None,
                             assignee: Optional[str] = None,
                             reason: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"documentId": doc_id}
        if remind_in_days:
            body["remindInDays"] = remind_in_days
        if assignee:
            body["assignee"] = assignee
        if reason:
            body["reason"] = reason
        return self._post("/addverificationreminder", body)

    # ---------------- messages ----------------

    def messages_get(self, msg_id: str, id_type: str,
                     datasource: str,
                     direction: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"id": msg_id, "idType": id_type,
                                 "datasource": datasource}
        if direction:
            body["direction"] = direction
        return self._post("/messages", body)

    # ---------------- activity ----------------

    def activity_report(self, url: str,
                         action: str = "VIEW") -> Dict[str, Any]:
        body = {"events": [{"url": url, "action": action,
                             "timestamp": int(time.time())}]}
        return self._post("/activity", body)

    # ---------------- insights ----------------

    def insights(self, overview: bool = True, assistant: bool = False,
                 agents: bool = False, disable_per_user: bool = False,
                 locale: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {}
        if overview:
            body["overviewRequest"] = {}
        if assistant:
            body["assistantRequest"] = {}
        if agents:
            body["agentsRequest"] = {}
        if disable_per_user:
            body["disablePerUserInsights"] = True
        if locale:
            body["locale"] = locale
        return self._post("/insights", body)


# -------------------- mocks --------------------

def _mock_datasource_filter(body: Dict[str, Any]) -> Optional[str]:
    """Read a datasource restriction off a search body, in either shape."""
    opts = body.get("requestOptions") or {}
    ds = opts.get("datasourceFilter")
    if isinstance(ds, str) and ds:
        return ds
    for f in (opts.get("facetFilters") or []):
        if f.get("fieldName") != "datasource":
            continue
        for v in (f.get("values") or []):
            value = v.get("value") if isinstance(v, dict) else v
            if value:
                return str(value)
    return None


# Client API paths the personal index can answer. Everything else genuinely has
# no local counterpart — there are no agents, announcements or collections in a
# folder of files — and saying so beats returning a plausible-looking stub that
# an agent would treat as real.
LOCAL_PATHS = ("/search", "/chat", "/autocomplete", "/getdocuments")


def _local_response(path: str, body: Dict[str, Any]) -> Dict[str, Any]:
    """Serve a Client API call from ~/.gleancode/personal.db.

    Responses use the Client API's own shapes, so every renderer in the REPL
    and every consumer of the MCP server works unchanged — the same reason
    mock mode can reuse the whole command surface.
    """
    if path == "/search":
        wants_facets = "datasource" in (
            (body.get("requestOptions") or {}).get("facets") or []
        )
        return personal.search_response(
            body.get("query", ""),
            page_size=body.get("pageSize", 10),
            datasource=_mock_datasource_filter(body),
            facets=wants_facets,
        )
    if path == "/chat":
        try:
            message = body["messages"][-1]["fragments"][0]["text"]
        except (KeyError, IndexError, TypeError):
            message = ""
        return personal.chat_response(message, chat_id=body.get("chatId"))
    if path == "/autocomplete":
        return personal.autocomplete_response(body.get("query", ""))
    if path == "/getdocuments":
        return personal.documents_response(body)
    raise GleanError(
        f"{path} has no local equivalent. Local mode serves "
        f"{', '.join(LOCAL_PATHS)} from the personal index; use /mode live for "
        f"the rest of the Glean API, or /mode mock to explore it offline."
    )


def _mock_response(path: str, body: Dict[str, Any]) -> Dict[str, Any]:
    time.sleep(0.25)  # feel of a network call
    if path == "/chat":
        q = body["messages"][-1]["fragments"][0]["text"]
        cites = mock_corpus.citations(q, 3)
        titles = ", ".join(c["sourceDocument"]["title"] for c in cites)
        return {
            "chatId": body.get("chatId") or f"chat_{uuid.uuid4().hex[:8]}",
            "messages": [{
                "author": "GLEAN_AI",
                "messageType": "CONTENT",
                "fragments": [{"text":
                    f"[mock] You asked: {q}\n\n"
                    f"Drawing on {len(cites)} documents from the {mock_corpus.COMPANY} index "
                    f"({titles}), here is a simulated answer. Configure a real token with "
                    f"/login — or sign in with /auth login — to hit live Glean."}],
                "citations": cites,
            }],
        }
    if path == "/search":
        q = body.get("query", "")
        n = body.get("pageSize", 10)
        datasource = _mock_datasource_filter(body)
        wants_facets = "datasource" in (
            (body.get("requestOptions") or {}).get("facets") or []
        )
        resp: Dict[str, Any] = {
            "results": mock_corpus.search(q, page_size=n, datasource=datasource)
        }
        if wants_facets:
            resp["facetResults"] = [{
                "sourceName": "datasource",
                "buckets": mock_corpus.facet_buckets(datasource),
            }]
        return resp
    if path == "/autocomplete":
        q = body.get("query", "")
        return {"results": [{"suggestion": s} for s in mock_corpus.suggestions(q, 3)]}
    if path == "/recommendations":
        return {"results": mock_corpus.search("*", page_size=5)}
    if path == "/feedback":
        return {"status": "ok"}
    if path == "/agents/search":
        agents = [
            {"id": "agt_research", "name": "Research Agent",
             "description": "Deep research across company knowledge."},
            {"id": "agt_sales", "name": "Sales Assistant",
             "description": "Summarises accounts and prepares call notes."},
        ]
        # The API filters by agent name; mirror that so offline behaviour and
        # live behaviour agree.
        name = (body.get("name") or "").strip().lower()
        if name:
            agents = [a for a in agents if name in a["name"].lower()]
        return {"agents": agents}
    if path in ("/agents/runs/wait", "/agents/runs/stream"):
        return {"runId": f"run_{uuid.uuid4().hex[:6]}",
                "output": f"[mock agent {body.get('agent_id')}] Finished task: "
                          f"{body.get('input','')[:60]}..."}
    if path == "/tools/list":
        return {"tools": [
            {"name": "search", "description": "Search the company index."},
            {"name": "create_doc", "description": "Create a new document."},
        ]}
    if path == "/tools/call":
        return {"name": body.get("name"), "result": "ok",
                "output": f"[mock tool {body.get('name')}] args={body.get('parameters')}"}
    if path == "/getdocuments":
        specs = body.get("documentSpecs", [])
        docs = []
        for s in specs:
            hit = mock_corpus.find(s)
            if hit:
                doc = mock_corpus.as_document(hit)
                # Real /getdocuments returns document text; the mock used to
                # return metadata only, which left anything downstream of a
                # citation (the flow mapper especially) with nothing to read.
                doc["body"] = {"mimeType": "text/plain",
                               "textContent": mock_corpus.expand(hit["body"])}
                docs.append(doc)
                continue
            # Unknown id/url: echo it back so callers still get what they asked for.
            docs.append({"id": s.get("id") or s.get("url"),
                          "title": "Untitled document (not in the mock corpus)",
                          "url": s.get("url", "https://example.com")})
        return {"documents": docs}
    if path == "/getdocpermissions":
        doc = mock_corpus.find(body)
        owner = (doc or {}).get("author")
        roster = mock_corpus.people(4)
        perms = []
        if owner:
            perms.append({"email": owner, "role": "owner"})
        perms += [{"email": p["email"], "role": "viewer"}
                  for p in roster if p["email"] != owner]
        return {"permissions": perms}
    if path == "/listentities":
        return {"results": mock_corpus.people(int(body.get("pageSize") or 10))}
    if path == "/people":
        emails = body.get("emailIds") or []
        return mock_corpus.person(emails[0] if emails else None)
    if path == "/createannouncement":
        return {"id": f"ann_{uuid.uuid4().hex[:6]}", "status": "created"}
    if path == "/deleteannouncement":
        return {"id": body.get("id"), "status": "deleted"}
    if path == "/listcollections":
        return {"collections": [
            {"id": "col_1", "name": "Quarterly Planning", "itemCount": 6},
            {"id": "col_2", "name": "On-Call and Incidents", "itemCount": 5},
            {"id": "col_3", "name": "New Engineer Onboarding", "itemCount": 4},
        ]}
    if path == "/createcollection":
        return {"id": f"col_{uuid.uuid4().hex[:6]}", "name": body.get("name")}
    if path == "/listpins":
        return {"pins": [{"id": "pin_1", "query": "pto", "url": "https://example.com/pto"}]}
    if path == "/pin":
        return {"id": f"pin_{uuid.uuid4().hex[:6]}", "status": "created",
                "documentId": body.get("documentId"),
                "queries": body.get("queries") or []}
    if path == "/unpin":
        return {"id": body.get("id"), "status": "unpinned"}
    if path == "/deletecollection":
        return {"ids": body.get("ids"), "status": "deleted"}
    if path == "/listshortcuts":
        q = body.get("query", "")
        return {"shortcuts": [
            {"id": 1, "inputAlias": "pto", "destinationUrl": "https://hr.acme.com/pto",
             "description": "PTO policy"},
            {"id": 2, "inputAlias": "oncall", "destinationUrl": "https://wiki.acme.com/oncall",
             "description": "On-call runbook"},
        ] if not q else [
            {"id": 1, "inputAlias": q, "destinationUrl": f"https://acme.com/{q}",
             "description": f"Shortcut for {q}"}
        ]}
    if path == "/getshortcut":
        alias = body.get("alias", "")
        return {"shortcut": {"id": 1, "inputAlias": alias,
                              "destinationUrl": f"https://acme.com/{alias}",
                              "description": f"Shortcut for {alias}"}}
    if path == "/createshortcut":
        data = body.get("data", {})
        return {"id": int(uuid.uuid4().int % 10000),
                "inputAlias": data.get("inputAlias"),
                "destinationUrl": data.get("destinationUrl"),
                "status": "created"}
    if path == "/updateshortcut":
        return {"id": body.get("id"), "status": "updated"}
    if path == "/deleteshortcut":
        return {"id": body.get("id"), "status": "deleted"}
    if path == "/listanswers":
        return {"answers": [
            {"id": 1, "question": "What is our PTO policy?",
             "bodyText": "Employees get 20 days of PTO per year."},
            {"id": 2, "question": "Where is the oncall runbook?",
             "bodyText": "See go/oncall for the full runbook."},
        ]}
    if path == "/getanswer":
        return {"answer": {"id": body.get("id"),
                           "question": "Mock question?",
                           "bodyText": "Mock answer body."}}
    if path == "/createanswer":
        data = body.get("data", {})
        return {"id": int(uuid.uuid4().int % 10000),
                "question": data.get("question"),
                "status": "created"}
    if path == "/editanswer":
        return {"id": body.get("id"), "status": "updated"}
    if path == "/deleteanswer":
        return {"id": body.get("id"), "status": "deleted"}
    if path == "/summarize":
        specs = body.get("documentSpecs") or []
        spec = specs[0] if specs else {}
        src = spec.get("url") or spec.get("id") or "the document"
        doc = mock_corpus.find(spec)
        if doc:
            return {"summary": f"[mock] {mock_corpus.summarize(doc)}"}
        return {"summary": f"[mock] This is a summary of {src}. "
                           "Configure a real token with /login to get a live summary."}
    if path == "/listverifications":
        # Alternate verified / unverified over the freshest corpus documents.
        out = []
        for i, r in enumerate(mock_corpus.search("*", page_size=6)):
            verified = i % 2 == 1
            out.append({
                "documentId": r["id"],
                "title": r["title"],
                "url": r["url"],
                "datasource": r["datasource"],
                "lastVerifiedTs": (int(time.time()) - 86400 * 30) if verified else None,
                "status": "VERIFIED" if verified else "UNVERIFIED",
            })
        return {"verifications": out}
    if path == "/verify":
        return {"documentId": body.get("documentId"), "status": "VERIFIED"}
    if path == "/addverificationreminder":
        return {"documentId": body.get("documentId"), "status": "reminder_set"}
    if path == "/messages":
        thread = mock_corpus.rank("thread", datasource="slack")
        return {"messages": [
            {"id": body.get("id") or d["id"],
             "text": mock_corpus.expand(d["body"]),
             "channel": d.get("container", ""),
             "author": d.get("author", ""),
             "timestampMillis": (int(time.time()) - 86400 * int(d.get("updated_days_ago", 0))) * 1000}
            for d in thread[:3]
        ]}
    if path == "/activity":
        return {"status": "ok", "processed": len(body.get("events", []))}
    if path == "/insights":
        resp: Dict[str, Any] = {}
        if "overviewRequest" in body:
            resp["overviewResponse"] = {
                "monthlyActiveUsers": 312,
                "weeklyActiveUsers": 148,
                "employeeCount": 520,
                "totalSignups": 401,
                "searchSessionSatisfaction": 0.87,
                "lastUpdatedTs": int(time.time()) - 3600,
                "searchDatasourceCounts": {
                    "gdrive": 1840, "confluence": 920,
                    "slack": 610, "jira": 430,
                },
            }
        if "assistantRequest" in body:
            resp["assistantResponse"] = {
                "monthlyActiveUsers": 198,
                "weeklyActiveUsers": 94,
                "lastUpdatedTs": int(time.time()) - 3600,
            }
        if "agentsRequest" in body:
            resp["agentsResponse"] = {
                "monthlyActiveUsers": 67,
                "weeklyActiveUsers": 31,
                "lastUpdatedTs": int(time.time()) - 3600,
            }
        return resp
    return {"mock": True, "path": path, "body": body}


def _mock_indexing_response(path: str, body: Dict[str, Any],
                            method: str = "POST") -> Dict[str, Any]:
    time.sleep(0.15)
    now = int(time.time())

    # ---- read-only / debug ----
    if path.endswith("/status") and path.startswith("/debug/"):
        return {
            "datasourceStatusOverview": "OK",
            "documents": {"uploadedCount": 12450, "indexedCount": 12318,
                          "lastIndexedTs": now, "lastUploadedTs": now - 600},
            "users": {"uploadedCount": 980, "indexedCount": 980, "lastUploadedTs": now - 3600},
            "groups": {"uploadedCount": 64, "indexedCount": 64},
            "memberships": {"uploadedCount": 4210, "indexedCount": 4210},
            "recentEvents": [
                {"ts": now - 300,  "type": "INDEX_BATCH", "status": "SUCCESS", "count": 218},
                {"ts": now - 1200, "type": "INDEX_BATCH", "status": "SUCCESS", "count": 184},
            ],
        }
    if path.startswith("/debug/") and path.endswith("/document"):
        return {
            "status": {"uploadStatus": "UPLOADED",
                       "lastUploadedAt": now - 600,
                       "lastIndexedAt": now - 540,
                       "errors": []},
            "uploadedPermissions": {"allowedUsers": ["alice@example.com"],
                                     "allowedGroups": ["engineering"]},
        }
    if path.startswith("/debug/") and path.endswith("/documents"):
        items = body.get("debugDocuments") or []
        return {"debugResults": [{
            "docId": d.get("docId", ""),
            "objectType": d.get("objectType", ""),
            "debugInfo": {"status": {"uploadStatus": "UPLOADED",
                                      "lastUploadedAt": now - 600}},
        } for d in items]}
    if path.startswith("/debug/") and path.endswith("/user"):
        return {
            "status": {"uploadStatus": "UPLOADED", "lastUploadedAt": now - 1800},
            "uploadedGroups": [{"name": "engineering"}, {"name": "all-hands"}],
        }

    if path == "/getdatasourceconfig":
        return {"datasource": body.get("datasource"),
                "displayName": (body.get("datasource") or "").title(),
                "objectDefinitions": [
                    {"name": "Article",   "displayLabel": "Article"},
                    {"name": "Comment",   "displayLabel": "Comment"},
                ],
                "isUserReferencedByEmail": True,
                "trustedDomains": ["example.com"],
                "iconUrl": "https://example.com/icon.png",
                "datasourceCategory": "PUBLISHED_CONTENT"}
    if path == "/getdocumentstatus":
        return {"uploadStatus": "UPLOADED",
                "lastUploadedAt": now - 600,
                "lastIndexedAt": now - 540,
                "indexingErrors": []}
    if path == "/getdocumentcount":
        return {"documentCount": 12450}
    if path == "/getusercount":
        return {"userCount": 980}
    if path == "/checkdocumentaccess":
        return {"hasAccess": True}

    # ---- custom metadata API ----
    if path.startswith("/custom-metadata/schema/"):
        group = path.rsplit("/", 1)[-1]
        if method == "GET":
            return {"group": group,
                    "metadataKeys": [
                        {"name": "department", "propertyType": "PICKLIST"},
                        {"name": "region",     "propertyType": "TEXT"},
                    ]}
        keys = body.get("metadataKeys") or []
        return {"status": "ACCEPTED", "group": group,
                "metadataKeysAccepted": len(keys)}
    if path.startswith("/document/") and "/custom-metadata/" in path:
        items = body.get("customMetadata") or []
        return {"status": "ACCEPTED",
                "metadataAccepted": len(items)}

    # ---- write / bulk / process — ack-style ----
    if path.startswith("/bulkindex") or path == "/uploadshortcuts":
        return {"uploadId": body.get("uploadId") or f"upload_{now}",
                "ack": True}
    if path == "/indexdocuments":
        return {"uploadId": body.get("uploadId") or f"upload_{now}",
                "documentsAccepted": len(body.get("documents") or [])}
    if path == "/indexemployeelist":
        return {"employeesAccepted": len(body.get("employees") or [])}
    if path.startswith("/processall"):
        return {"status": "STARTED", "datasource": body.get("datasource")}
    if path.startswith("/index") or path.startswith("/delete") or path == "/updatepermissions":
        return {"status": "ACCEPTED"}

    return {"mock": True, "path": path, "body": body}