"""Every id on a confirmation card has to have been looked up.

The card shows ``套餐：Zoom Pro`` while the proposal carries ``plan_id=7f3a…``.
Those are two different values, and only one of them executes. The employee
cannot read a UUID, so the card deliberately hides it and shows the name the
model supplied beside it — which means the card's promise ("what you approved
is what runs") rests entirely on the model having paired them honestly. A model
that resolved the wrong row, or invented an id that happens to exist, produces a
card that is *correct on its face* and does something else.

Validation cannot fix that: an id is well-formed whatever it points at. What
can fix it is provenance. The turn already observes every tool result the model
read, so the id and its label can be required to have appeared **together in
one thing the model actually looked at**. An invented id was in no tool result;
an id lifted from the wrong row does not co-occur with the label claimed for it.

Refusing here costs the model one extra lookup and hands it a message saying so.
Refusing at execute time — after the tap — would spend the employee's trust.
"""
from __future__ import annotations

import logging

logger = logging.getLogger(__name__)

#: Around each id, this many characters of the result are treated as "the same
#: row". Wide enough for a JSON object of a few fields, narrow enough that a
#: label from a *different* record in the same list does not count.
_WINDOW = 400


class Corroborator:
    """Remembers what the model read this turn, and answers "did you see this?".

    One instance per turn. Empty until a tool returns something, so a turn that
    proposed without looking anything up corroborates nothing — which is the
    correct answer, not a gap.
    """

    __slots__ = ("_seen", "_repaired")

    def __init__(self) -> None:
        self._seen: list[str] = []
        #: Repair result per blob index, computed at most once. A turn checks
        #: every id against every blob, so without this a single card check
        #: re-parsed the same bytes ~18 times — 438 ms measured on a hostile
        #: input, all of it on the event loop.
        self._repaired: dict[int, object | None] = {}

    def observe(self, raw: str) -> None:
        """Record one tool result verbatim — unless it is a refusal.

        A rejected call echoes its own arguments back. FastMCP renders a type
        error as ``input_value='…'``, so a model that passes a fabricated id
        into an int-typed parameter gets that id read back to it, and a
        substring test would then call it corroborated (verified against four
        live registry tools — all four echoed). A tool that refused the call
        looked nothing up, so its output is not evidence of anything.
        """
        if not raw:
            return
        if _is_refusal(raw):
            return
        self._seen.append(raw)

    def saw(self, value: str) -> bool:
        """True when this exact string appeared in something the model read."""
        return any(value in blob for blob in self._seen)

    def saw_together(self, value: str, label: str) -> bool:
        """True when the two appeared in one record of something the model read.

        Two passes, because tool results come in two shapes. JSON gets read as
        JSON: "same record" then means *the same object*, which is exact — a
        label lifted from the neighbouring row in a list fails, where a purely
        positional check would pass it (two rows of a short list sit a few
        dozen characters apart). Everything else — prose, CSV-ish text — falls
        back to proximity, which is loose but is the only signal available.
        """
        for index, blob in enumerate(self._seen):
            verdict = _json_pairs(blob, value, label)
            if verdict is None and blob.lstrip()[:1] in ("[", "{"):
                # JSON that was cut off mid-way. Repair it rather than quietly
                # downgrading to proximity — see ``_repair_truncated_json``.
                if index not in self._repaired:
                    self._repaired[index] = _repair_truncated_json(blob)
                repaired = self._repaired[index]
                verdict = (
                    _pairs_in(repaired, value, label) if repaired is not None
                    else False
                )
            if verdict is True:
                return True
            if verdict is False:
                # Parsed cleanly and the pair is not in any one object. A
                # proximity retry on the same bytes would only re-admit what
                # the structure just ruled out.
                continue
            start = 0
            while (at := blob.find(value, start)) != -1:
                window = blob[max(0, at - _WINDOW) : at + len(value) + _WINDOW]
                if label in window:
                    return True
                start = at + len(value)
        return False


#: How the bridge and FastMCP say "this call did not happen". Matched against
#: the WHOLE blob, not a prefix — a refusal whose error text sits past the
#: first few hundred characters is still a refusal.
_REFUSAL_TEXT_MARKERS = (
    "validation error",
    "Error executing tool",
    "input_value=",
)


def _is_refusal(raw: str) -> bool:
    """Decided structurally for JSON, textually otherwise.

    Looking for the substring ``"error"`` was too blunt: a legitimate listing
    carrying ``"error": null``, or an article whose text contains the words
    "validation error", was discarded wholesale — and the model was then told
    to look up what it had just looked up, which it would retry into the same
    result.
    """
    import json  # noqa: PLC0415

    try:
        parsed = json.loads(raw)
    except (ValueError, TypeError):
        # Not JSON, so the only signal available is the wording.
        return any(marker in raw for marker in _REFUSAL_TEXT_MARKERS)
    # Parsed cleanly: the envelope decides, and the text pass MUST NOT run.
    # Applying it here discarded real results whose CONTENT mentions an error —
    # 「表单提交报 validation error 怎么办」 is a plausible help-desk ticket
    # title, and this assistant's whole domain is help-desk tickets. Same class
    # as the ``"error"``-substring bug this function was rewritten to fix.
    return isinstance(parsed, dict) and bool(parsed.get("error"))


