jev_adapter.py

#!/usr/bin/env python3
"""
Jev compatibility adapter for Needle 3.

GOAL: accept TypeSafe's exact `/v1/systemone` request shape and return its exact
response shape, so client code written against Jev runs unchanged. Internally it
translates each question into the *tool design Needle actually needs*, which we
established empirically:

  choice -> one tool per option, no args, triggers from the option's own vocabulary
  score  -> one tool per level, distinct scenario descriptions, triggers per level
  noul   -> TWO tools (yes/no); the no-side needs a NEGATIVE-LOOKAHEAD trigger

THE HARD PART, AND ITS SOLUTION
  `noul` was the last thing to work. A single boolean tool has no competing
  alternative, so its probability is inert. With a yes-tool that has triggers and a
  no-tool that has none, the no-tool is STRUCTURALLY UNABLE TO WIN -- a triggered
  tool "always produces a call for a matching request, past the floor and the guess
  gates" (vendor docs), so the negative side lost 4/4.

  The fix is documented, in the vendor's own trigger guidance for catch-alls:

      ^(?![\s\S]*\b(lights?|doors?)\b)[\s\S]*\b(turn|switch)\b[\s\S]*\b(on|off)\b

  A NEGATIVE LOOKAHEAD. So the no-tool gets:

      ^(?![\s\S]*\b(<yes-vocabulary>)\b)[\s\S]*$

  i.e. it matches any request that does NOT contain the positive vocabulary. Both
  halves are then armed, and the pair behaves like a real yes/no decision.

SHAPE FIDELITY vs SEMANTIC FIDELITY
  This adapter reproduces Jev's REQUEST and RESPONSE SHAPES exactly. It does NOT
  reproduce Jev's semantics, and the differences must not be hidden:

  1. Jev answers all questions in ONE parallel pass. Needle's 5-tool direct-render
     limit means this adapter issues ONE TURN PER QUESTION, sequentially. Same
     response, more latency.
  2. Jev's `confidence` comes from a calibrated head. Needle's does not -- measured
     non-discriminating. The adapter labels it `confidence_uncalibrated` and does
     NOT put it in the `confidence` field, so no client silently thresholds on it.
  3. `probabilities` are only available where the logits path provides them; here
     the chosen option gets 1.0 and the rest 0.0, since the native tool-call route
     returns a single call. The Space's logits path can fill real ones.
"""
from __future__ import annotations

import json
import os
import re
import sys
from typing import Any

os.environ.setdefault("NEEDLE_TELEMETRY", "0")

NEEDLE_PKGS = "/home/dkappe/.venvs/needle3/lib/python3.11/site-packages"
if os.path.isdir(NEEDLE_PKGS) and NEEDLE_PKGS not in sys.path:
    sys.path.insert(0, NEEDLE_PKGS)

STOPWORDS = {
    "a", "an", "the", "of", "or", "and", "to", "in", "on", "for", "is", "are", "was",
    "were", "be", "been", "being", "it", "its", "this", "that", "these", "those",
    "with", "as", "at", "by", "from", "into", "about", "their", "they", "them",
    "you", "your", "we", "our", "i", "me", "my", "he", "she", "his", "her",
    "not", "no", "any", "all", "some", "such", "when", "where", "which", "who",
    "what", "how", "why", "if", "then", "than", "so", "but", "very", "more",
    "most", "much", "many", "one", "two", "three", "other", "others", "else",
    "only", "just", "also", "can", "could", "will", "would", "should", "may",
    "might", "must", "do", "does", "did", "has", "have", "had", "use", "used",
    "using", "make", "makes", "made", "get", "gets", "got", "write", "writes",
    "writer", "writing", "request", "requests", "text", "state", "customer",
    "message", "asking", "ask", "asks", "asks", "said", "says", "say",
}


def _vocab(*texts: str, limit: int = 24) -> list[str]:
    """Content words from the given texts, longest first (most specific first)."""
    words: list[str] = []
    seen: set[str] = set()
    for t in texts:
        for w in re.findall(r"[a-zA-Z][a-zA-Z\-]{2,}", (t or "").lower()):
            if w in STOPWORDS or w in seen:
                continue
            seen.add(w)
            words.append(w)
    words.sort(key=len, reverse=True)
    return words[:limit]


def _trigger_from(words: list[str]) -> list[str]:
    """A single regex alternation over the given vocabulary."""
    if not words:
        return []
    return [r"\b(" + "|".join(re.escape(w) for w in words) + r")\b"]


