"""think_deep — the flagship agentic tool (docs/V3-P4-RESEARCH.md).

Given a question, spawn parallel research subagents that spider THIS vault's
wikilink graph — PEAK-partitioned fan-out (an Areas agent per matched area,
People, Events+Daily, Knowledge) over a shared frontier — and merge their
distilled findings into a ``[[wikilink]]``-cited synthesis with gaps,
contradictions, and staleness flags. Composes P1 ``vault_core.graph`` (through
the service's warm index), P2 ``run_tool_loop``, and P3 JobManager/seat scope.

Why subagents, not one loop: a real brain has hundreds of notes — one loop
reading 50+ blows its window. Each agent holds one region and returns findings
(claims + the source paths it read, leads, gaps), never raw note bodies, so
the orchestrator only ever holds findings. The pillars also explore
concurrently (wall-clock).

The two load-bearing properties, both structural rather than prompt-enforced:

- **Filtering at the frontier** (the traversal-leak rule): the frontier's note
  universe is built visible-only, so a path a link points at but the seat may
  not read — outside the jail, a non-content container, the raw config file —
  is never queued, never resolvable, never in a links/grep/list result, and a
  read attempt answers exactly like a missing note. Research runs under the
  calling seat's view of the vault; nothing an agent can request widens it.
  (This was a per-area ACL until 2.62.0/2.68.0 made reads whole-brain; the
  frontier discipline outlived the ACL, because the jail and the container and
  config guards still have to hold at every hop, not just at the seed.)
- **Hard budgets**: ``vault_mcp.research`` caps agents / depth /
  notes-per-agent (+ an optional per-run ``token_budget``); the tools
  themselves stop feeding an over-budget agent, ``max_turns`` backstops a
  runaway loop, and a metering wrapper counts every token OUTSIDE the loop so
  usage is reported even for an agent that dies at the turn limit. A capped
  run stops, reports partial findings, and says so.
"""
import json
import re
import threading
from collections import OrderedDict
from pathlib import PurePosixPath

from alexandria_vault.models import (
    ModelProviderError, ToolLoopCancelled, ToolLoopLimitError, run_tool_loop)
from alexandria_vault.vault_core import JailError, fs

# Spec defaults (docs/V3-P4-RESEARCH.md §Config); config vault_mcp.research
# overrides. These are HARD caps — a request may ask for less, never more.
DEFAULT_MAX_AGENTS = 4
DEFAULT_MAX_DEPTH = 2
DEFAULT_MAX_NOTES = 20

READ_CHAR_CAP = 8000     # per-note text served to an agent
VERIFY_CHAR_CAP = 4000   # per-source text served to the verify pass
GREP_HIT_CAP = 40
LIST_CAP = 60
SEEDS_PER_AGENT = 8
LEADS_CAP = 10

PILLAR_TOPS = ("Areas", "People", "Events", "Daily", "Knowledge")

# One message shape for unsurfaced / missing / out-of-jail — an existence oracle
# at the frontier would itself report what the spider was refused (P3's
# ensure_visible rule).
_NOT_FOUND = ("not found: {ref} (only notes surfaced by your seeds, "
              "research_list, research_grep, or research_links are readable)")

_STOPWORDS = frozenset(
    "a an and are as at be but by did do does for from had has have how i in is it its "
    "me my not of on or our so that the their them then there these they this to was we "
    "were what when where which who why will with you your".split())


# =====================================================================================
# plan — validate + clamp synchronously, before a job starts
# =====================================================================================

def scope_prefixes(svc, scope):
    """User scope names (bare areas or pillar-rooted paths) → visible folder
    prefixes — the one scope dialect every graph-traversal tool speaks
    (think_deep and process_thought). Raises ValueError; hidden
    and nonexistent entries answer identically (the no-existence-oracle rule
    holds at plan time, not just at the frontier)."""
    if not isinstance(scope, list):
        raise ValueError("scope must be a list of area/pillar names")
    prefixes = []
    for s in scope:
        s = str(s).strip().strip("/")
        if not s:
            continue
        prefix = (s + "/") if s.split("/", 1)[0] in PILLAR_TOPS else f"Areas/{s}/"
        if (not svc.visible_rel(prefix.rstrip("/"))
                or not (svc.vault / prefix.rstrip("/")).is_dir()):
            raise ValueError(f"scope entry {s!r} matches nothing researchable in this vault")
        prefixes.append(prefix)
    return prefixes