def _element_boundaries(text: str) -> list[tuple[int, str]]:
    """Every position a complete value ends at, with the closers still needed.

    ONE left-to-right pass. The first shape called a bracket-stack helper per
    candidate — O(n) each, O(n²) overall — and a truncated tail that ends
    inside a long string makes every candidate scan the whole blob before
    failing. Measured 265 ms for one 5.8 KB blob and ~5 s for a single card
    check, on the event loop, reachable by pasting ``}]`` into your own support
    ticket. Bounding the number of scans capped the damage but also capped the
    recovery; computing the state once removes the need to choose.

    String-aware, because a brace inside a value is not structure.
    """
    stack: list[str] = []
    in_string = False
    escaped = False
    out: list[tuple[int, str]] = []
    opener = {"]": "[", "}": "{"}
    for i, ch in enumerate(text):
        if escaped:
            escaped = False
            continue
        if ch == "\\":
            escaped = in_string
            continue
        if ch == '"':
            in_string = not in_string
            continue
        if in_string:
            continue
        if ch in "[{":
            stack.append(ch)
        elif ch in "]}":
            if not stack or stack[-1] != opener[ch]:
                return out  # malformed from here on; keep what we have
            stack.pop()
            out.append((
                i + 1,
                "".join("]" if b == "[" else "}" for b in reversed(stack)),
            ))
    return out


def _repair_truncated_json(blob: str) -> object | None:
    """Parse a JSON listing that was cut mid-way, by dropping the partial tail.

    ``mcp_bridge`` hands the model ``json.dumps(result)[:6000]`` — a hard slice
    with no marker. The cut blob does not parse, and without this the check
    silently fell back to proximity matching for exactly the results where it
    matters most: a listing is truncated *because* it is long, and a long
    listing is where row A's id and row B's name are plausible neighbours.

    Returns ``None`` when nothing parseable can be salvaged — the caller then
    refuses to corroborate from this blob at all, rather than reaching for a
    weaker test on the same bytes.
    """
    import json  # noqa: PLC0415

    text = blob.rstrip()
    if not text.startswith(("[", "{")):
        return None

    #: Parses to attempt, newest boundary first. The complete prefix of a cut
    #: listing ends at one of the LAST few, so a real repair is found in single
    #: digits; the rest is headroom.
    for cut, closing in reversed(_element_boundaries(text)[-24:]):
        try:
            parsed: object = json.loads(text[:cut] + closing)
        except (ValueError, TypeError):
            continue
        return parsed
    return None


def _pairs_in(parsed: object, value: str, label: str) -> bool:
    """The structural test, against an already-parsed document."""
    stack: list[object] = [parsed]
    while stack:
        node = stack.pop()
        if isinstance(node, list):
            for element in node:
                scalars = _record_scalars(element)
                if value in scalars and label in scalars:
                    return True
            stack.extend(node)
        elif isinstance(node, dict):
            scalars = _record_scalars(node)
            if value in scalars and label in scalars:
                return True
            stack.extend(node.values())
    return False


def _record_scalars(node: object) -> set[str]:
    """The scalars that belong to ONE record.

    A record's own scalar fields, plus those of a nested object that is its
    only structured child. It stops at:

    * **any list** — its elements are records of their own, so
      ``{"rows": [rowA, rowB]}`` must not let row A's id pair with row B's name;
    * **sibling objects** — ``{"current_plan": {…}, "recommended_plan": {…}}``
      is two records in one envelope, and flattening it let the current plan's
      id pair with the recommended plan's name. That shape is everywhere:
      current/recommended, from/to, plan/upgrade, before/after.

    A single nested object still counts as the same record
    (``{"plan": {"id": …}, "name": …}``), because there is no second record for
    it to be confused with — and refusing that shape is what made a whole class
    of flows unproposable two rounds ago.
    """
    own: set[str] = set()
    children: list[object] = []
    if isinstance(node, dict):
        for value in node.values():
            if isinstance(value, (str, int, float)):
                own.add(str(value))
            elif isinstance(value, dict):
                children.append(value)
    elif isinstance(node, (str, int, float)):
        own.add(str(node))
    if len(children) == 1:
        own |= _record_scalars(children[0])
    return own


def _json_pairs(blob: str, value: str, label: str) -> bool | None:
    """Do ``value`` and ``label`` share one JSON object in ``blob``?

    ``None`` when the text is not JSON at all, so the caller knows to fall back
    rather than reading "not JSON" as "not corroborated".
    """
    import json  # noqa: PLC0415

    try:
        parsed = json.loads(blob)
    except (ValueError, TypeError):
        return None

    stack: list[object] = [parsed]
    while stack:
        node = stack.pop()
        if isinstance(node, list):
            # Each element is one record; ``_record_scalars`` does not descend
            # into further lists, so a wrapper cannot merge two rows into one.
            for element in node:
                scalars = _record_scalars(element)
                if value in scalars and label in scalars:
                    return True
            stack.extend(node)
        elif isinstance(node, dict):
            scalars = _record_scalars(node)
            if value in scalars and label in scalars:
                return True
            stack.extend(node.values())
    return False