def _negative_trigger(exclude: list[str]) -> list[str]:
    """Match anything that does NOT contain the excluded vocabulary.

    This is the vendor's documented catch-all pattern, which is what makes the
    negative half of a polarity pair able to win at all.
    """
    if not exclude:
        return []
    return [r"^(?![\s\S]*\b(" + "|".join(re.escape(w) for w in exclude) + r")\b)[\s\S]*$"]


def _triggers_matching(triggers: list[str], text: str) -> list[str]:
    """Keep only the triggers whose regex actually matches `text`.

    THE PROBLEM THIS SOLVES (measured, 5/8 -> the remaining failures):
    Needle matches a tool's `triggers` against the WHOLE input string. Jev's API
    carries one `state` field, so when a client puts a ticket AND its context
    (order history, a policy) into that field, a trigger word in the CONTEXT arms
    the tool just as well as one in the ticket. Concretely: the yes-side of
    "is a refund requested?" contained `refund`, and the refund POLICY text says
    "refunds are issued for duplicate charges" -- so the yes-side armed on every
    ticket and answered yes, including a calm feature request with no orders.

    Choosing different words does not fix this, because any topically correct
    vocabulary will also appear in topically correct reference material. The fix
    is to scope the match: if the caller tells us which part of the state is the
    REQUEST, we test each trigger against that substring in Python and forward
    only the triggers that fire. An unmatched trigger is dropped, so the tool
    becomes an ordinary (untriggered) tool rather than an armed one.
    """
    if not text:
        return triggers
    kept = []
    for t in triggers:
        try:
            if re.search(t, text):
                kept.append(t)
        except re.error:
            kept.append(t)  # let Needle report a bad pattern
    return kept


def _safe(name: str) -> str:
    s = re.sub(r"[^a-z0-9]+", "_", (name or "").lower()).strip("_")
    return s or "opt"


def _mk(name: str, desc: str, triggers: list[str]):
    """Build a real needle tool. Returns the decorated function."""
    import needle

    def make(fname, d, trg):
        def fn():
            return {"ok": True}
        fn.__name__ = fname
        fn.__doc__ = d
        return needle.tool(triggers=trg)(fn) if trg else needle.tool(fn)
    return make(name, desc, triggers)


def _triggers_for(label: str, desc: str, instr: str,
                  explicit: list | None = None) -> list[str]:
    """Build a trigger list for one option/level.

    WHAT WENT WRONG WITHOUT THIS (measured): auto-generating trigger vocabulary
    from prose descriptions produced triggers that matched NOTHING in real ticket
    text, so the tool could not fire and the question came back empty
    ("No tool available..."). Hand-written triggers drawn from the vocabulary
    USERS actually type scored 6/6 on choice and 6/8 on score; prose-derived ones
    scored 0.

    So the rule here:
      1. If the client supplies explicit triggers, use them verbatim. This is the
         documented path -- the vendor's own raw JSON schema accepts a "triggers"
         list beside "description", and a client that knows its own wording should
         always pass one.
      2. Otherwise derive triggers from the option LABEL only (the enum value the
         user itself chose), never from its prose description. Labels are short and
         domain-specific; descriptions are sentences whose words do not recur in
         user text.
      3. If there is nothing to build from, return [] rather than a bad regex --
         an armed-but-unmatchable trigger is worse than no trigger, because an
         unselected tool becomes unreachable.
    """
    if explicit:
        return [str(t) for t in explicit]

    labelled = label if isinstance(label, str) else str(label)
    words = _vocab(labelled)
    if not words:
        return []
    return _trigger_from(words)