def plan(svc, args):
    """The validated, cap-clamped run plan. Raises ValueError on bad input (the
    tool maps it to ToolError) so mistakes surface at call time, not as a
    failed job. Scope errors answer identically for hidden and nonexistent
    areas — the plan stage must not become an existence oracle either."""
    rc = ((svc.config.get("vault_mcp") or {}).get("research")) or {}
    question = str(args.get("question") or "").strip()
    if not question:
        raise ValueError("question must be a non-empty string")
    mode = args.get("mode") or "quick"
    if mode not in ("quick", "deep"):
        raise ValueError("mode must be 'quick' or 'deep'")
    cap_agents = max(1, int(rc.get("max_agents", DEFAULT_MAX_AGENTS)))
    requested = args.get("max_agents")
    try:
        n_agents = min(cap_agents, max(1, int(requested))) if requested is not None \
            else cap_agents
    except (TypeError, ValueError):
        raise ValueError("max_agents must be an integer")
    prefixes = scope_prefixes(svc, args.get("scope") or [])
    token_budget = rc.get("token_budget")
    # Tier routing (V3-PLAN §4 P2): the spider subagents GATHER (cheap), the
    # verify + synthesis passes SYNTHESIZE (frontier). ``vault_mcp.research.model``
    # still pins the whole run when set — the "run all of research on one model"
    # knob wins over the tiers; unset tiers collapse to model.default (no-op).
    override = rc.get("model")
    return {
        "question": question,
        "mode": mode,
        "scope": prefixes,
        "n_agents": n_agents,
        "max_depth": max(0, int(rc.get("max_depth", DEFAULT_MAX_DEPTH))),
        "max_notes": max(1, int(rc.get("max_notes_per_agent", DEFAULT_MAX_NOTES))),
        "token_budget": int(token_budget) if token_budget else None,
        "model_gather": svc.resolve_tier_model("research", "gather", override=override),
        "model_synth": svc.resolve_tier_model("research", "synthesize", override=override),
        "save": bool(args.get("save")),
    }


# =====================================================================================
# the token meter — usage survives anything the loops do
# =====================================================================================

class _Meter:
    """Run-wide token ledger. Wraps every provider the run touches so usage is
    counted at the chat boundary — the flagship Library-token consumer must
    never under-report, even for a loop that dies at max_turns."""

    def __init__(self, budget=None):
        self.budget = budget
        self._lock = threading.Lock()
        self._stages = OrderedDict()

    def wrap(self, provider, stage):
        return _MeteredProvider(provider, self, stage)

    def add(self, stage, usage):
        usage = usage or {}
        with self._lock:
            slot = self._stages.setdefault(
                stage, {"input_tokens": 0, "output_tokens": 0, "calls": 0})
            slot["input_tokens"] += int(usage.get("input_tokens") or 0)
            slot["output_tokens"] += int(usage.get("output_tokens") or 0)
            for k in ("cache_creation_input_tokens", "cache_read_input_tokens"):
                v = int(usage.get(k) or 0)
                if v:   # reported only when prompt caching actually engaged
                    slot[k] = slot.get(k, 0) + v
            slot["calls"] += 1

    def total(self):
        with self._lock:
            t = {"input_tokens": 0, "output_tokens": 0}
            for u in self._stages.values():
                for k, v in u.items():
                    if k != "calls":
                        t[k] = t.get(k, 0) + v
            return t

    def by_stage(self):
        with self._lock:
            return {k: dict(v) for k, v in self._stages.items()}

    def exhausted(self):
        if not self.budget:
            return False
        t = self.total()
        return (t["input_tokens"] + t["output_tokens"]) >= self.budget


class _MeteredProvider:
    def __init__(self, inner, meter, stage):
        self._inner, self._meter, self._stage = inner, meter, stage

    def chat(self, *args, **kwargs):
        reply = self._inner.chat(*args, **kwargs)
        self._meter.add(self._stage, reply.usage)
        return reply

    def caps(self):
        """Transparent to capability gating (the loop's prompt-cache sniff)."""
        return getattr(self._inner, "caps", dict)()


# =====================================================================================
# the frontier — shared visited set, budgets, and the visibility boundary
# =====================================================================================