def _claims_a_name(claimed: object) -> bool:
    """Would this display value put a name in front of the employee?"""
    if claimed is None:
        return False
    if isinstance(claimed, (list, tuple)):
        return any(str(x).strip() for x in claimed if x is not None)
    return bool(str(claimed).strip())


def _pair_up(
    value: object, claimed: object,
) -> list[tuple[str, str]] | None:
    """Line ids up with the names claimed for them, or ``None`` if they cannot.

    Both sides are coerced to lists of strings first, because neither is
    type-validated anywhere: ``_make_normalise`` does no type checking, and
    ``_strip_display`` removes the display field before FastMCP would ever see
    the declared ``"type": "array"``. So the shapes that arrive here are
    whatever the model sent.
    """
    ids = list(value) if isinstance(value, (list, tuple)) else [value]
    labels = list(claimed) if isinstance(claimed, (list, tuple)) else [claimed]

    # One name offered for a list of ids says nothing about which id it names.
    if len(labels) != len(ids):
        return None

    pairs: list[tuple[str, str]] = []
    for one, label in zip(ids, labels, strict=False):
        text_id = "" if one is None else str(one).strip()
        text_label = "" if label is None else str(label).strip()
        if not text_id and not text_label:
            # Neither side says anything. Nothing on the card, nothing to prove.
            continue
        if not text_id or not text_label:
            # Exactly one side is blank, and that is a mismatch the employee
            # cannot see. Skipping it — the first shape — let ONE empty entry
            # carry an uncorroborated id:
            #
            #   ids     = ["REAL-1", "SMUGGLED-CEO-ID"]
            #   display = ["Alice",  ""]
            #
            # ``_claims_a_name`` saw a name, ``_make_normalise`` saw a truthy
            # list, the card rendered 「Alice, 」 — and both ids executed. The
            # blank half is exactly where the smuggling fits.
            return None
        pairs.append((text_id, text_label))
    return pairs


def unclaimed_labels(
    params: dict[str, object], corroborator: Corroborator | None,
) -> list[str]:
    """Which ``*_id_display`` NAMES on this card were never actually looked up.

    Only labelled ids are judged, and that is the whole point. The first cut
    demanded provenance for every ``*_id``, on the theory that an id is a
    reference to a row. Two of those turned out to be wrong:

    * ``create_procurement_supplier.tax_id`` is a tax registration number the
      employee dictates. No lookup will ever contain it, so requiring one made
      that flow permanently unproposable — unsatisfiable by construction.
    * A tool returning a nested or enveloped shape cannot always pair id and
      label, so its flow died the same way, with a message telling the model to
      do what it had already done.

    The invariant actually at risk is narrower than "ids are real". It is: **the
    card shows a NAME and executes an ID, so a name the employee reads must be
    the name of the row that will run.** An id with no claimed name is not
    exposed to that mismatch — the card renders it raw, which is unreadable but
    honest, and execute-time validation still refuses a fabricated one.

    So the rule is: **claim a name, prove the name.** Returns the offending id
    field names; empty means nothing on this card asserts an unverified pairing.

    ``None`` for the corroborator means observation was never wired up — a
    caller bug, not an employee-facing one, so it passes rather than blocking
    every proposal in the product.
    """
    if corroborator is None:
        return []

    from app.assistant.derived_actions import DISPLAY_SUFFIX  # noqa: PLC0415

    bad: list[str] = []
    for name, value in params.items():
        if name.endswith(DISPLAY_SUFFIX):
            continue
        if not (name.endswith("_id") or name.endswith("_ids")):
            continue
        claimed = params.get(f"{name}{DISPLAY_SUFFIX}")
        if not _claims_a_name(claimed):
            continue  # nothing on the card asserts a name → nothing to disprove

        # The card renders ANY truthy display value, so anything truthy has to
        # be proven. Type-checking here and skipping on a mismatch let one
        # square bracket defeat the whole invariant: a fabricated name passed
        # as ``["CEO 张三"]`` instead of ``"CEO 张三"`` fell through the
        # ``isinstance(str)`` gate, was never corroborated, and still went on
        # the card. What the employee reads is the contract, not the type.
        pairs = _pair_up(value, claimed)
        if pairs is None:
            # Names and ids do not line up, so the card would show N names
            # beside M ids and the employee could not tell which is which.
            bad.append(name)
            continue
        for one, label in pairs:
            if not corroborator.saw_together(one, label):
                bad.append(name)
                break
    if bad:
        logger.info("assistant: proposal refused, names not corroborated: %s", bad)
    return bad