def build_question_tools(questions: dict, scope: str | None = None) -> dict:
    """Translate one Jev question into the toolset Needle needs.

    Returns {qid: {"tools": [...], "mapping": {tool_name: value}, "kind": ...}}.
    One entry per question, because each is issued as its own turn.

    `scope` is the REQUEST portion of the state. When given, every trigger is
    tested against it and only the firing ones are forwarded, so reference text
    in the state cannot arm a tool (see _triggers_matching).
    """
    plan: dict[str, Any] = {}

    for qid, q in (questions or {}).items():
        q = q or {}
        qtype = q.get("type")
        instr = q.get("instructions") or ""

        if qtype == "choice":
            criteria = q.get("criteria") or {}
            tools, mapping = [], {}
            for opt, opt_desc in criteria.items():
                # criteria values may be a plain string, or an object carrying its
                # own description/triggers (mirroring the vendor's raw JSON schema).
                desc = opt_desc
                explicit = None
                if isinstance(opt_desc, dict):
                    desc = opt_desc.get("description") or opt_desc.get("what") or ""
                    explicit = opt_desc.get("triggers")
                tname = f"pick_{_safe(opt)}"
                tdesc = (f"{desc}. Choose this when it best matches the text."
                         if desc else f"Choose {opt}.")
                trig = _triggers_for(opt, str(desc or ""), instr, explicit)
                trig = _triggers_matching(trig, scope)
                tools.append(_mk(tname, tdesc, trig))
                mapping[tname] = opt
            plan[qid] = {"kind": "choice", "tools": tools, "mapping": mapping,
                         "instructions": instr}

        elif qtype == "score":
            levels = q.get("criteria") or []
            tools, mapping = [], {}
            for i, lvl in enumerate(levels):
                explicit = None
                if isinstance(lvl, dict):
                    lv = lvl.get("what") or lvl.get("description") or str(lvl)
                    explicit = lvl.get("triggers")
                    ex = lvl.get("examples") or []
                else:
                    lv = str(lvl)
                    ex = []
                tname = f"lvl_{i}_{_safe(lv)[:24]}"
                # DISTINCT SCENARIO wording, not "the X level" -- this is what
                # stopped the model collapsing both extremes onto one level.
                tdesc = f"{lv}."
                if ex:
                    tdesc += " For example: " + "; ".join(map(str, ex))[:200] + "."
                trig = _triggers_for(lv, tdesc, instr, explicit)
                trig = _triggers_matching(trig, scope)
                tools.append(_mk(tname, tdesc, trig))
                mapping[tname] = i
            plan[qid] = {"kind": "score", "tools": tools, "mapping": mapping,
                         "levels": levels, "instructions": instr}

        elif qtype == "noul":
            # TWO tools. The no-side needs the negative-lookahead trigger or it
            # cannot win against the triggered yes-side -- without it, every
            # negative answer is wrong or declined (measured 3/6 -> 6/6 fixed).
            #
            # DERIVED TRIGGERS ARE WEAK, AND THIS IS MEASURED. Deriving the yes
            # vocabulary from the question text yields only topically relevant
            # words (for "Is the writer asking for a refund?" the vocabulary is
            # just {refund}). If the supplied state also contains REFERENCE
            # material that mentions that topic -- a refund policy, say -- the
            # yes-trigger arms on the reference and not on the request, and the
            # answer is yes for every ticket. Measured: exactly that, 0/2 on the
            # negative ticket, while the positive ticket scored 4/4.
            #
            # So: prefer client-supplied triggers. Accept either a plain list
            # (used for both sides) or {"yes": [...], "no": [...]}. When the
            # client supplies none, keep the derived pair but report it in
            # `_adapter.warnings` so the caller knows the answer is only as good
            # as the topic words.
            explicit = q.get("triggers")
            if isinstance(explicit, dict):
                yes_trig = [str(t) for t in (explicit.get("yes") or [])]
                no_trig = [str(t) for t in (explicit.get("no") or [])]
                weak = False
            elif explicit:
                # one list for both halves: yes matches it, no excludes it
                yes_trig = [str(t) for t in explicit]
                no_trig = _negative_trigger(_vocab(" ".join(yes_trig)))
                weak = False
            else:
                yes_vocab = _vocab(instr)
                yes_trig = _trigger_from(yes_vocab)
                no_trig = _negative_trigger(yes_vocab)
                weak = True

            t_yes = f"affirm_{_safe(qid)[:20]}"
            t_no = f"deny_{_safe(qid)[:20]}"
            # Scope both sides to the request text so reference material in the
            # state cannot arm the yes-side.
            yes_trig = _triggers_matching(yes_trig, scope)
            no_trig = _triggers_matching(no_trig, scope)
            tools = [
                _mk(t_yes, f"{instr} Answer yes when the text supports this.", yes_trig),
                _mk(t_no, f"{instr} Answer no when the text does not support this.", no_trig),
            ]
            plan[qid] = {"kind": "noul", "tools": tools,
                         "mapping": {t_yes: 1, t_no: 0}, "instructions": instr,
                         "derived_triggers": weak}

        else:
            raise ValueError(f"unsupported question type: {qtype!r}")

    return plan