class Frontier:
    """The shared traversal state every subagent claims notes through.

    The universe is built once per run from the warm graph index, VISIBLE
    (and in-scope) notes only — the filter applies here, at the frontier, not
    just at the seed: a path a note links to but the seat may not read is not in
    the universe, so it never appears in a links/grep/list result, never
    resolves, and a read attempt raises the same "not found" a missing note gets.

    Depth is min-hops from an entry point (seeds, grep hits, and listings are
    depth 0; link discoveries are parent+1) and is enforced at DISCOVERY —
    neighbors past ``max_depth`` are simply never recorded, so they stay
    unreadable. The visited set partitions reads across agents: a note is read
    once, by exactly one agent.
    """

    def __init__(self, svc, scope_prefixes, max_depth, max_notes):
        g = svc.graph_index.get()
        self.vault = svc.vault
        self.max_depth = max_depth
        self.max_notes = max_notes
        # The visible-only universe comes from the shared frontier primitive
        # (service.visible_universe) — the ONE gate every context-gathering
        # agentic tool routes through, so the traversal-leak rule is uniform.
        self.notes = svc.visible_universe(scope_prefixes)
        self.out = {rel: [t for t in g["outlinks"].get(rel, ()) if t in self.notes]
                    for rel in self.notes}
        self.back = {rel: [s for s in g["backlinks"].get(rel, ()) if s in self.notes]
                     for rel in self.notes}
        self.stems = {}
        for rel in sorted(self.notes):
            self.stems.setdefault(PurePosixPath(rel).stem.lower(), []).append(rel)

        self._lock = threading.Lock()
        self.depth = {}      # rel -> min hops from an entry point
        self.read_by = {}    # rel -> agent label (the visited set / the partition)
        self.reads = {}      # agent label -> [rel, ...] in claim order
        self.stops = {}      # agent label -> why its tools stopped feeding it

    # ---- resolution + discovery ---------------------------------------------------

    def resolve(self, ref):
        """A model-supplied reference → a universe rel, or None. Hidden,
        out-of-scope, and nonexistent all resolve to None alike."""
        ref = str(ref or "").strip().strip("/")
        if not ref:
            return None
        if "/" in ref or ref.endswith(".md"):
            rel = ref if ref.endswith(".md") else ref + ".md"
            return rel if rel in self.notes else None
        hits = self.stems.get(ref.lower()) or []
        return hits[0] if hits else None   # ambiguity: deterministic first (sorted)

    def discover(self, rels, depth):
        """Record notes as reachable at ``depth`` (keeping the minimum). The
        depth cap is enforced by the callers never offering depth > max_depth."""
        with self._lock:
            for rel in rels:
                if rel in self.notes:
                    d = self.depth.get(rel)
                    if d is None or depth < d:
                        self.depth[rel] = depth

    def _known(self, ref):
        """Resolve to a DISCOVERED rel or raise the uniform not-found."""
        rel = self.resolve(ref)
        with self._lock:
            if rel is None or rel not in self.depth:
                raise LookupError(_NOT_FOUND.format(ref=ref))
            return rel, self.depth[rel]

    def mark_stop(self, agent, reason):
        with self._lock:
            self.stops.setdefault(agent, reason)

    # ---- the agent tools ------------------------------------------------------------

    def read(self, agent, ref):
        rel, d = self._known(ref)
        with self._lock:
            mine = self.reads.setdefault(agent, [])
            owner = self.read_by.get(rel)
            if owner == agent:
                return {"note": rel, "status": "you already read this note — use what you have"}
            if owner is not None:
                return {"note": rel, "status": "claimed by another research agent — "
                                               "its findings will cover this; move on"}
            if len(mine) >= self.max_notes:
                self.stops.setdefault(agent, f"note budget ({self.max_notes})")
                return {"status": f"BUDGET REACHED: you have read {self.max_notes} notes "
                                  "— stop exploring and submit_findings now"}
            self.read_by[rel] = agent
            mine.append(rel)
        try:
            # gather reads state their budget (+1 keeps the truncated flag honest)
            text = fs.read_text(self.vault, rel, max_chars=READ_CHAR_CAP + 1)
        except (OSError, JailError):   # raced away / broken symlink — same answer,
            with self._lock:           # and the claim is released: an unread note
                self.read_by.pop(rel, None)   # must never count as visited/citable
                if rel in mine:
                    mine.remove(rel)
            raise LookupError(_NOT_FOUND.format(ref=ref))
        if d + 1 <= self.max_depth:    # reading a note surfaces its neighborhood
            self.discover(self.out.get(rel, []) + self.back.get(rel, []), d + 1)
        out = {"note": rel, "depth": d, "text": text[:READ_CHAR_CAP]}
        if len(text) > READ_CHAR_CAP:
            out["truncated"] = True
        return out

    def links(self, agent, ref):
        rel, d = self._known(ref)
        if d >= self.max_depth:
            return {"note": rel, "at_max_depth": True, "outlinks": [], "backlinks": [],
                    "hint": "max traversal depth here — do not go deeper"}
        out, back = self.out.get(rel, []), self.back.get(rel, [])
        self.discover(out + back, d + 1)
        return {"note": rel, "outlinks": out, "backlinks": back}

    def grep(self, agent, pattern):
        if not str(pattern or "").strip():
            raise LookupError("empty pattern")
        try:
            hits = fs.grep(self.vault, str(pattern), ignore_case=True)
        except re.error as e:
            raise LookupError(f"bad regex: {e}")
        rows = [(p, n, line) for p, n, line in hits if p in self.notes][:GREP_HIT_CAP]
        self.discover([p for p, _, _ in rows], 0)   # lexical entry points
        return {"pattern": str(pattern),
                "hits": [{"note": p, "line": n, "text": line.strip()[:200]}
                         for p, n, line in rows]}

    def list_notes(self, agent, folder):
        folder = str(folder or "").strip().strip("/")
        pattern = f"{folder}/**/*.md" if folder else "**/*.md"
        try:
            rels = [p for p in fs.glob(self.vault, pattern) if p in self.notes]
        except JailError:
            raise LookupError(_NOT_FOUND.format(ref=folder))
        rels = rels[:LIST_CAP]
        self.discover(rels, 0)   # structural entry points
        return {"folder": folder or ".", "notes": rels}

    # ---- snapshots -------------------------------------------------------------------

    def visited(self):
        with self._lock:
            return dict(self.read_by)

    def reads_of(self, agent):
        with self._lock:
            return list(self.reads.get(agent, ()))


# =====================================================================================
# seeding — lexical + graph-name + PEAK semantics (spec §"How the spider works" 1)
# =====================================================================================

