#!/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()\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()