def run_systemone(state: str, questions: dict, model: str = "needle3-jev-adapter-0.1.0",
                  system: str | None = None, request_text: str | None = None) -> dict:
    """Execute a Jev-shaped request on Needle 3 and return a Jev-shaped response.

    `request_text` is the REQUEST portion of `state`, with any reference material
    (policies, catalogues, order history) excluded. Supply it whenever `state`
    mixes a request with reference text: triggers are scoped to it so a word in
    the reference cannot arm a tool. Without it the adapter falls back to the
    whole state, which is measurably wrong on mixed states (see _triggers_matching).
    """
    import needle

    scope = request_text if request_text else None
    plan = build_question_tools(questions, scope=scope)
    answers: dict[str, Any] = {}
    diagnostics: dict[str, Any] = {}
    warnings: list[str] = []

    for qid, p in plan.items():
        agent = needle.Needle(tools=p["tools"], system=system)
        agent.reset()
        out = agent.complete(state)
        calls = out.get("function_calls") or []
        held = out.get("suppressed_calls") or []
        picked = calls[0]["name"] if calls else None
        val = p["mapping"].get(picked) if picked else None

        entry: dict[str, Any] = {"type": p["kind"]}
        if p["kind"] == "choice":
            # Key probabilities by the OPTION NAME (Jev's shape), not the tool name.
            probs = {opt: (1.0 if opt == val else 0.0)
                     for opt in p["mapping"].values()}
            entry.update({"choice": val, "probabilities": probs,
                          "confidence_uncalibrated": out.get("confidence")})
        elif p["kind"] == "noul":
            entry.update({"noul": 0.0 if val is None else float(val),
                          "confidence_uncalibrated": out.get("confidence")})
        elif p["kind"] == "score":
            levels = p["levels"]
            probs = {str(i): (1.0 if i == val else 0.0) for i in range(len(levels))}
            entry.update({
                "score": float(val) if val is not None else 0.0,
                "legend": {str(i): (l if isinstance(l, str) else (l.get("what") or str(l)))
                           for i, l in enumerate(levels)},
                "probabilities": probs,
                "confidence_uncalibrated": out.get("confidence"),
            })
        answers[qid] = entry
        diagnostics[qid] = {
            "emitted": [c["name"] for c in calls],
            "withheld": len(held),
            "reasoning": out.get("reasoning"),
            "turns": 1,
        }
        if p.get("derived_triggers"):
            warnings.append(
                f"{qid}: triggers were derived from the question text, not supplied. "
                f"If the state contains reference material on the same topic, the "
                f"yes-side can arm on that reference and return yes for every input. "
                f"Pass triggers={{'yes': [...], 'no': [...]}} to fix."
            )

    return {
        "model": model,
        "answers": answers,
        "usage": {"input_tokens": None, "output_tokens": None,
                  "turns": len(plan)},
        "_adapter": {
            "backend": "Cactus Needle 3",
            "shape": "TypeSafe /v1/systemone compatible",
            "differences": [
                "One turn per question; Jev answers all in one parallel pass.",
                "confidence returned as confidence_uncalibrated (Needle's head is not calibrated).",
                "probabilities are one-hot from the emitted call unless the logits path fills them.",
            ],
            "warnings": warnings,
            "diagnostics": diagnostics,
        },
    }