def _terms(question):
    words = re.findall(r"[\w'-]+", str(question).lower())
    return [w for w in words if len(w) >= 3 and w not in _STOPWORDS]


def seed(svc, frontier, question):
    """Ranked entry-point rels + matched area names (best-first). Name matches
    outrank content hits; the PEAK step falls out of the ranking (a matched
    area's folder note and TODO score through their name/content)."""
    terms = _terms(question)
    scores = {}
    ql = str(question).strip().lower()
    for rel in frontier.notes:
        stem = PurePosixPath(rel).stem.lower()
        s = sum(3 for t in terms if t in stem)
        if ql and ql in stem:
            s += 5
        if s:
            scores[rel] = s
    if terms:
        pattern = "|".join(re.escape(t) for t in terms)
        per_note = {}
        for p, _n, _line in fs.grep(svc.vault, pattern, ignore_case=True):
            if p in frontier.notes:
                per_note[p] = min(5, per_note.get(p, 0) + 1)   # cap content weight
        for p, c in per_note.items():
            scores[p] = scores.get(p, 0) + c
    ranked = sorted(scores, key=lambda r: (-scores[r], r))
    matched_areas = []
    for rel in ranked:
        parts = rel.split("/")
        if parts[0] == "Areas" and len(parts) > 2 and parts[1] not in matched_areas:
            matched_areas.append(parts[1])
    return ranked, matched_areas


def _agent_specs(frontier, ranked_seeds, matched_areas, n_agents):
    """The PEAK-partitioned fan-out: an Areas agent per matched area, then the
    People / Events+Daily / Knowledge pillars, capped at ``n_agents``. Every
    agent also gets the global top seeds as fill — the shared visited set
    makes overlap free (a note is read once, by whoever claims it first)."""
    def populated(prefixes):
        return any(rel.startswith(prefixes) for rel in frontier.notes)

    specs = []
    for area in matched_areas:
        prefix = f"Areas/{area}/"
        if populated((prefix,)):
            specs.append({
                "label": f"areas:{area}", "prefixes": (prefix,),
                "region": f"the '{area}' project area ({prefix}) — its folder note, "
                          "threads, and TODO are your home turf."})
    for label, prefixes, region in (
        ("people", ("People/",),
         "People/ — the humans involved; a person note's backlinks are hubs into "
         "everything they touch."),
        ("events-daily", ("Events/", "Daily/"),
         "Events/ and Daily/ — the dated record; weight recent entries over old ones."),
        ("knowledge", ("Knowledge/",),
         "Knowledge/ — subject articles and reference notes."),
    ):
        if populated(prefixes):
            specs.append({"label": label, "prefixes": prefixes, "region": region})
    specs = specs[:n_agents]
    if not specs:
        specs = [{"label": "generalist", "prefixes": ("",),
                  "region": "the whole vault (no PEAK pillars found)."}]
    for spec in specs:
        pillar = [r for r in ranked_seeds if r.startswith(spec["prefixes"])]
        fill = [r for r in ranked_seeds if r not in pillar]
        spec["seeds"] = (pillar + fill)[:SEEDS_PER_AGENT]
    return specs


# =====================================================================================
# the subagent loop
# =====================================================================================

_AGENT_SYSTEM = """\
You are the {label} research agent, one of several spidering a PEAK-organized Obsidian \
vault (the owner's private "brain") to answer a research question. Your region: {region}

Method: read your seed notes first; spider the wikilink graph from what you read — \
research_links gives a note's outlinks AND backlinks; read what looks relevant and prune \
what doesn't (judge from names and link context). research_grep / research_list find \
lexical and structural entry points. Budgets are enforced by the tools: when a result \
says a budget is reached, STOP exploring immediately and produce your findings.

When done (or stopped), call submit_findings EXACTLY ONCE with your findings — \
claims (each: text, the source paths you actually read, date), leads (cross-pillar \
threads another agent should chase), gaps (what the brain lacks on this question in \
your region) — then reply with a one-line wrap-up.

Rules: every claim carries at least one source path you actually read — never invent \
sources. Prefer newer dated notes. "date" is the note's own date (from its name or \
frontmatter) when apparent, else null."""

_S = {"type": "string"}


def _agent_tools():
    def schema(props, required):
        return {"type": "object", "properties": props, "required": list(required),
                "additionalProperties": False}
    return [
        {"name": "research_read",
         "description": "Read one note by name or vault-relative path (budgeted; "
                        "the frontier dedupes reads across agents).",
         "input_schema": schema({"note": _S}, ("note",))},
        {"name": "research_links",
         "description": "A note's outlinks and backlinks — the spider's move.",
         "input_schema": schema({"note": _S}, ("note",))},
        {"name": "research_grep",
         "description": "Case-insensitive regex over the vault -> note:line hits.",
         "input_schema": schema({"pattern": _S}, ("pattern",))},
        {"name": "research_list",
         "description": "List notes under a folder (e.g. 'People').",
         "input_schema": schema({"folder": _S}, ())},
        # schema-validated findings kill the JSON-scrape failure mode ("reply
        # was not JSON — findings lost"): the API validates the shape, and the
        # executor captures it whole
        {"name": "submit_findings",
         "description": "Submit your final findings EXACTLY ONCE when done "
                        "(or when told to stop).",
         "input_schema": schema(
             {"claims": {"type": "array", "items": {
                 "type": "object", "additionalProperties": False,
                 "properties": {"text": _S,
                                "sources": {"type": "array", "items": _S},
                                "date": {"type": ["string", "null"]}},
                 "required": ["text", "sources"]}},
              "leads": {"type": "array", "items": _S},
              "gaps": {"type": "array", "items": _S}},
             ("claims",))},
    ]


