"""The per-vault service object every tool operates on.

Owns the resolved vault root, the (strict-loaded) config, the seat table, the
one-writer git discipline, the JobManager, and the ModelProvider seam for the
agentic tools. The path-jail (vault_core.fs), the non-content container guard,
and the raw-config guard are applied *here*, before any result reaches a model
or a seat (docs/V3-PLAN.md §3) — tool handlers call the ``visible_rel``/
``ensure_visible`` helpers instead of re-implementing policy. (There was a
per-area ACL here through 2.67; 2.68.0 removed it — brain access is the
boundary, so a seat that reaches this vault reads all of it.)
"""
import itertools
import os
import sys
import threading
import time
from pathlib import Path, PurePosixPath

from alexandria_vault import vault_core
from alexandria_vault.vault_core import JailError, fs
from alexandria_vault.vault_core import graph as _graph

from .gitops import VaultWriter
from .jobs import JobManager
from .seats import load_seats
from .usage import UsageMeter

# Config is served redacted through config_get, never raw: the file carries the
# seat table and secret names, which must not leak to any seat's
# context. (Secrets themselves are names-only in config by the config.py rule.)
_CONFIG_PATHS = {"Agent/config.json", "Alexandria/config.json"}
# Lowercased for case-insensitive matching: macOS (APFS) is case-insensitive by
# default, so `agent/config.json` reads the same raw config as `Agent/config.json`
# — a case-sensitive check would answer "visible" for it and leak the roster.
_CONFIG_PATHS_CI = {p.lower() for p in _CONFIG_PATHS}
_REDACTED_CONFIG_KEYS = ("secrets", "team", "peers")


class GraphIndex:
    """The warm in-memory link graph (docs/V3-P4-RESEARCH.md §Performance).

    The KeepAlive server is long-lived, so the ``vault_core.graph`` scan runs
    once and is reused by ``links_get``/``graph_get`` and the research spider;
    the writer's ``on_change`` hook invalidates it on every write and pull, so
    a rebuild happens lazily on the next use. This is NOT the persisted index
    (that stays V4/Electron-FTS). Callers must treat the returned dict as
    read-only — it is shared across threads.

    ``on_change`` alone was not enough: it only fires for writes that go through
    :class:`VaultWriter`. A vault edited out of band — Obsidian on the same box,
    a shell ``git pull``, a sidecar's own checkout — left this cache serving the
    pre-edit graph indefinitely, so ``links_get``/``graph_get``/``think_deep``
    contradicted the uncached ``vaultq_*`` verbs about the same file in the same
    session. A cheap stat-only stamp (file count + newest mtime) is re-checked at
    most every :data:`STAMP_INTERVAL_S`, so an out-of-band edit self-heals
    without making every read pay for a walk.
    """

    STAMP_INTERVAL_S = 2.0

    def __init__(self, vault):
        self.vault = vault
        self._lock = threading.Lock()
        self._cached = None
        self._stamp = None
        self._checked_at = 0.0

    def invalidate(self):
        with self._lock:
            self._cached = None
            self._stamp = None

    def _corpus_stamp(self):
        """(file count, newest mtime_ns) over the note corpus — stat only, no reads."""
        n, newest = 0, 0
        jail_root = fs.vault_root(self.vault)
        for dirpath, dirnames, filenames in os.walk(self.vault):
            dirnames[:] = [d for d in dirnames if d not in fs.EXCLUDE_DIRS]
            for name in filenames:
                if not name.endswith(".md"):
                    continue
                # Count the same corpus the graph builders do: an out-of-vault
                # symlink they drop must not move this stamp, or the cache
                # invalidates on a file that is not in the graph it caches.
                if not fs.within_vault(self.vault, os.path.join(dirpath, name),
                                       root=jail_root):
                    continue
                try:
                    m = os.stat(os.path.join(dirpath, name)).st_mtime_ns
                except OSError:
                    continue
                n += 1
                if m > newest:
                    newest = m
        return n, newest

    def get(self) -> dict:
        with self._lock:
            if self._cached is not None:
                now = time.monotonic()
                if now - self._checked_at >= self.STAMP_INTERVAL_S:
                    self._checked_at = now
                    stamp = self._corpus_stamp()
                    if stamp != self._stamp:      # edited outside the writer
                        self._cached = None
            if self._cached is None:
                self._cached = _graph.build_graph(self.vault)
                self._stamp = self._corpus_stamp()
                self._checked_at = time.monotonic()
            return self._cached


