diff --git a/docs/commands.md b/docs/commands.md index 9eb6d353..2b58132d 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -48,6 +48,8 @@ pipeline / run / ingest | `sources --route live` | Preview resolve | | `ingest "" --route offline` | Ingest only (structured + fingerprint) | | `analyze "" --route offline` | Analyze without auto-receipt | +| `classify ""` | Family of one short term. Registry `match_lineage`, else `none` | +| `classify "" --jevgate` | Same cascade, then opt-in [jevgate-1](jevgate.md) on a miss | Prefer **`--route offline|live|glossary|social`** over raw adapter names. Aliases: `real`→glossary, `x`→x_search, `firecrawl`→crawl4ai (Crawl4AI 0.9.3 default; no paid Firecrawl without Danny yes), `live`→combined. diff --git a/docs/jevgate.md b/docs/jevgate.md new file mode 100644 index 00000000..0468b3bc --- /dev/null +++ b/docs/jevgate.md @@ -0,0 +1,49 @@ +# jevgate-1 (opt-in classify) + +Family classification of one short term into the eight hyperlexical families or `none`. + +The default path is unchanged: lexical `match_lineage` (vector re-rank off). If that rule fires, its family decides. If it misses, the cascade returns `none`. + +## Enable + +```bash +export HYPERLEX_JEVGATE=1 +python3 scripts/hyperlex.py classify "" + +# or one shot, without the env var +python3 scripts/hyperlex.py classify "" --jevgate +``` + +`--no-jevgate` forces the registry cascade even when `HYPERLEX_JEVGATE` is set. + +Package entry point: `python -m hyperlex classify "" --jevgate`. + +## Gate + +Frozen by prereg v5. Do not tune. + +| Item | Value | +|------|--------| +| Model | `jev-1.13.0` (pinned; a missing or different `model` id is an error) | +| Passes | 3 identical requests; probabilities are means | +| Tau | `0.30333` | +| Decision | a family only when mean `P(none)` is **strictly below** tau; otherwise `none` | +| Ties | earliest family in `FAMILIES` from `hyperlexical.layout` | +| Prompt sha256 | `0dab58787402cb6e5aadc92e3123a65f09d281444e100b8485ede273c7d9af53` | + +`sha256(JSON.stringify(QUESTIONS))` is checked when the module loads. If it differs, the Jev path is refused. + +The gate **abstains often by design**. `none` is the expected output for ordinary English, names, noise, and slang from communities outside the eight. + +`jev_best_guess`, `jev_p_none`, and the per-family means are low-confidence fields. They never change `family`. `jev_best_guess` is the argmax over the eight families, ignoring tau, except when Jev gives no family signal: every family mean is 0, or all eight means are equal so the argmax would be only the layout tie-break. In that case `jev_best_guess` is null. The gated `family` is unchanged. Jev's own `choice`, `in_scope`, and `any_slang` may be logged under `jev_log` and never decide. + +On provider error, timeout, parse failure, model mismatch, or a missing key, the call is retried once. If it still fails, the result is `none` with `jev_error.flag` and `jev_error.reason` set. Jev failures do not raise into callers. + +## API key + +The key is read only from the environment: + +- `JEV_API_KEY` (preferred), or `TYPESAFE_AI_API_KEY` +- `JEV_KEY_FILE` — path of a file that contains the key (no default path) + +There is no key in the repo. `HYPERLEX_OFFLINE=1` refuses the network call and fail-closes to `none`. diff --git a/mkdocs.yml b/mkdocs.yml index 54d0e670..393e6660 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -78,6 +78,7 @@ nav: - Virality: modules/virality.md - Vector DB: modules/vectordb.md - LLM: modules/llm.md + - jevgate-1 classify: jevgate.md - Phase 5 research: phase5.md - Simulation modules: modules/simulation.md - Brier design: brier-calibration.md diff --git a/scripts/hyperlex.py b/scripts/hyperlex.py index f8a35d0c..7a8c2bfe 100755 --- a/scripts/hyperlex.py +++ b/scripts/hyperlex.py @@ -726,6 +726,18 @@ def _resolve_log_path(args: argparse.Namespace) -> Path: return default_log_path() +def cmd_classify(args: argparse.Namespace) -> int: + from hyperlex.analysis.jevgate import classify_term, jevgate_from_args + + term = (getattr(args, "term_pos", None) or getattr(args, "term", None) or "").strip() + if not term: + _emit({"ok": False, "command": "classify", "error": "empty term", "brier": None}) + return 2 + out = classify_term(term, jevgate=jevgate_from_args(args)) + _emit({"ok": True, "command": "classify", **out}) + return 0 + + def cmd_analyze(args: argparse.Namespace) -> int: pkg, err = _import_hyperlex() if pkg is None: @@ -922,6 +934,8 @@ def cmd_commands(_args: argparse.Namespace) -> int: {"cmd": "sources --route live", "why": "Preview resolve for a route"}, {"cmd": "ingest \"\" --route offline", "why": "Ingest only (structured + fingerprint)"}, {"cmd": "analyze \"\" --route offline", "why": "Analyze without auto-receipt"}, + {"cmd": "classify \"\"", "why": "Family cascade: match_lineage, else none (jevgate off)"}, + {"cmd": "classify \"\" --jevgate", "why": "Opt-in jevgate-1 after a registry miss (prereg v5)"}, ], "research": [ {"cmd": "simulate --term --mode scenario", "why": "Phase 5 research (SPECULATIVE, brier null)"}, @@ -2528,6 +2542,24 @@ def _add_pipeline_args(p: argparse.ArgumentParser, *, default_route: str = "offl analyze_parser.add_argument("--out") analyze_parser.set_defaults(func=cmd_analyze) + classify_parser = subparsers.add_parser( + "classify", + help="Classify a short term into 8 hyperlexical families or none", + ) + classify_parser.add_argument("term_pos", nargs="?", default="", help="Short slang term") + classify_parser.add_argument("--term", default="", help="Short slang term (flag form)") + classify_parser.add_argument( + "--jevgate", + action="store_true", + help="Opt in to jevgate-1 after match_lineage misses (prereg v5). Default off.", + ) + classify_parser.add_argument( + "--no-jevgate", + action="store_true", + help="Force the registry cascade even if HYPERLEX_JEVGATE is set", + ) + classify_parser.set_defaults(func=cmd_classify) + run_parser = subparsers.add_parser( "run", help="AUTO backend (alias of pipeline): ingest → full results packet", diff --git a/src/hyperlex/analysis/jevgate.py b/src/hyperlex/analysis/jevgate.py new file mode 100644 index 00000000..835b4fee --- /dev/null +++ b/src/hyperlex/analysis/jevgate.py @@ -0,0 +1,604 @@ +"""Opt-in jevgate-1 family classifier. + +Cascade (flag off is the default): + +1. Lexical ``match_lineage`` (vector re-rank off). If it fires, that family decides. +2. Otherwise, if ``HYPERLEX_JEVGATE`` / ``--jevgate`` is off, return ``none``. +3. Otherwise call TypeSafe Jev, model pinned to ``jev-1.13.0``, three identical + passes. Emit a family only when mean ``P(none)`` is strictly below tau. + Tau ``0.30333`` is frozen by prereg v5. + +Jev failures fail closed to ``none`` after one identical retry. They do not raise +into callers. ``jev_best_guess`` is low-confidence and never decides. + +The prompt object is frozen. ``sha256(JSON.stringify(QUESTIONS))`` is checked at +import; a mismatch refuses the Jev path. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import re +import socket +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any, Callable, Mapping + +# Frozen by prereg v5. Do not tune. +TAU = 0.30333 +MODEL_ID = "jev-1.13.0" +PASSES = 3 +REQUEST_TIMEOUT_S = 55.0 +FROZEN_PROMPT_SHA256 = "0dab58787402cb6e5aadc92e3123a65f09d281444e100b8485ede273c7d9af53" +JEVGATE_ENV = "HYPERLEX_JEVGATE" +DEFAULT_BASE_URL = "https://api.typesafe.ai/v1" + +CLASSIFY_CONFIG = { + "gate": "jevgate-1", + "env": JEVGATE_ENV, + "default": False, + "model": MODEL_ID, + "tau": TAU, + "passes": PASSES, + "prompt_sha256": FROZEN_PROMPT_SHA256, +} + +_FAMILY: tuple[tuple[str, str], ...] = ( + ( + "betting-sharp", + "Sports-betting, gambling and poker bettor jargon (odds, lines, wagers, bankroll, handicapping culture).", + ), + ( + "crypto-degen", + "Cryptocurrency, DeFi, NFT and speculative-trading community slang.", + ), + ( + "ai-native", + "AI / LLM / machine-learning and AI-builder vernacular (models, prompting, agents, training, evals, AI-era coinages).", + ), + ( + "brainrot-aura", + 'Gen-Z / Gen-Alpha internet "brainrot" slang: status, vibe, aura, meme and TikTok-era coinages.', + ), + ( + "kinship-address", + "Forms of address and pseudo-kinship terms used for friends or strangers (bro/sis-style, family-style address).", + ), + ( + "political-status", + "Online political, ideological and status-signalling discourse slang.", + ), + ( + "gaming-meta", + "Video-game, esports and gamer-community jargon (game mechanics, balance, competitive play).", + ), + ( + "workplace-corp", + "Corporate, office and workplace jargon and career-culture coinages.", + ), +) + +_NONE_CRITERION = ( + "Not slang/jargon of any listed community: ordinary English, a proper name, " + "noise, a fragment, or slang from an unlisted community." +) + +_EIGHT: tuple[str, ...] | None = None + + +class JevCallError(Exception): + """One failed Jev attempt. Callers of ``classify_term`` do not see this.""" + + def __init__(self, reason: str, detail: str = "") -> None: + self.reason = reason + self.detail = detail + super().__init__(reason) + + +def _js_stringify(obj: Any) -> str: + """Match JavaScript ``JSON.stringify`` (no spaces, insertion order).""" + return json.dumps(obj, ensure_ascii=False, separators=(",", ":")) + + +def _build_questions() -> dict[str, Any]: + family = dict(_FAMILY) + famlist = "\n".join(f"- {key}: {value}" for key, value in family.items()) + in_scope = ( + "STATE.text is a short term or phrase (often a single word or stem, sometimes lowercased or truncated). " + "Answer yes only if STATE.text is slang, jargon or vernacular that clearly belongs to at least one of these eight communities:\n" + f"{famlist}\n" + "Answer no if it is: ordinary standard English in its everyday dictionary sense; a proper name, brand, place or hashtag with no community-slang sense; a fragment, typo, number or noise; or slang/informal language that belongs to some OTHER community or general informal English not listed above. When unsure, answer no." + ) + any_slang = ( + "STATE.text is a short term or phrase. Is it slang, jargon, internet vernacular or an informal coinage of ANY kind (any community, any era), as opposed to ordinary standard English, a proper name, or noise?" + ) + family_q = ( + "STATE.text is a short term or phrase. Which single community does it belong to as slang/jargon? Choose none if it is not slang of any of the listed communities (ordinary English, a name, noise, or slang of another community)." + ) + criteria = dict(family) + criteria["none"] = _NONE_CRITERION + return { + "in_scope": {"type": "boolean", "instructions": in_scope}, + "any_slang": {"type": "boolean", "instructions": any_slang}, + "family": {"type": "choice", "instructions": family_q, "criteria": criteria}, + } + + +QUESTIONS = _build_questions() +CRITERIA_KEYS = tuple(QUESTIONS["family"]["criteria"].keys()) + + +def prompt_sha256() -> str: + return hashlib.sha256(_js_stringify(QUESTIONS).encode("utf-8")).hexdigest() + + +def assert_prompt_sha() -> None: + """Load-time and call-time check. Raises AssertionError if the prompt drifted.""" + digest = prompt_sha256() + assert digest == FROZEN_PROMPT_SHA256, digest + + +def prompt_is_frozen() -> bool: + try: + assert_prompt_sha() + except AssertionError: + return False + return True + + +def _prompt_ok_at_load() -> bool: + """Check the frozen prompt when the module loads. + + A mismatch does not raise here: the registry cascade must still import. + ``classify_term`` refuses the Jev path, and ``assert_prompt_sha`` raises. + """ + try: + assert_prompt_sha() + except AssertionError: + return False + return True + + +PROMPT_SHA_OK = _prompt_ok_at_load() +PROMPT_SHA256 = prompt_sha256() + + +def _truthy(raw: str | None) -> bool: + return str(raw or "").strip().lower() in {"1", "true", "yes", "on"} + + +def jevgate_enabled(flag: bool | None = None) -> bool: + """Explicit bool wins. Otherwise ``HYPERLEX_JEVGATE`` (default off).""" + if flag is not None: + return bool(flag) + return _truthy(os.environ.get(JEVGATE_ENV)) + + +def jevgate_from_args(args: Any) -> bool | None: + """``--no-jevgate`` forces the cascade off; ``--jevgate`` forces it on.""" + if bool(getattr(args, "no_jevgate", False)): + return False + if bool(getattr(args, "jevgate", False)): + return True + return None + + +def scrub_text(text: str, key: str = "") -> str: + """Drop the API key and bearer tokens from anything we might surface.""" + cleaned = str(text if text is not None else "") + if key: + cleaned = cleaned.replace(key, "[R]") + cleaned = re.sub(r"Bearer\s+\S+", "Bearer [R]", cleaned, flags=re.IGNORECASE) + return cleaned[:200] + + +def _offline() -> bool: + return _truthy(os.environ.get("HYPERLEX_OFFLINE")) + + +def load_api_key() -> str: + """Key from the environment only. No default path and no hardcoded key.""" + key = os.environ.get("JEV_API_KEY", "").strip() + if key: + return key + key = os.environ.get("TYPESAFE_AI_API_KEY", "").strip() + if key: + return key + path = os.environ.get("JEV_KEY_FILE", "").strip() + if not path: + raise JevCallError("missing_key", "JEV_API_KEY is not set") + try: + key = Path(path).expanduser().read_text(encoding="utf-8").strip() + except OSError: + raise JevCallError("missing_key", "JEV_KEY_FILE could not be read") from None + if not key: + raise JevCallError("missing_key", "JEV_KEY_FILE is empty") + return key + + +def _is_prob(value: Any) -> bool: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return False + return math.isfinite(float(value)) + + +def wire_questions(questions: Mapping[str, Any] | None = None) -> dict[str, Any]: + """SDK sends boolean questions as TypeSafe ``noul``. Does not mutate QUESTIONS.""" + source = QUESTIONS if questions is None else questions + wired: dict[str, Any] = {} + for key, question in source.items(): + copied = dict(question) + if copied.get("type") == "boolean": + copied["type"] = "noul" + criteria = copied.get("criteria") + if isinstance(criteria, dict): + copied["criteria"] = dict(criteria) + wired[key] = copied + return wired + + +def _layout_module(): + import importlib + import importlib.util + + try: + return importlib.import_module("hyperlexical.layout") + except ImportError: + path = Path(__file__).resolve().parents[3] / "scripts" / "shadow" / "hyperlexical" / "layout.py" + if not path.is_file(): + raise JevCallError("layout_unavailable", "hyperlexical.layout is not importable") from None + spec = importlib.util.spec_from_file_location("hyperlexical.layout", path) + if spec is None or spec.loader is None: + raise JevCallError("layout_unavailable", "hyperlexical.layout is not importable") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def eight_families() -> tuple[str, ...]: + """Family order from ``hyperlexical.layout.FAMILIES``, excluding ``none``.""" + global _EIGHT + if _EIGHT is not None: + return _EIGHT + layout = _layout_module() + order = tuple(name for name in layout.FAMILIES if name != "none") + if len(order) != 8: + raise JevCallError("layout_unavailable", "FAMILIES must list 8 families plus none") + _EIGHT = order + return order + + +def _normalize_answer(answer: Any) -> dict[str, Any]: + if not isinstance(answer, dict): + raise JevCallError("parse_error", "answer is not an object") + kind = answer.get("type") + if kind == "noul": + return {"type": "boolean", "probability": answer.get("noul")} + if kind == "boolean": + return {"type": "boolean", "probability": answer.get("probability")} + if kind == "choice": + return { + "type": "choice", + "choice": answer.get("choice"), + "probabilities": answer.get("probabilities"), + } + raise JevCallError("parse_error", "unsupported answer type") + + +def normalize_evaluate_payload(payload: Any) -> dict[str, Any]: + """Map a ``/systemone`` body onto the SDK answer shape. + + ``model`` must be present and equal ``jev-1.13.0``. A missing id is a + mismatch, not the requested model filled in after the fact. + """ + if not isinstance(payload, dict): + raise JevCallError("parse_error", "response is not an object") + answers_raw = payload.get("answers") + if not isinstance(answers_raw, dict): + raise JevCallError("parse_error", "answers missing") + answers = {key: _normalize_answer(value) for key, value in answers_raw.items()} + return {"model_id": payload.get("model"), "answers": answers} + + +def _validate_pass(result: Mapping[str, Any]) -> dict[str, Any]: + model_id = result.get("model_id") + if model_id != MODEL_ID: + raise JevCallError("model_mismatch", f"model_id={model_id}") + answers = result.get("answers") + if not isinstance(answers, dict): + raise JevCallError("parse_error", "answers missing") + in_scope = answers.get("in_scope") + any_slang = answers.get("any_slang") + family = answers.get("family") + if not isinstance(in_scope, dict) or in_scope.get("type") != "boolean": + raise JevCallError("parse_error", "in_scope") + if not isinstance(any_slang, dict) or any_slang.get("type") != "boolean": + raise JevCallError("parse_error", "any_slang") + if not _is_prob(in_scope.get("probability")) or not _is_prob(any_slang.get("probability")): + raise JevCallError("parse_error", "boolean probability") + if not isinstance(family, dict): + raise JevCallError("parse_error", "family") + choice = family.get("choice") + if choice not in CRITERIA_KEYS: + raise JevCallError("parse_error", "family choice") + probs = family.get("probabilities") + if not isinstance(probs, dict) or any(not _is_prob(probs.get(key)) for key in CRITERIA_KEYS): + raise JevCallError("parse_error", "family probabilities") + return { + "model_id": model_id, + "choice": choice, + "in_scope": float(in_scope["probability"]), + "any_slang": float(any_slang["probability"]), + "probabilities": {key: float(probs[key]) for key in CRITERIA_KEYS}, + } + + +def _mean(values: list[float]) -> float: + return sum(values) / len(values) + + +def _argmax(means: Mapping[str, float], order: tuple[str, ...]) -> str: + """First family in ``order`` wins ties (strict ``>``).""" + best = order[0] + best_p = float(means[best]) + for name in order[1:]: + score = float(means[name]) + if score > best_p: + best = name + best_p = score + return best + + +def _flat_family_means(means: Mapping[str, float], order: tuple[str, ...]) -> bool: + """True when every family mean is the same, including all zeros. + + Argmax would be only the layout tie-break, so there is no family signal. + """ + if not order: + return True + first = float(means[order[0]]) + return all(float(means[name]) == first for name in order) + + +def decide_family(passes: list[Mapping[str, Any]], order: tuple[str, ...]) -> dict[str, Any]: + """Aggregate pass means. Choice / in_scope / any_slang are logged only.""" + p_none = _mean([float(item["probabilities"]["none"]) for item in passes]) + means = { + name: _mean([float(item["probabilities"][name]) for item in passes]) + for name in order + } + ranked = _argmax(means, order) + family = ranked if p_none < TAU else "none" + best = None if _flat_family_means(means, order) else ranked + return { + "family": family, + "jev_best_guess": best, + "jev_p_none": p_none, + "jev_family_means": means, + "jev_log": { + "choice": [item["choice"] for item in passes], + "in_scope": [item["in_scope"] for item in passes], + "any_slang": [item["any_slang"] for item in passes], + }, + } + + +class JevHttpClient: + """POST ``{base}/systemone``, the same body ``@ai-sdk/typesafe-ai`` sends.""" + + def __init__(self, *, timeout: float = REQUEST_TIMEOUT_S) -> None: + self.timeout = timeout + + def evaluate(self, state: Mapping[str, Any]) -> dict[str, Any]: + if _offline(): + raise JevCallError("offline", "HYPERLEX_OFFLINE refuses the Jev call") + key = load_api_key() + base = os.environ.get("JEV_BASE_URL", DEFAULT_BASE_URL).strip() or DEFAULT_BASE_URL + try: + from hyperlex.guards import require_http_url + + base = require_http_url(base, name="JEV_BASE_URL").rstrip("/") + except Exception as exc: + raise JevCallError("provider_error", scrub_text(str(exc), key)) from None + body = { + "model": MODEL_ID, + "state": {"text": state.get("text")}, + "questions": wire_questions(), + } + payload = json.dumps(body, ensure_ascii=False).encode("utf-8") + request = urllib.request.Request( + f"{base}/systemone", + data=payload, + method="POST", + headers={ + "Authorization": f"Bearer {key}", + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": "hyperlex-jevgate/1", + }, + ) + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + raw = response.read() + except urllib.error.HTTPError as exc: + try: + detail = exc.read().decode("utf-8", "replace") + except Exception: + detail = "" + raise JevCallError("provider_error", scrub_text(detail or str(exc), key)) from None + except (urllib.error.URLError, TimeoutError, socket.timeout) as exc: + message = str(getattr(exc, "reason", None) or exc) + reason = "timeout" if re.search(r"timed out|timeout|aborted", message, re.I) else "provider_error" + raise JevCallError(reason, scrub_text(message, key)) from None + except Exception as exc: + raise JevCallError("provider_error", scrub_text(str(exc), key)) from None + try: + parsed = json.loads(raw.decode("utf-8")) + except (UnicodeError, json.JSONDecodeError): + raise JevCallError("parse_error", "response was not json") from None + try: + return normalize_evaluate_payload(parsed) + except JevCallError: + raise + except Exception as exc: + raise JevCallError("parse_error", scrub_text(str(exc), key)) from None + + +def _error_payload(exc: JevCallError, key: str = "") -> dict[str, Any]: + detail = scrub_text(exc.detail, key) + payload: dict[str, Any] = {"flag": True, "reason": exc.reason} + if detail: + payload["detail"] = detail + return payload + + +def _blank( + term: str, + *, + family: str, + source: str, + enabled: bool, + lineage: dict[str, Any] | None = None, + jev_error: dict[str, Any] | None = None, + gate: dict[str, Any] | None = None, +) -> dict[str, Any]: + return { + "schema": "hyperlex.classify.v1", + "gate": "jevgate-1", + "term": term, + "family": family, + "source": source, + "jevgate_enabled": enabled, + "tau": TAU, + "model": MODEL_ID, + "prompt_sha256": FROZEN_PROMPT_SHA256, + "lineage": lineage, + "jev_best_guess": None if gate is None else gate.get("jev_best_guess"), + "jev_p_none": None if gate is None else gate.get("jev_p_none"), + "jev_family_means": None if gate is None else gate.get("jev_family_means"), + "jev_log": None if gate is None else gate.get("jev_log"), + "jev_error": jev_error, + "brier": None, + } + + +def _one_pass(client: Any, state: dict[str, str], key: str) -> dict[str, Any]: + """One request, then one identical retry. The second failure propagates.""" + last: JevCallError | None = None + for _attempt in (0, 1): + try: + raw = client.evaluate(state) + if not isinstance(raw, Mapping): + raise JevCallError("parse_error", "evaluate did not return an object") + # Accept either a wire payload or the already-normalized SDK shape. + if "answers" in raw and "model_id" not in raw: + raw = normalize_evaluate_payload(raw) + return _validate_pass(raw) + except JevCallError as exc: + last = JevCallError(exc.reason, scrub_text(exc.detail, key)) + except Exception as exc: + last = JevCallError("provider_error", scrub_text(str(exc), key)) + assert last is not None + raise last + + +def _current_key() -> str: + """Best-effort key for scrubbing errors. Empty if none is configured.""" + key = os.environ.get("JEV_API_KEY", "").strip() or os.environ.get("TYPESAFE_AI_API_KEY", "").strip() + if key: + return key + path = os.environ.get("JEV_KEY_FILE", "").strip() + if not path: + return "" + try: + return Path(path).expanduser().read_text(encoding="utf-8").strip() + except OSError: + return "" + + +def classify_term( + term: str, + *, + jevgate: bool | None = None, + client: Any = None, + match_fn: Callable[..., Any] | None = None, +) -> dict[str, Any]: + """Classify one short term. Never raises for Jev failures.""" + text = str(term or "").strip() + enabled = jevgate_enabled(jevgate) + if not text: + return _blank(text, family="none", source="cascade", enabled=enabled) + + if match_fn is None: + from hyperlex.analysis import match_lineage + + match_fn = match_lineage + try: + hit = match_fn(text, use_vector=False) + except TypeError: + hit = match_fn(text) + if isinstance(hit, dict) and hit.get("family_id"): + return _blank( + text, + family=str(hit["family_id"]), + source="match_lineage", + enabled=enabled, + lineage=dict(hit), + ) + if not enabled: + return _blank(text, family="none", source="cascade", enabled=False) + + if not PROMPT_SHA_OK or not prompt_is_frozen(): + return _blank( + text, + family="none", + source="jevgate", + enabled=True, + jev_error={"flag": True, "reason": "prompt_sha_mismatch"}, + ) + + caller = client if client is not None else JevHttpClient() + state = {"text": text} + key = _current_key() + passes: list[dict[str, Any]] = [] + try: + for _ in range(PASSES): + passes.append(_one_pass(caller, state, key)) + try: + order = eight_families() + except JevCallError as exc: + return _blank( + text, + family="none", + source="jevgate", + enabled=True, + jev_error=_error_payload(exc, key), + ) + gate = decide_family(passes, order) + except JevCallError as exc: + return _blank( + text, + family="none", + source="jevgate", + enabled=True, + jev_error=_error_payload(exc, key), + ) + except Exception as exc: + return _blank( + text, + family="none", + source="jevgate", + enabled=True, + jev_error=_error_payload(JevCallError("provider_error", scrub_text(str(exc), key)), key), + ) + return _blank( + text, + family=str(gate["family"]), + source="jevgate", + enabled=True, + gate=gate, + ) diff --git a/src/hyperlex/cli.py b/src/hyperlex/cli.py index ff15a83a..c995583d 100644 --- a/src/hyperlex/cli.py +++ b/src/hyperlex/cli.py @@ -51,6 +51,18 @@ def cmd_check(_: argparse.Namespace) -> int: return 0 if ok else 2 +def cmd_classify(args: argparse.Namespace) -> int: + from hyperlex.analysis.jevgate import classify_term, jevgate_from_args + + term = (getattr(args, "term_pos", None) or getattr(args, "term", None) or "").strip() + if not term: + _emit({"ok": False, "command": "classify", "error": "empty term", "brier": None}) + return 2 + out = classify_term(term, jevgate=jevgate_from_args(args)) + _emit({"ok": True, "command": "classify", **out}) + return 0 + + def cmd_analyze(args: argparse.Namespace) -> int: from hyperlex import detect_memetic_patterns, extract_forecasts, emit_receipt, relay_from_result from hyperlex.intake.sources import pick_source @@ -127,6 +139,8 @@ def cmd_commands(_: argparse.Namespace) -> int: 'mutation trace "" --human', "mutation predict ", "mutation watch", + 'classify ""', + 'classify "" --jevgate', ], "routes": ["offline", "mock", "default", "live", "glossary", "social"], "docs": "docs/operator-loop.md · docs/commands.md", @@ -503,6 +517,24 @@ def build_parser() -> argparse.ArgumentParser: mp_alias.add_argument("--family", default="") mp_alias.set_defaults(func=cmd_mutation_predict) + cl = sub.add_parser( + "classify", + help="Classify a short term into 8 hyperlexical families or none", + ) + cl.add_argument("term_pos", nargs="?", default="", help="Short slang term") + cl.add_argument("--term", default="", help="Short slang term (flag form)") + cl.add_argument( + "--jevgate", + action="store_true", + help="Opt in to jevgate-1 after match_lineage misses (prereg v5). Default off.", + ) + cl.add_argument( + "--no-jevgate", + action="store_true", + help="Force the registry cascade even if HYPERLEX_JEVGATE is set", + ) + cl.set_defaults(func=cmd_classify) + a = sub.add_parser("analyze") a.add_argument("query_pos", nargs="?", default="") a.add_argument("--query", default="") diff --git a/tests/test_jevgate.py b/tests/test_jevgate.py new file mode 100644 index 00000000..777f985e --- /dev/null +++ b/tests/test_jevgate.py @@ -0,0 +1,364 @@ +"""jevgate-1 unit tests. Synthetic strings only. The Jev client is mocked.""" + +from __future__ import annotations + +import json +import os +import sys +import urllib.request +from pathlib import Path + +import pytest + +from hyperlex.analysis.jevgate import ( + FROZEN_PROMPT_SHA256, + MODEL_ID, + TAU, + JevCallError, + JevHttpClient, + assert_prompt_sha, + classify_term, + eight_families, + prompt_sha256, +) + +ROOT = Path(__file__).resolve().parents[1] +TERM = "zzsynthetic-jevgate-term" + + +class FakeJev: + def __init__(self, responses): + self.responses = list(responses) + self.calls = [] + + def evaluate(self, state): + self.calls.append(dict(state)) + if not self.responses: + raise JevCallError("provider_error", "no scripted response") + item = self.responses.pop(0) + if isinstance(item, Exception): + raise item + return item + + +def _miss(_term, **_kwargs): + return None + + +def _probs(none_p, peaks): + order = eight_families() + probs = {name: 0.0 for name in order} + probs.update(peaks) + probs["none"] = none_p + return probs + + +def _pass(none_p, peaks, *, model=MODEL_ID, choice="workplace-corp"): + return { + "model_id": model, + "answers": { + "in_scope": {"type": "boolean", "probability": 0.11}, + "any_slang": {"type": "boolean", "probability": 0.22}, + "family": { + "type": "choice", + "choice": choice, + "probabilities": _probs(none_p, peaks), + }, + }, + } + + +def _three(none_p, peaks, **kwargs): + return [_pass(none_p, peaks, **kwargs) for _ in range(3)] + + +def test_prompt_sha_assert(monkeypatch): + assert prompt_sha256() == FROZEN_PROMPT_SHA256 + assert_prompt_sha() + monkeypatch.setitem( + __import__("hyperlex.analysis.jevgate", fromlist=["QUESTIONS"]).QUESTIONS, + "in_scope", + {"type": "boolean", "instructions": "drift"}, + ) + # setitem may not replace the object identity used by prompt_sha256 if QUESTIONS + # is the same dict. Mutating in_scope changes JSON.stringify. + with pytest.raises(AssertionError): + assert_prompt_sha() + client = FakeJev([]) + out = classify_term(TERM, jevgate=True, client=client, match_fn=_miss) + assert out["family"] == "none" + assert out["jev_error"]["flag"] is True + assert out["jev_error"]["reason"] == "prompt_sha_mismatch" + assert client.calls == [] + + +def test_rule_first_precedence(): + seen = {} + + def match(term, use_vector=True): + seen["term"] = term + seen["use_vector"] = use_vector + return { + "family_id": "gaming-meta", + "matched_terms": ["zz"], + "confidence": 0.77, + "provenance": "INFERRED", + } + + client = FakeJev([AssertionError("jev must not run")]) + out = classify_term(TERM, jevgate=True, client=client, match_fn=match) + assert seen == {"term": TERM, "use_vector": False} + assert out["family"] == "gaming-meta" + assert out["source"] == "match_lineage" + assert out["jev_best_guess"] is None + assert out["jev_p_none"] is None + assert out["jev_error"] is None + assert client.calls == [] + assert out["brier"] is None + + +def test_flag_off_keeps_cascade(): + client = FakeJev([AssertionError("jev must not run")]) + out = classify_term(TERM, jevgate=False, client=client, match_fn=_miss) + assert out["family"] == "none" + assert out["source"] == "cascade" + assert out["jevgate_enabled"] is False + assert client.calls == [] + + +@pytest.mark.parametrize( + "none_p, expect_family", + [ + (0.30332999999999993, "ai-native"), # just below 0.30333 + (0.30333, "none"), # at tau, strict + (0.30333000000000004, "none"), # just above + ], +) +def test_tau_boundary(none_p, expect_family): + assert (0.30332999999999993 < TAU) and not (TAU < TAU) and not (0.30333000000000004 < TAU) + client = FakeJev(_three(none_p, {"ai-native": 0.62}, choice="workplace-corp")) + out = classify_term(TERM, jevgate=True, client=client, match_fn=_miss) + assert len(client.calls) == 3 + assert all(call == {"text": TERM} for call in client.calls) + assert out["family"] == expect_family + assert out["source"] == "jevgate" + assert out["jev_p_none"] == none_p + assert out["jev_p_none"] < TAU or expect_family == "none" + # choice is workplace-corp on every pass and must not decide + assert out["jev_log"]["choice"] == ["workplace-corp", "workplace-corp", "workplace-corp"] + assert out["jev_best_guess"] == "ai-native" + assert out["jev_family_means"]["ai-native"] == 0.62 + assert set(out["jev_family_means"]) == set(eight_families()) + assert "none" not in out["jev_family_means"] + assert out["jev_error"] is None + + +def test_tie_break_follows_layout_order(): + order = eight_families() + shadow = str(ROOT / "scripts" / "shadow") + if shadow not in sys.path: + sys.path.insert(0, shadow) + from hyperlexical.layout import FAMILIES + + assert order == tuple(name for name in FAMILIES if name != "none") + later = order[5] + earlier = order[2] + assert order.index(earlier) < order.index(later) + client = FakeJev(_three(0.1, {earlier: 0.4, later: 0.4})) + out = classify_term(TERM, jevgate=True, client=client, match_fn=_miss) + assert out["family"] == earlier + assert out["jev_best_guess"] == earlier + assert out["jev_family_means"][earlier] == out["jev_family_means"][later] + + +def test_best_guess_when_output_is_none(): + client = FakeJev(_three(0.9, {"crypto-degen": 0.05, "kinship-address": 0.01})) + out = classify_term(TERM, jevgate=True, client=client, match_fn=_miss) + assert out["family"] == "none" + assert out["jev_best_guess"] == "crypto-degen" + assert out["jev_p_none"] == 0.9 + assert out["jev_error"] is None + + +def test_best_guess_null_when_all_family_means_are_zero(): + client = FakeJev(_three(1.0, {})) + out = classify_term(TERM, jevgate=True, client=client, match_fn=_miss) + assert out["jev_p_none"] == 1.0 + assert out["jev_p_none"] >= TAU + assert all(value == 0.0 for value in out["jev_family_means"].values()) + assert out["family"] == "none" + assert out["jev_best_guess"] is None + + +def test_best_guess_null_when_all_family_means_are_equal(): + order = eight_families() + client = FakeJev(_three(0.52, {name: 0.06 for name in order})) + out = classify_term(TERM, jevgate=True, client=client, match_fn=_miss) + assert out["jev_p_none"] == 0.52 + assert out["jev_p_none"] >= TAU + assert len(set(out["jev_family_means"].values())) == 1 + assert out["jev_family_means"][order[0]] == 0.06 + assert out["family"] == "none" + assert out["jev_best_guess"] is None + + +def test_retry_then_fail_closed(monkeypatch): + secret = "synthetic-jev-key-not-real" + monkeypatch.setenv("JEV_API_KEY", secret) + client = FakeJev( + [ + JevCallError("provider_error", f"boom {secret} Bearer {secret}"), + JevCallError("timeout", f"still {secret}"), + ] + ) + out = classify_term(TERM, jevgate=True, client=client, match_fn=_miss) + assert len(client.calls) == 2 + assert out["family"] == "none" + assert out["source"] == "jevgate" + assert out["jev_error"]["flag"] is True + assert out["jev_error"]["reason"] == "timeout" + blob = json.dumps(out) + assert secret not in blob + assert "Bearer [R]" in blob or "[R]" in blob + + +def test_retry_then_success_uses_the_pass(): + order = eight_families() + winner = order[0] + client = FakeJev( + [ + JevCallError("provider_error", "transient"), + *_three(0.05, {winner: 0.7}), + ] + ) + out = classify_term(TERM, jevgate=True, client=client, match_fn=_miss) + assert len(client.calls) == 4 + assert out["family"] == winner + assert out["jev_error"] is None + + +def test_model_id_mismatch(monkeypatch): + monkeypatch.delenv("JEV_API_KEY", raising=False) + client = FakeJev( + [ + _pass(0.01, {"ai-native": 0.9}, model="jev-other"), + _pass(0.01, {"ai-native": 0.9}, model="jev-1.13.0-alias"), + ] + ) + out = classify_term(TERM, jevgate=True, client=client, match_fn=_miss) + assert len(client.calls) == 2 + assert out["family"] == "none" + assert out["jev_error"]["flag"] is True + assert out["jev_error"]["reason"] == "model_mismatch" + assert out["jev_best_guess"] is None + + +def test_http_client_wire_format(monkeypatch): + monkeypatch.setenv("JEV_API_KEY", "synthetic-jev-key-not-real") + monkeypatch.delenv("HYPERLEX_OFFLINE", raising=False) + monkeypatch.delenv("JEV_BASE_URL", raising=False) + captured = {} + order = eight_families() + probs = {name: 0.0 for name in order} + probs["ai-native"] = 0.7 + probs["none"] = 0.2 + + class _Resp: + def __enter__(self): + return self + + def __exit__(self, *_exc): + return False + + def read(self): + body = { + "model": MODEL_ID, + "answers": { + "in_scope": {"type": "noul", "noul": 0.15}, + "any_slang": {"type": "noul", "noul": 0.25}, + "family": { + "type": "choice", + "choice": "none", + "probabilities": probs, + }, + }, + } + return json.dumps(body).encode("utf-8") + + def fake_urlopen(req, timeout=0): + captured["url"] = req.full_url + captured["timeout"] = timeout + captured["body"] = json.loads(req.data.decode("utf-8")) + captured["auth"] = req.get_header("Authorization") + return _Resp() + + monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen) + out = JevHttpClient().evaluate({"text": TERM}) + assert captured["url"] == "https://api.typesafe.ai/v1/systemone" + assert captured["timeout"] == 55.0 + assert captured["body"]["model"] == MODEL_ID + assert captured["body"]["state"] == {"text": TERM} + assert captured["body"]["questions"]["in_scope"]["type"] == "noul" + assert captured["body"]["questions"]["any_slang"]["type"] == "noul" + assert captured["body"]["questions"]["family"]["type"] == "choice" + assert captured["auth"] == "Bearer synthetic-jev-key-not-real" + assert out["model_id"] == MODEL_ID + assert out["answers"]["in_scope"]["type"] == "boolean" + assert out["answers"]["in_scope"]["probability"] == 0.15 + assert "synthetic-jev-key-not-real" not in json.dumps( + {k: v for k, v in captured.items() if k != "auth"} + ) + + +def test_missing_key_does_not_call_network(monkeypatch): + monkeypatch.delenv("JEV_API_KEY", raising=False) + monkeypatch.delenv("TYPESAFE_AI_API_KEY", raising=False) + monkeypatch.delenv("JEV_KEY_FILE", raising=False) + monkeypatch.setenv("HYPERLEX_OFFLINE", "0") + + def boom(*_a, **_k): + raise AssertionError("network") + + monkeypatch.setattr(urllib.request, "urlopen", boom) + out = classify_term(TERM, jevgate=True, match_fn=_miss) + assert out["family"] == "none" + assert out["jev_error"]["flag"] is True + assert out["jev_error"]["reason"] == "missing_key" + + +def test_cli_classify_default_is_cascade(capsys, monkeypatch): + monkeypatch.delenv("HYPERLEX_JEVGATE", raising=False) + from hyperlex.cli import main + + rc = main(["classify", TERM]) + assert rc == 0 + data = json.loads(capsys.readouterr().out) + assert data["ok"] is True + assert data["command"] == "classify" + assert data["family"] == "none" + assert data["source"] == "cascade" + assert data["jevgate_enabled"] is False + assert data["brier"] is None + + +def test_skill_cli_classify_default_is_cascade(monkeypatch): + monkeypatch.delenv("HYPERLEX_JEVGATE", raising=False) + env = os.environ.copy() + env.pop("HYPERLEX_JEVGATE", None) + env["HYPERLEX_OFFLINE"] = "1" + env["PYTHONPATH"] = str(ROOT / "src") + import subprocess + + proc = subprocess.run( + [sys.executable, str(ROOT / "scripts" / "hyperlex.py"), "classify", TERM], + capture_output=True, + text=True, + cwd=ROOT, + env=env, + check=False, + ) + assert proc.returncode == 0, proc.stderr + data = json.loads(proc.stdout) + assert data["family"] == "none" + assert data["source"] == "cascade" + assert data["brier"] is None