def _executor(frontier, label, meter, captured):
    def execute(name, args):
        args = args or {}
        if name == "submit_findings":   # accepted even over budget — findings
            captured[label] = args      # are the point of the spend
            return ("findings received — you are done; reply with a one-line "
                    "wrap-up")
        if meter.exhausted():
            frontier.mark_stop(label, f"token budget ({meter.budget})")
            return ("BUDGET REACHED: the run's token budget is exhausted — stop "
                    "exploring and submit_findings now")
        if name == "research_read":
            return frontier.read(label, args.get("note"))
        if name == "research_links":
            return frontier.links(label, args.get("note"))
        if name == "research_grep":
            return frontier.grep(label, args.get("pattern"))
        if name == "research_list":
            return frontier.list_notes(label, args.get("folder"))
        raise LookupError(f"unknown tool {name!r}")
    return execute


def _parse_json_reply(reply):
    text = reply.unfenced()
    for candidate in (text, text[text.find("{"): text.rfind("}") + 1]):
        try:
            obj = json.loads(candidate)
        except ValueError:
            continue
        if isinstance(obj, dict):
            return obj
    return None


def _vet_claims(raw_claims, own_reads):
    """Findings must carry sources (spec §Risks: citation hallucination): keep
    only sources this agent actually read; a claim with none left is dropped."""
    own = set(own_reads)
    kept, dropped = [], 0
    for c in raw_claims or []:
        if not isinstance(c, dict):
            dropped += 1
            continue
        sources = [s for s in (c.get("sources") or []) if s in own]
        if not sources:
            dropped += 1
            continue
        kept.append({"text": str(c.get("text") or "").strip(),
                     "sources": sources,
                     "date": c.get("date") if isinstance(c.get("date"), str) else None})
    return kept, dropped


def _run_agent(svc, spec, frontier, meter, plan_, results, cancelled=None, report=None):
    label = spec["label"]
    finding = {"agent": label, "region": spec["region"], "status": "done",
               "claims": [], "leads": [], "gaps": [], "notes_read": [],
               "dropped_sourceless_claims": 0}

    def on_turn(info):
        if report is not None:
            report({"stage": "gather", "agent": label, "turn": info["turn"],
                    "notes_read": len(frontier.reads_of(label))})

    captured = {}
    try:
        provider = meter.wrap(svc.get_provider(), label)
        user = {"question": plan_["question"], "seeds": spec["seeds"],
                "hint": "seeds are ranked entry points; other agents hold other regions"}
        reply, _msgs, _usage = run_tool_loop(
            provider,
            [{"role": "user", "content": json.dumps(user, ensure_ascii=False)}],
            _agent_tools(), _executor(frontier, label, meter, captured),
            system=_AGENT_SYSTEM.format(label=label, region=spec["region"]),
            model=plan_["model_gather"], max_tokens=4096,
            max_turns=2 * plan_["max_notes"] + 8,
            # the Frontier is lock-protected by design — a multi-read turn
            # may fan out; a cancelled job stops paying before its next call
            parallel_tools=True, should_stop=cancelled, on_turn=on_turn)
        # the schema-validated submission wins; prose JSON stays as fallback
        parsed = captured.get(label) or _parse_json_reply(reply)
        if parsed is None:
            # the loop already retried a max_tokens cutoff once with a doubled
            # ceiling — name the real cause if it STILL truncated
            finding["status"] = ("reply truncated at max_tokens — findings lost"
                                 if reply.stop_reason == "max_tokens"
                                 else "reply was not JSON — findings lost")
        else:
            claims, dropped = _vet_claims(parsed.get("claims"), frontier.reads_of(label))
            finding["claims"] = claims
            finding["dropped_sourceless_claims"] = dropped
            finding["leads"] = [str(l) for l in (parsed.get("leads") or [])][:LEADS_CAP]
            finding["gaps"] = [str(g) for g in (parsed.get("gaps") or [])][:LEADS_CAP]
    except ToolLoopCancelled:
        finding["status"] = "cancelled"
    except ToolLoopLimitError:
        finding["status"] = "turn-limit backstop hit — partial, findings lost"
    except ModelProviderError as e:
        finding["status"] = f"provider error: {e}"
    finding["notes_read"] = frontier.reads_of(label)
    stop = frontier.stops.get(label)
    if stop and finding["status"] == "done":
        finding["status"] = f"stopped at {stop}"
    results.append(finding)