class VaultMCPService:
    def __init__(self, vault, env=None, provider_factory=None):
        self.vault = vault_core.locate_vault(explicit=vault)
        self.config = vault_core.load_config(self.vault, strict=True)
        self.env = env               # the seat-token source ONLY (tests inject here);
        self.seats = load_seats(self.config, env=env)
        # the writer/jobs read the live process env (SWEEP_STATE, SYNC_*) — the
        # launchd wrapper sources alexandria.env into exactly that.
        self.writer = VaultWriter(self.vault, self.config)
        self.graph_index = GraphIndex(self.vault)
        self.writer.on_change = self.graph_index.invalidate
        self.jobs = JobManager()
        self.usage = UsageMeter()    # per-seat tool-call rollup (the metering spine)
        self.manifests = {}          # ingest_content → apply_ingest handoff
        self._manifest_seq = itertools.count(1)
        self._provider_factory = provider_factory  # test seam (mocked ModelProvider)

    # ---- auth --------------------------------------------------------------------

    def seat_for_token(self, token):
        """The Seat a Bearer identifies, or None (transport answers 401)."""
        return self.seats.get(token or "")

    # ---- config ------------------------------------------------------------------

    def reload_config(self):
        self.config = vault_core.load_config(self.vault, strict=True)
        self.writer.config = self.config

    def redacted_config(self) -> dict:
        """The config as any seat may see it: whole-brain (all areas shown — a seat
        that reads the brain reads its config too), but seat/secret/roster blocks
        withheld so CREDENTIALS never reach a seat."""
        cfg = dict(self.config)
        vm = dict(cfg.get("vault_mcp") or {})
        vm.pop("seats", None)
        vm.pop("private_areas", None)
        cfg["vault_mcp"] = vm
        for key in _REDACTED_CONFIG_KEYS:
            cfg.pop(key, None)
        return cfg

    # ---- read boundary (jail in vault_core.fs; whole-brain reads + config redaction) ----

    def visible_rel(self, rel) -> bool:
        """Vault-relative path the seats may see: inside the jail, outside every
        non-content container (``EXCLUDE_DIRS``), and not the raw config file.

        The check judges the CANONICAL target — ``..`` and symlinks collapsed via
        :func:`vault_core.fs.canonical_rel` — so a path that *resolves* into a
        container (``Agent/x/../config.json``, ``.git/config``) can't slip past a
        textual prefix check while the filesystem op lands it there anyway. Never
        raises; fails closed (unprovable ⇒ not visible) — it is a filter
        predicate."""
        raw = str(rel).strip()
        if raw.startswith("~") or PurePosixPath(raw.rstrip("/")).is_absolute():
            return False
        try:
            canon = fs.canonical_rel(self.vault, raw.rstrip("/") or ".")
        except (ValueError, OSError, RuntimeError):
            return False  # escapes the jail / underivable — never call that "visible"
        # Case-insensitive on both the config file and the container set: on a
        # case-insensitive filesystem the canonical path keeps its typed case but
        # still reads the real file, so `agent/config.json` / `.GIT/config` must
        # be caught too (fs.EXCLUDE_DIRS is all-lowercase, so lowercase the parts).
        if canon.lower() in _CONFIG_PATHS_CI:
            return False
        if any(seg.lower() in fs.EXCLUDE_DIRS for seg in PurePosixPath(canon).parts):
            return False
        # Whole-brain: a seat that can reach this brain reads the whole brain.
        # Brain access IS the boundary (one vault per principal, gated by the
        # seat/OAuth key) — there is no per-area wall inside a brain. What
        # remains above is NOT an ACL: the fs jail, the non-content container
        # guard, and the raw-config guard (that file carries other seats'
        # bearer tokens).
        return True

    def ensure_visible(self, rel):
        """Refuse a direct fetch of a non-visible path.

        A jail escape (a ``..`` or symlink that leaves the vault) surfaces as
        ``escapes`` — the hard boundary the jailed fs op would flag too. An
        inside-but-hidden path gets the *same* ``not found`` a missing file gets,
        so hidden and missing stay indistinguishable (no existence oracle)."""
        from .tools import ToolError
        raw = str(rel).strip()
        if not (raw.startswith("~") or PurePosixPath(raw.rstrip("/")).is_absolute()):
            try:
                fs.canonical_rel(self.vault, raw.rstrip("/") or ".")
            except JailError as e:
                raise ToolError(str(e))
            except (OSError, RuntimeError):
                pass  # the real fs op will surface the true error on the read
        if not self.visible_rel(rel):
            raise ToolError(f"not found: {rel}")

    # ---- the frontier primitive (docs/V3-PLAN.md §1.6, §3) --------------------------
    # ONE place a "visible-only view under this seat" is built, so every agentic
    # tool that gathers vault context for a model (the research spider, ingest)
    # filters at the FRONTIER, not just the seed — a container or config path
    # linked from a note is never handed to the model. This was a per-area ACL
    # until 2.62.0 made reads whole-brain; the frontier discipline outlived it,
    # because the jail and container guards still have to hold at every hop the
    # spider takes. The traversal-leak / injection-to-exfil test is a per-tool
    # release gate on each caller (docs/V3-PLAN.md §8b).

    def visible_filter(self, paths, scope_prefixes=()):
        """The subset of ``paths`` this seat may see, order preserved, optionally
        restricted to a pillar/area subtree (``scope_prefixes``)."""
        pref = tuple(scope_prefixes or ())
        return [p for p in paths
                if self.visible_rel(p) and (not pref or str(p).startswith(pref))]

    def visible_universe(self, scope_prefixes=()) -> set:
        """The set of visible, in-scope note (``.md``) paths — the spider's
        traversal universe, built visible-only so a hidden note is never
        enqueued, resolved, or read (the traversal-leak rule)."""
        return set(self.visible_filter(fs.glob(self.vault, "**/*.md"), scope_prefixes))

    # ---- models (the agentic seam) --------------------------------------------------

    def get_provider(self):
        """The ModelProvider for server-side judgment; injectable for tests."""
        if self._provider_factory is not None:
            return self._provider_factory()
        from alexandria_vault import models
        name = ((self.config.get("model") or {}).get("provider")) or "anthropic"
        return models.get_provider(name)

    def resolve_model(self, purpose=None) -> str:
        from alexandria_vault.models import resolve_model
        return resolve_model(self.config, purpose)

    def resolve_tier_model(self, tool, tier, override=None) -> str:
        """The model for a (tool, tier) pair — the gather/synthesize routing the
        agentic loops use (docs/V3-PLAN.md §4 P2). ``override`` lets a tool's own
        whole-run pin (e.g. ``vault_mcp.research.model``) win over the tiers."""
        from alexandria_vault.models import resolve_model_tiered
        return resolve_model_tiered(self.config, tool, tier, override=override)

    # ---- engine entrypoints (subprocess: a threaded server must never mutate the
    # engine's import-frozen module state in-process) ---------------------------------

    def engine_argv(self, module, *args) -> list:
        return [sys.executable, "-m", f"alexandria_vault.engine.{module}",
                "--vault", str(self.vault), *args]

    # ---- misc ------------------------------------------------------------------------

    def daily_rel(self, date=None) -> str:
        """Daily/<date>.md, defaulting to today in the owner's timezone (the
        digest's convention: the owner's calendar day, not the host's)."""
        if not date:
            import datetime as dt
            tzname = (self.config.get("owner") or {}).get("timezone_default")
            tz = None
            if tzname:
                try:
                    from zoneinfo import ZoneInfo
                    tz = ZoneInfo(tzname)
                except Exception:
                    tz = None
            date = dt.datetime.now(tz).date().isoformat()
        return f"Daily/{date}.md"

    def vault_rel(self, p) -> str:
        return str(Path(p).resolve().relative_to(self.vault))