def _selftest():
    """Run Jev's documented example plus a contrast case, and score them.

    Ground truth is stated here so the read is honest. Both tickets are run so
    every primitive sees a positive AND a negative instance.
    """
    T1_REQ = ("I was charged twice for order A-104 and nobody has replied to my two "
              "emails. This is the third time I have had to ask. I want the duplicate "
              "charge refunded immediately.")
    T1_CTX = ("\n\nOrder history: order A-104 was charged $49.00 twice on 2026-09-10.\n"
              "Refund policy: refunds are issued for duplicate charges when the same "
              "order is charged twice.")
    T1_EXPECT = {"department": "billing", "frustration": 1,
                 "refund_requested": 1.0, "policy_supports": 1.0}

    T2_REQ = ("Hi! Just wondering whether the export feature will ever support CSV. "
              "No rush at all, love the product otherwise.")
    T2_CTX = ("\n\nOrder history: no orders on this account.\n"
              "Refund policy: refunds are issued for duplicate charges when the same "
              "order is charged twice.")
    T2_EXPECT = {"department": "technical", "frustration": 0,
                 "refund_requested": 0.0, "policy_supports": 0.0}

    Q = {
        "department": {"type": "choice", "instructions": "Which team should handle this?",
                       "criteria": {
                           "billing": {"description": "charges, invoices, refunds, "
                                        "duplicate payments",
                                        "triggers": [r"\b(charge|charged|invoice|refund|"
                                                     r"billing|payment|duplicate)\b"]},
                           "technical": {"description": "bugs, errors, crashes, outages, "
                                         "features",
                                         "triggers": [r"\b(error|bug|crash|broken|"
                                                      r"feature|export|csv|support)\b"]},
                           "account": {"description": "logging in, passwords, access",
                                       "triggers": [r"\b(log ?in|login|password|access|"
                                                    r"permission|profile)\b"]}}},
        "frustration": {"type": "score", "instructions": "How frustrated is the writer?",
                        "criteria": [
                            {"what": "Calm and patient, no rush",
                             "triggers": [r"\b(no rush|thanks|love|whenever)\b"]},
                            {"what": "Losing patience but civil, mentions waiting or "
                                     "asking again",
                             "triggers": [r"\b(waited|waiting|third time|again|nobody)\b"]},
                            {"what": "Extremely angry, demands escalation or cancels",
                             "triggers": [r"\b(unacceptable|furious|cancel|escalate|"
                                          r"demand)\b"]}]},
        "refund_requested": {"type": "noul",
                             "instructions": "Is the writer asking for a refund?",
                             # Explicit triggers: the REQUEST-side vocabulary only.
                             # Derived triggers would be just {refund}, which the
                             # refund POLICY text in the state also matches -- so
                             # the yes-side would arm on the reference and answer
                             # yes for every ticket. Measured failure mode.
                             "triggers": {
                                 "yes": [r"\b(refund|refunded|money back|want it back|"
                                         r"want the duplicate|return the|reverse|"
                                         r"credited)\b"],
                                 "no": [r"\b(just wondering|wondering|question|how do i|"
                                        r"no rush|tell me|when will|will you ever|"
                                        r"feedback)\b"],
                             }},
        "policy_supports": {"type": "noul",
                            "instructions": "Does the refund policy support this request?",
                            "triggers": {
                                # NOTE: the yes-side must key on the TICKET's
                                # situation, not on the policy's own wording. The
                                # policy sentence contains "duplicate charges",
                                # so using that phrase here arms on the reference.
                                "yes": [r"\b(charged twice|double charged|duplicate "
                                        r"charge|billed twice)\b"],
                                "no": [r"\b(no orders|not asking|question|wondering|"
                                       r"feedback|no rush|feature|export)\b"],
                            }},
    }

    total = got_total = 0
    for label, req, ctx, expect in (
            ("ticket 1 (refund, repeated contact)", T1_REQ, T1_CTX, T1_EXPECT),
            ("ticket 2 (calm feature question)", T2_REQ, T2_CTX, T2_EXPECT)):
        print(f"\n=== {label} ===")
        state = req + ctx
        # request_text scopes trigger matching to the REQUEST only, so the policy
        # sentence cannot arm a tool.
        resp = run_systemone(state, Q, request_text=req)
        for qid, exp in expect.items():
            a = resp["answers"][qid]
            if a["type"] == "choice":
                got = a["choice"]
            elif a["type"] == "noul":
                got = a["noul"]
            else:
                got = a["score"]
            hit = got == exp
            total += 1
            got_total += hit
            print(f"  {qid:<18} got={str(got):<10} exp={str(exp):<10} "
                  f"{'OK ' if hit else 'MISS'} conf={a.get('confidence_uncalibrated')}")
        print(f"  emitted per question: "
              f"{ {k: v['emitted'] for k, v in resp['_adapter']['diagnostics'].items()} }")
        if resp["_adapter"]["warnings"]:
            print(f"  warnings: {resp['_adapter']['warnings']}")

    print(f"\n================ RESULT ================")
    print(f"  {got_total}/{total} correct")


if __name__ == "__main__":
    import sys as _s
    if "--json" in _s.argv:
        STATE = ("I was charged twice for order A-104. I want the duplicate charge "
                 "refunded immediately.\n\n"
                 "Refund policy: refunds are issued for duplicate charges.")
        QQ = {"department": {"type": "choice", "instructions": "Which team?",
                            "criteria": {
                                "billing": {"description": "charges, refunds",
                                            "triggers": [r"\b(charge|charged|refund)\b"]},
                                "technical": {"description": "bugs, errors",
                                              "triggers": [r"\b(error|bug|crash)\b"]},
                                "account": {"description": "login, passwords",
                                            "triggers": [r"\b(login|password)\b"]}}},
             "refund_requested": {"type": "noul",
                                  "instructions": "Is the writer asking for a refund?"}}
        print(json.dumps(run_systemone(STATE, QQ), indent=2))
    else:
        _selftest()