# =====================================================================================
# deep mode — the verify pass
# =====================================================================================

_VERIFY_SYSTEM = """\
You verify one research claim against the notes it cites. "supported" means the cited \
text actually states or clearly implies the claim — not merely mentions its subject. \
Call the verdict tool: {"supported": true|false, "reason": "<one line>"}"""

# tool_choice forces this call — the verdict arrives schema-validated instead
# of scraped out of prose (the fail-open on a broken reply stays either way)
_VERDICT_TOOL = [{
    "name": "verdict",
    "description": "Submit the verification verdict.",
    "input_schema": {"type": "object", "additionalProperties": False,
                     "properties": {"supported": {"type": "boolean"},
                                    "reason": {"type": "string"}},
                     "required": ["supported"]},
}]


def _verify_claims(svc, frontier, findings, meter, plan_):
    """Re-open each cited note and confirm it supports the claim; demote what
    fails (kills plausible-but-wrong citations). Fail-open on a broken verify
    reply — verification exists to kill bad claims, not to eat good ones."""
    provider = meter.wrap(svc.get_provider(), "verify")
    demoted, stopped = [], False
    for finding in findings:
        kept = []
        for claim in finding["claims"]:
            if meter.exhausted():
                stopped = True
                claim["verified"] = None   # ran out of budget — kept, unverified
                kept.append(claim)
                continue
            payload = {"claim": claim["text"], "sources": {
                s: _safe_read(frontier, s) for s in claim["sources"][:3]}}
            try:
                reply = provider.chat(
                    [{"role": "user", "content": json.dumps(payload, ensure_ascii=False)}],
                    system=_VERIFY_SYSTEM, model=plan_["model_synth"], max_tokens=300,
                    tools=_VERDICT_TOOL,
                    tool_choice={"type": "tool", "name": "verdict"})
                verdict = (reply.tool_calls[0]["input"] if reply.tool_calls
                           else _parse_json_reply(reply))
            except ModelProviderError:
                verdict = None
            if verdict is not None and verdict.get("supported") is False:
                demoted.append({"agent": finding["agent"], "claim": claim["text"],
                                "sources": claim["sources"],
                                "reason": str(verdict.get("reason") or "not supported")})
            else:
                claim["verified"] = bool(verdict and verdict.get("supported"))
                kept.append(claim)
        finding["claims"] = kept
    return demoted, stopped


def _safe_read(frontier, rel):
    try:
        return fs.read_text(frontier.vault, rel, max_chars=VERIFY_CHAR_CAP)
    except (OSError, JailError):
        return "(unreadable)"


# =====================================================================================
# synthesis + citation discipline
# =====================================================================================

_SYNTH_SYSTEM = """\
You are the synthesis pass of a vault research run: merge the research agents' findings \
into one report answering the question. The findings come from the user's private \
Obsidian vault (their "brain"); every claim carries the source note paths it was read from.

Reply STRICT JSON only — no prose, no code fence:
{"tldr": "<2-4 sentences answering the question>",
 "summary": "<the executive section, ~300-500 words, markdown>",
 "report": "<the full report, markdown, sectioned>",
 "gaps": ["<what the brain has nothing on>"],
 "contradictions": ["<notes that disagree — cite both>"],
 "staleness": ["<claims resting only on old notes — cite them>"]}

Citations: cite claims inline as [[<note stem>]] Obsidian wikilinks — a claim sourced \
from 'Areas/Acme/Board Deck.md' cites [[Board Deck]]. Cite ONLY notes appearing in claim \
sources; never invent a citation. Weigh dates: prefer newer sources, flag claims whose \
only support is old, and surface disagreements between notes instead of papering over \
them. Name gaps honestly — "the brain has nothing on X" is a finding."""

_WIKILINK = re.compile(r"(!?)\[\[([^\]]+)\]\]")


def _sanitize_citations(text, frontier, visited, dropped):
    """Every [[wikilink]] in the output must resolve to a note this run actually
    read; anything else is downgraded to plain text and recorded — the report
    must never point at a note the spider didn't read (spec §Tests)."""
    def repl(m):
        raw = m.group(2).replace("\\|", "|")
        target = raw.split("|")[0].split("#")[0].strip()
        alias = raw.split("|", 1)[1].strip() if "|" in raw else None
        rel = frontier.resolve(target)
        if rel is not None and rel in visited:
            return m.group(0)
        dropped.append(target)
        return alias or target
    return _WIKILINK.sub(repl, str(text or ""))


def _fallback_synthesis(question, findings):
    """A deterministic zero-token report for when the synthesis model can't run
    (token budget exhausted) or returns no JSON — the run already spent the
    tokens; the findings must still come back."""
    lines, gaps = [], []
    for f in findings:
        for c in f["claims"]:
            cites = " ".join(f"[[{PurePosixPath(s).stem}]]" for s in c["sources"])
            lines.append(f"- {c['text']} {cites}".rstrip())
        gaps.extend(f["gaps"])
    n = sum(len(f["claims"]) for f in findings)
    return {
        "tldr": (f"Partial findings on: {question} — {n} claim(s) from "
                 f"{len(findings)} agent(s); synthesis pass skipped."),
        "summary": "\n".join(lines) or "No findings.",
        "report": "\n".join(lines) or "No findings.",
        "gaps": gaps, "contradictions": [], "staleness": [],
    }


def _synthesize(svc, frontier, findings, demoted, meter, plan_, partial_reasons):
    if meter.exhausted():
        partial_reasons.append("token budget exhausted before synthesis "
                               "(deterministic fallback report)")
        return _fallback_synthesis(plan_["question"], findings)
    provider = meter.wrap(svc.get_provider(), "synthesis")
    payload = {
        "question": plan_["question"], "mode": plan_["mode"],
        "findings": [{k: f[k] for k in ("agent", "region", "claims", "leads",
                                        "gaps", "status")} for f in findings],
        "demoted_claims": demoted,
        "coverage": {"agents": len(findings),
                     "notes_read": len(frontier.visited()),
                     "partial": partial_reasons},
    }
    try:
        reply = provider.chat(
            [{"role": "user", "content": json.dumps(payload, ensure_ascii=False)}],
            system=_SYNTH_SYSTEM, model=plan_["model_synth"], max_tokens=8192)
    except ModelProviderError as e:
        partial_reasons.append(f"synthesis provider error: {e}")
        return _fallback_synthesis(plan_["question"], findings)
    synth = _parse_json_reply(reply)
    if synth is None:
        partial_reasons.append("synthesis reply was not JSON (fallback report)")
        return _fallback_synthesis(plan_["question"], findings)
    for key in ("tldr", "summary", "report"):
        synth[key] = str(synth.get(key) or "")
    for key in ("gaps", "contradictions", "staleness"):
        synth[key] = [str(x) for x in (synth.get(key) or []) if str(x).strip()]
    return synth


# =====================================================================================
# save — the report as a dated, well-linked note
# =====================================================================================

def _save_report(svc, seat, plan_, synth, usage, matched_area):
    date = PurePosixPath(svc.daily_rel()).stem   # the owner's calendar day
    words = re.sub(r"[^A-Za-z0-9]+", " ", plan_["question"]).split()
    slug = " ".join(words[:6]) or "research"
    folder = f"Areas/{matched_area}/Research" if matched_area else "Knowledge/Research"
    body = (f"---\ntype: research\nquestion: {plan_['question']}\ndate: {date}\n"
            f"author: vault-mcp\n---\n\n# Research: {slug}\n\n"
            f"**TL;DR** — {synth['tldr']}\n\n## Summary\n\n{synth['summary']}\n\n"
            f"## Full report\n\n{synth['report']}\n")
    for name, items in (("Gaps", synth["gaps"]), ("Contradictions", synth["contradictions"]),
                        ("Staleness", synth["staleness"])):
        if items:
            body += f"\n## {name}\n\n" + "".join(f"- {i}\n" for i in items)
    body += (f"\n---\n*usage: {usage['total']['input_tokens']} in / "
             f"{usage['total']['output_tokens']} out tokens*\n")
    with svc.writer.lock:
        rel, n = f"{folder}/{date} {slug}.md", 2
        while (svc.vault / rel).exists():
            rel, n = f"{folder}/{date} {slug} ({n}).md", n + 1
        svc.writer.atomic_write(rel, body)
        sha = svc.writer.commit_and_push(f"vault-mcp({seat.id}): research {rel}")
    return rel, sha


# =====================================================================================
# the orchestrator — what the job runs
# =====================================================================================

def run(svc, seat, plan_, ctx=None):
    """The job body: seed → PEAK fan-out (parallel) → [deep: leads wave +
    verify] → synthesis → citation sanitation → the three disclosure layers
    (tldr / summary / detail — the P3 job_result convention).

    ``ctx`` is the JobManager's callable context: ``ctx["cancelled"]()`` gates
    every stage (agents stop paying mid-loop, later model stages are skipped,
    partials still come back), ``ctx["progress"]`` feeds job_result."""
    cancelled = (ctx or {}).get("cancelled") or (lambda: False)
    report = (ctx or {}).get("progress") or (lambda info: None)
    meter = _Meter(plan_["token_budget"])
    frontier = Frontier(svc, plan_["scope"], plan_["max_depth"], plan_["max_notes"])
    if not frontier.notes:
        empty = "nothing to research: no visible notes in scope."
        return {"tldr": empty, "summary": empty,
                "detail": {"question": plan_["question"], "agents": [],
                           "visited": [], "usage": {"total": meter.total(),
                                                    "by_stage": {}}}}

    ranked, matched_areas = seed(svc, frontier, plan_["question"])
    frontier.discover(ranked, 0)   # every seed is an entry point, whoever reads it

    # Deep mode reserves one agent slot for the wave-2 leads chase — max_agents
    # is a hard per-run cap, not a per-wave one.
    wave1_n = max(1, plan_["n_agents"] - 1) if plan_["mode"] == "deep" else plan_["n_agents"]
    specs = _agent_specs(frontier, ranked, matched_areas, wave1_n)

    findings = []
    threads = [threading.Thread(target=_run_agent, name=f"research-{s['label']}",
                                args=(svc, s, frontier, meter, plan_, findings,
                                      cancelled, report))
               for s in specs]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    order = {s["label"]: i for i, s in enumerate(specs)}
    findings.sort(key=lambda f: order.get(f["agent"], 99))

    partial_reasons = [f"{f['agent']}: {f['status']}" for f in findings
                       if f["status"] != "done"]

    # ---- wave 2 (deep): chase cross-pillar leads ------------------------------------
    if plan_["mode"] == "deep" and not meter.exhausted() and not cancelled():
        leads = [l for f in findings for l in f["leads"]]
        lead_seeds = _lead_seeds(frontier, leads)
        if lead_seeds:
            frontier.discover(lead_seeds, 0)
            spec = {"label": "leads", "prefixes": ("",), "seeds": lead_seeds,
                    "region": "no fixed pillar — chase these specific cross-pillar "
                              "leads from the first wave: " + "; ".join(leads[:LEADS_CAP])}
            _run_agent(svc, spec, frontier, meter, plan_, findings, cancelled, report)
            if findings[-1]["status"] != "done":
                partial_reasons.append(f"leads: {findings[-1]['status']}")

    # ---- verify pass (deep) ----------------------------------------------------------
    demoted = []
    if plan_["mode"] == "deep" and not cancelled():
        report({"stage": "verify"})
        demoted, verify_stopped = _verify_claims(svc, frontier, findings, meter, plan_)
        if verify_stopped:
            partial_reasons.append("token budget hit during verify "
                                   "(remaining claims kept unverified)")

    # ---- synthesis + citation discipline ---------------------------------------------
    if cancelled():
        # a cancelled run spends no further model calls — partials still ship
        partial_reasons.append("cancelled (deterministic fallback report)")
        synth = _fallback_synthesis(plan_["question"], findings)
    else:
        report({"stage": "synthesis"})
        synth = _synthesize(svc, frontier, findings, demoted, meter, plan_, partial_reasons)
    visited = set(frontier.visited())
    dropped_citations = []
    for key in ("tldr", "summary", "report"):
        synth[key] = _sanitize_citations(synth[key], frontier, visited, dropped_citations)
    for key in ("gaps", "contradictions", "staleness"):
        synth[key] = [_sanitize_citations(x, frontier, visited, dropped_citations)
                      for x in synth[key]]

    usage = {"total": meter.total(), "by_stage": meter.by_stage()}

    saved_rel = saved_sha = None
    if plan_["save"]:
        saved_rel, saved_sha = _save_report(
            svc, seat, plan_, synth, usage, matched_areas[0] if matched_areas else None)

    # ---- the three disclosure layers --------------------------------------------------
    tldr = " ".join(synth["tldr"].split()) or "research complete (no tldr produced)."
    if partial_reasons:
        tldr += " (partial run)"
    summary = synth["summary"].rstrip() + "\n"
    for name, items in (("Gaps", synth["gaps"]), ("Contradictions", synth["contradictions"]),
                        ("Staleness", synth["staleness"])):
        if items:
            summary += f"\n**{name}:**\n" + "".join(f"- {i}\n" for i in items)
    total = usage["total"]
    summary += (f"\n**Run:** {len(findings)} agent(s), {len(visited)} notes read; "
                f"{total['input_tokens']} in / {total['output_tokens']} out tokens.")
    if partial_reasons:
        summary += "\n**Partial:** " + "; ".join(partial_reasons)
    if saved_rel:
        summary += f"\nSaved to [[{PurePosixPath(saved_rel).stem}]] ({saved_rel})."

    detail = {
        "question": plan_["question"], "mode": plan_["mode"],
        "models": {"gather": plan_["model_gather"], "synthesize": plan_["model_synth"]},
        "report": synth["report"],
        "gaps": synth["gaps"], "contradictions": synth["contradictions"],
        "staleness": synth["staleness"],
        "agents": [{k: f[k] for k in ("agent", "region", "status", "claims", "leads",
                                      "gaps", "notes_read", "dropped_sourceless_claims")}
                   for f in findings],
        "demoted_claims": demoted,
        "dropped_citations": sorted(set(dropped_citations)),
        "visited": sorted(visited),
        "budgets": {"max_agents": plan_["n_agents"], "max_depth": plan_["max_depth"],
                    "max_notes_per_agent": plan_["max_notes"],
                    "token_budget": plan_["token_budget"]},
        "partial": partial_reasons,
        "usage": usage,
        "saved": {"path": saved_rel, "committed": saved_sha} if saved_rel else None,
    }
    return {"tldr": tldr, "summary": summary, "detail": detail}


def _lead_seeds(frontier, leads):
    """Resolve wave-1 lead strings to unread notes by stem-term match."""
    terms = set()
    for lead in leads:
        terms.update(_terms(lead))
    if not terms:
        return []
    unread = frontier.notes - set(frontier.visited())
    return [r for r in sorted(unread)
            if any(t in PurePosixPath(r).stem.lower() for t in terms)][:SEEDS_PER_AGENT]
