diff --git a/.github/labeler.yml b/.github/labeler.yml index 47eda92a5..b00200745 100644 --- a/.github/labeler.yml +++ b/.github/labeler.yml @@ -120,6 +120,12 @@ contract:tracker: - 'tools/jira/**' - 'tools/sourcehut/**' +contract:typed-decision: + - any: + - changed-files: + - any-glob-to-any-file: + - 'tools/typed-decision/**' + substrate:action-guard: - any: - changed-files: diff --git a/docs/adapters/registry.md b/docs/adapters/registry.md index b85d37aa4..b30633a85 100644 --- a/docs/adapters/registry.md +++ b/docs/adapters/registry.md @@ -58,6 +58,7 @@ extension point = a documented, labelled slot with a tracking issue. | [`tools/chat`](../../tools/chat/) | [`chat-slack`](../../tools/chat-slack/) | Discord [#1421](https://github.com/apache/magpie/issues/1421) | | Agent harness | Claude Code, [Codex](codex.md) `experimental` ([#313](https://github.com/apache/magpie/issues/313)), [Gemini CLI](gemini.md) `experimental` ([#314](https://github.com/apache/magpie/issues/314)), [Local LLM (Ollama / llama.cpp / vLLM)](local-llm.md) ([#315](https://github.com/apache/magpie/issues/315)), [Cursor](cursor.md) ([#316](https://github.com/apache/magpie/issues/316)), [Goose](goose.md) `guide only` ([#319](https://github.com/apache/magpie/issues/319)), [Aider](aider.md) `guide only` ([#317](https://github.com/apache/magpie/issues/317)), [GitHub Copilot](copilot.md) `guide only` ([#318](https://github.com/apache/magpie/issues/318)), Grok `reviewer backend only` ([#1416](https://github.com/apache/magpie/issues/1416)) | Amazon Q [#320](https://github.com/apache/magpie/issues/320)–OpenHands [#322](https://github.com/apache/magpie/issues/322) | | Security cross-ref | [`tools/osv`](../../tools/osv/) | — | +| [`tools/typed-decision`](../../tools/typed-decision/) | TypeSafe Jev API | Local models / llama.cpp / Ollama ([#1431](https://github.com/apache/magpie/issues/1431)) | ## In-tree organizations diff --git a/docs/labels-and-capabilities.md b/docs/labels-and-capabilities.md index 6f5ff38c0..5c4854ebe 100644 --- a/docs/labels-and-capabilities.md +++ b/docs/labels-and-capabilities.md @@ -141,6 +141,7 @@ framework substrate: | `contract:scan-format` | contract | Security-scanner report parsing. | | `contract:project-metadata` | contract | Governance rosters / people / releases. | | `contract:security-cross-ref` | contract | Vulnerability database / cross-reference alias lookup (OSV.dev / NVD). | +| `contract:typed-decision` | contract | Provider-agnostic typed decision backend (Choice / Score / Noul). | | `substrate:analytics` | substrate | Read-only metrics / dashboards / renderers. | | `substrate:sandbox` | substrate | Agent isolation, egress control, settings audit. | | `substrate:action-guard` | substrate | Deterministic pre-tool-use command guards. | @@ -367,6 +368,7 @@ or a contract-free mix of substrates (e.g. `tools/spec-inventory` is | [`tools/skill-reconciler-diff`](../tools/skill-reconciler-diff/) | `substrate:framework-dev` | Deterministic structural diff between two skill trees — parses frontmatter, section headings, step inventory, placeholders, support files, and safety-baseline clauses into a JSON diff object for the `skill-reconciler` skill | | [`tools/vcs`](../tools/vcs/) | `contract:source-control` | Backend-dispatching implementation of the source-control (VCS) capability ([`tools/github/source-control.md`](../tools/github/source-control.md)); complete Git, Mercurial (Hg), and Fossil backends, plus detected extension point for SVN (#602) | | [`tools/sourcehut`](../tools/sourcehut/) | `contract:tracker` + `contract:source-control` + `contract:mail-archive` | SourceHut (sr.ht) forge bridge: todo.sr.ht, lists.sr.ht, builds.sr.ht, and git/hg repository reads | +| [`tools/typed-decision`](../tools/typed-decision/) | `contract:typed-decision` | Provider-agnostic typed decision tool contract (Choice / Score / Noul) with TypeSafe Jev API as initial backend | A tool's capability is the **interface it provides**, not which skills happen to consume it (RFC-AI-0005). `tools/github` provides the diff --git a/docs/vendor-neutrality.md b/docs/vendor-neutrality.md index cf90995a2..059524200 100644 --- a/docs/vendor-neutrality.md +++ b/docs/vendor-neutrality.md @@ -164,6 +164,7 @@ with pluggable backends already include: | [`tools/forwarder-relay`](../tools/forwarder-relay/) | ASF Security relay, huntr.com, HackerOne triagers | | [`tools/scan-format`](../tools/scan-format/) | security-scanner report formats (ASVS reference) | | [`tools/vcs`](../tools/vcs/) | Git (complete), Mercurial (complete), Subversion, … (extension points) | +| [`tools/typed-decision`](../tools/typed-decision/) | TypeSafe Jev API (cloud), local models via llama.cpp / Ollama (extension point) | The security-team surface follows the same pattern: CNA backends live behind [`tools/cve-tool`](../tools/cve-tool/) (the ASF Vulnogram adapter @@ -173,6 +174,11 @@ reference), inbound report relays behind behind [`tools/scan-format`](../tools/scan-format/), and an OSV.dev vulnerability cross-reference client [`tools/osv`](../tools/osv/) implements the `contract:security-cross-ref` capability. +Similarly, structured low-latency decisions sit behind the +`contract:typed-decision` contract ([`tools/typed-decision`](../tools/typed-decision/)), +with TypeSafe's Jev API as the initial reference backend and local +inference (via llama.cpp, Ollama, or vLLM) tracked as the second backend to +ensure cross-vendor choice. The distinction Magpie enforces: **vendor-specific *integrations* are expected and welcome; vendor-specific *workflows* are forbidden.** A @@ -209,6 +215,7 @@ contract for one vendor: | [`tools/forwarder-relay`](../tools/forwarder-relay/) | ASF-security ([`tools/gmail/asf-relay.md`](../tools/gmail/asf-relay.md)) | huntr.com, HackerOne | | [`tools/scan-format`](../tools/scan-format/) | ASVS | other scanner formats | | [`tools/vcs`](../tools/vcs/) | Git | Mercurial, Subversion, … | +| [`tools/typed-decision`](../tools/typed-decision/) | TypeSafe Jev API | Local models (llama.cpp, Ollama, vLLM) ([#1431](https://github.com/apache/magpie/issues/1431)) | A project selects an adapter per capability in its config (`cve_authority.tool: vulnogram`, `archive_system.kind: ponymail`, @@ -566,7 +573,7 @@ generated block below. -**Overall vendor-neutrality score: 10/12 capability contracts (83%).** Generated by [`tools/vendor-neutrality-score`](../tools/vendor-neutrality-score/); re-run it to refresh this section. +**Overall vendor-neutrality score: 10/13 capability contracts (77%).** Generated by [`tools/vendor-neutrality-score`](../tools/vendor-neutrality-score/); re-run it to refresh this section. | Capability contract | Neutral? | Class | Backends today | Basis | |---|---|---|---|---| @@ -582,6 +589,7 @@ generated block below. | `contract:scan-format` | ✅ | agnostic | — | vendor-neutral by construction — one spec serves every backend | | `contract:project-metadata` | ✅ | single-org | ASF | single-organisation capability (ASF); no vendor choice to make | | `contract:security-cross-ref` | ❌ | vendor-backed | OSV.dev | only 1 backend vendor (OSV.dev); needs 1 more | +| `contract:typed-decision` | ❌ | vendor-backed | TypeSafe | only 1 backend vendor (TypeSafe); needs 1 more | **Per-skill assessment: 78/78 skills carry no vendor lock-in.** A skill is *capability-pure* when it names no backend at all, *portable* when every backend it names has an alternative (its contract is green), and *vendor-coupled* only when it reaches for a backend that is the sole implementation of a capability. diff --git a/pyproject.toml b/pyproject.toml index beebf78c2..f9e4939b3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -109,6 +109,8 @@ package = false # here once so every member can name it in its `dev` group without repeating # the source; workspace members inherit the root's sources. magpie-dev = { workspace = true } +checker = { workspace = true } + [tool.uv.workspace] members = [ @@ -154,4 +156,5 @@ members = [ "tools/fossil", "tools/sourcehut", "tools/gitlab", + "tools/typed-decision", ] diff --git a/tools/privacy-llm/checker/src/checker/__init__.py b/tools/privacy-llm/checker/src/checker/__init__.py index c83a5dcc0..9402d96d8 100644 --- a/tools/privacy-llm/checker/src/checker/__init__.py +++ b/tools/privacy-llm/checker/src/checker/__init__.py @@ -27,4 +27,15 @@ Skills shell out to this command at Step 0 (pre-flight) when they may read ```` content. The contract is documented in ``tools/privacy-llm/wiring.md``. + +:func:`check_endpoint` is the runtime counterpart for a single +outbound endpoint, used by tools that make their own LLM calls. """ + +from checker.check import Verdict, check_endpoint, check_stack + +__all__ = [ + "Verdict", + "check_endpoint", + "check_stack", +] diff --git a/tools/privacy-llm/checker/src/checker/check.py b/tools/privacy-llm/checker/src/checker/check.py index 1b45e7e97..3e210d528 100644 --- a/tools/privacy-llm/checker/src/checker/check.py +++ b/tools/privacy-llm/checker/src/checker/check.py @@ -35,12 +35,15 @@ import argparse import dataclasses +import re import sys +import urllib.parse from checker.config import ( LLMEntry, OptInEntry, ParsedConfig, + _first_url, host_of, locate_config_path, parse_config, @@ -185,6 +188,216 @@ def check_stack(config: ParsedConfig) -> list[Verdict]: return out +_HOSTNAME_LABEL_RE = re.compile(r"^[a-zA-Z0-9]([a-zA-Z0-9-]*[a-zA-Z0-9])?$") + + +def _is_valid_hostname(token: str) -> bool: + """Validate that candidate token is a genuine hostname. + + Rejects tokens containing userinfo (@), fragments (#), path slashes (/), + version numbers (all-numeric dot-separated segments like 3.5, 4.0, 1.0.0), + TLDs without ASCII alphabetical characters, or invalid label syntax. + """ + if not token or len(token) > 253: + return False + if any(c in token for c in ("@", "#", "/", ":", "?")): + return False + labels = token.split(".") + if len(labels) < 2: + return False + # Reject tokens where all dot-separated segments are purely numeric + if all(lbl.isdigit() for lbl in labels): + return False + # Require at least one ASCII alphabetical character in top-level domain / rightmost label + if not any(c.isascii() and c.isalpha() for c in labels[-1]): + return False + # Require valid hostname label syntax for each label + for lbl in labels: + if not lbl or len(lbl) > 63: + return False + if not _HOSTNAME_LABEL_RE.match(lbl): + return False + return True + + +def _extract_opt_in_host(name: str) -> str | None: + """Extract destination hostname from an opt-in entry name. + + Recognises full URLs (https://api.example.com/v1) and bare hostnames + at the start of the entry (api.example.com (Provider)). Returns None + for name-only opt-ins (e.g. 'TypeSafe — Jev API', 'AWS Bedrock', + '3.5 Sonnet (AWS Bedrock)'). + """ + url = _first_url(name) + if url is not None: + return host_of(url) + + # Hyphens are not separators here: they are legal inside hostname labels + # (``bedrock-runtime.eu-central-1.amazonaws.com``). + first_token = re.split(r"[\s—,(:]+", name.strip())[0] + if _is_valid_hostname(first_token): + return host_of("//" + first_token) + return None + + +_URL_IN_TEXT_RE = re.compile(r"[a-zA-Z][a-zA-Z0-9+.-]*://\S+") + + +def _name_matches_description(opt_name: str, raw_desc: str) -> bool: + """Match a name-only opt-in against the provider description. + + URLs are stripped from the description first, and the opt-in name (or its + short name) must appear as whole words. Otherwise a short opt-in name such + as ``AI (internal)`` would match inside ``https://api.typesafe.ai/...`` and + approve a provider nobody opted in to. + """ + desc_lc = _URL_IN_TEXT_RE.sub(" ", raw_desc).strip().lower() + if not desc_lc: + return False + name_lc = opt_name.lower().strip() + for candidate in (name_lc, _shortname(name_lc)): + if candidate and re.search(rf"(? Verdict: + """Match endpoint against opt-in entries. + + For URL opt-in entries: requires exact host_of(endpoint) == host_of(opt_url). + For name-only opt-in entries: matches only if endpoint matches default_endpoint + and the provider name matches the opt-in entry name. + """ + for opt in opt_in: + opt_host = _extract_opt_in_host(opt.name) + matched = False + + if opt_host is not None: + if ep_host == opt_host: + matched = True + else: + # Name-only opt-in: matches only if endpoint is the default endpoint + if default_endpoint is not None: + def_host = host_of(default_endpoint) + if (entry.url == default_endpoint) or (def_host is not None and ep_host == def_host): + matched = _name_matches_description(opt.name, raw_desc or "") + + if not matched: + continue + + if not opt.data_residency: + return Verdict( + entry, + False, + f"opt-in entry {opt.name!r} matches but is missing the 'Data-residency contract' sub-bullet", + ) + if not opt.approved_by or _is_placeholder(opt.approved_by): + return Verdict( + entry, + False, + f"opt-in entry {opt.name!r} matches but the 'Approved-by' " + f"sub-bullet is missing or still has placeholder text " + f"({opt.approved_by!r}); a real PMC member must sign off.", + ) + return Verdict(entry, True, f"opt-in entry {opt.name!r} (data-residency + approved-by present)") + + return Verdict( + entry, + False, + f"third-party endpoint {entry.url} denied: no opt-in entry was declared " + f"with matching host in /privacy-llm.md", + ) + + +def check_endpoint( + endpoint: str, + config: ParsedConfig | None = None, + *, + default_endpoint: str | None = None, + raw_desc: str | None = None, +) -> Verdict: + """Verify whether a single outbound LLM endpoint URL is approved. + + Checks default-approval rules (e.g. localhost, *.apache.org), falling back + to host-bound opt-in entries in ``/privacy-llm.md``. + URLs containing userinfo or fragments are strictly rejected. The free-text + Claude Code rule does not apply to endpoint URL checks. + """ + desc = raw_desc if raw_desc is not None else endpoint + entry = LLMEntry(raw=desc, url=endpoint) + + # 1. URL validation: reject userinfo, fragments, or malformed URLs + try: + parsed = urllib.parse.urlsplit(endpoint) + except ValueError as err: + return Verdict(entry, False, f"unparsable URL {endpoint!r}: {err}") + + if not parsed.scheme or not parsed.netloc: + return Verdict(entry, False, f"invalid URL (missing scheme or host): {endpoint!r}") + + if parsed.scheme.lower() not in ("http", "https"): + return Verdict(entry, False, f"unsupported URL scheme {parsed.scheme!r} in {endpoint!r}") + + if parsed.fragment or "#" in endpoint: + return Verdict(entry, False, f"endpoint URL must not contain a fragment: {endpoint!r}") + + if parsed.username is not None or parsed.password is not None or "@" in parsed.netloc: + return Verdict(entry, False, f"endpoint URL must not contain userinfo: {endpoint!r}") + + ep_host = parsed.hostname + if ep_host is None: + return Verdict(entry, False, f"unparsable URL host in {endpoint!r}") + ep_host = ep_host.lower() + + # 2. Default rules for URL-only check (no Claude Code free-text rule!) + if ep_host in _LOCAL_HOSTS: + return Verdict(entry, True, f"local-only inference at {ep_host} (default-approved)") + if ep_host in _APACHE_ORG_CARVE_OUTS: + return Verdict( + entry, + False, + f"{ep_host} is carved out of the *.apache.org default approval " + f"({_APACHE_ORG_CARVE_OUTS[ep_host]}); declare it as an opt-in " + f"entry if the PMC accepts the residency terms", + ) + if ep_host.endswith(".apache.org") or ep_host == "apache.org": + return Verdict(entry, True, f"*.apache.org-hosted endpoint at {ep_host} (default-approved)") + + # 3. Locate and parse config if not passed + if config is None: + try: + path = locate_config_path() + except FileNotFoundError as err: + return Verdict( + entry, + False, + f"Third-party endpoint {endpoint} denied (no privacy-llm config found): {err}", + ) + try: + config = parse_config(path) + except Exception as err: + return Verdict( + entry, + False, + f"Failed to parse privacy-llm config at {path}: {err}", + ) + + return _approve_endpoint_by_opt_in( + entry, + ep_host, + config.opt_in, + default_endpoint=default_endpoint, + raw_desc=raw_desc, + ) + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="privacy-llm-check", diff --git a/tools/privacy-llm/checker/src/checker/py.typed b/tools/privacy-llm/checker/src/checker/py.typed new file mode 100644 index 000000000..1242d4327 --- /dev/null +++ b/tools/privacy-llm/checker/src/checker/py.typed @@ -0,0 +1 @@ +# Marker file for PEP 561. diff --git a/tools/privacy-llm/checker/tests/test_check.py b/tools/privacy-llm/checker/tests/test_check.py index 7f23b387b..0c9c5ad76 100644 --- a/tools/privacy-llm/checker/tests/test_check.py +++ b/tools/privacy-llm/checker/tests/test_check.py @@ -27,7 +27,7 @@ def _write(path: pathlib.Path, body: str) -> pathlib.Path: - path.write_text(textwrap.dedent(body)) + path.write_text(textwrap.dedent(body), encoding="utf-8") return path @@ -312,3 +312,254 @@ def test_framework_template_is_approved_by_default(tmp_path: pathlib.Path, monke assert verdicts, "template should declare at least Claude Code in its stack" bad = [v for v in verdicts if not v.approved] assert not bad, f"unapproved entries in shipped template: {bad}" + + +# -- check_endpoint helper tests --------------------------------------- + + +def test_check_endpoint_default_approved(): + v = check.check_endpoint("http://localhost:8000/v1") + assert v.approved is True + assert "local-only" in v.reason + + +def test_check_endpoint_denied_no_config(monkeypatch, tmp_path): + monkeypatch.delenv("PRIVACY_LLM_CONFIG", raising=False) + monkeypatch.chdir(tmp_path) + v = check.check_endpoint("https://api.example.com/v1") + assert v.approved is False + assert "denied" in v.reason + + +def test_check_endpoint_approved_opt_in(): + opt = OptInEntry( + name="Example Provider", + data_residency="Strict US", + approved_by="PMC 2026-09-01", + ) + cfg = ParsedConfig(path=pathlib.Path("/dev/null"), llm_stack=[], opt_in=[opt]) + v = check.check_endpoint( + "https://api.example.com/v1", + config=cfg, + default_endpoint="https://api.example.com/v1", + raw_desc="Example Provider (https://api.example.com/v1)", + ) + assert v.approved is True + + +def test_check_endpoint_rejects_url_fragment_and_claude_code(): + """A URL containing a fragment is rejected and cannot trigger the Claude Code rule.""" + v = check.check_endpoint("https://evil.example.com/v1#claude code") + assert v.approved is False + assert "fragment" in v.reason + + +def test_check_endpoint_rejects_subdomain_opt_in_bypass(): + """An opt-in for api.typesafe.ai does not approve api.typesafe.ai.evil.example.""" + opt = OptInEntry( + name="https://api.typesafe.ai", + data_residency="eu-central-1", + approved_by="PMC 2026-09-01", + ) + cfg = ParsedConfig(path=pathlib.Path("/dev/null"), llm_stack=[], opt_in=[opt]) + v = check.check_endpoint("https://api.typesafe.ai.evil.example/v1", config=cfg) + assert v.approved is False + assert "denied" in v.reason + + +def test_check_endpoint_rejects_userinfo_opt_in_bypass(): + """A URL containing userinfo (e.g. user@host) is strictly rejected.""" + opt = OptInEntry( + name="https://api.typesafe.ai", + data_residency="eu-central-1", + approved_by="PMC 2026-09-01", + ) + cfg = ParsedConfig(path=pathlib.Path("/dev/null"), llm_stack=[], opt_in=[opt]) + v = check.check_endpoint("https://api.typesafe.ai@evil.example/v1", config=cfg) + assert v.approved is False + assert "userinfo" in v.reason + + +def test_check_endpoint_name_only_opt_in_rejects_different_host(): + """A name-only opt-in like 'TypeSafe — Jev API' never approves a non-default host.""" + opt = OptInEntry( + name="TypeSafe — Jev API", + data_residency="eu-central-1", + approved_by="PMC 2026-09-01", + ) + cfg = ParsedConfig(path=pathlib.Path("/dev/null"), llm_stack=[], opt_in=[opt]) + v = check.check_endpoint( + "https://typesafe.evil.example/v1", + config=cfg, + default_endpoint="https://api.typesafe.ai/v1/systemone", + ) + assert v.approved is False + assert "denied" in v.reason + + +def test_check_endpoint_approved_opt_in_url(): + """An opt-in with a full URL approves endpoints sharing the exact same host.""" + opt = OptInEntry( + name="https://api.typesafe.ai", + data_residency="eu-central-1", + approved_by="PMC 2026-09-01", + ) + cfg = ParsedConfig(path=pathlib.Path("/dev/null"), llm_stack=[], opt_in=[opt]) + v = check.check_endpoint("https://api.typesafe.ai/v1/systemone", config=cfg) + assert v.approved is True + + +def test_check_endpoint_name_only_opt_in_rejects_empty_raw_desc(): + """Configure opt-in - AWS Bedrock — eu-central-1. Call check_endpoint without raw_desc. Assert approved is False.""" + opt = OptInEntry( + name="AWS Bedrock — eu-central-1", + data_residency="eu-central-1", + approved_by="PMC 2026-09-01", + ) + cfg = ParsedConfig(path=pathlib.Path("/dev/null"), llm_stack=[], opt_in=[opt]) + + # No raw_desc (raw_desc=None) + v1 = check.check_endpoint( + "https://api.typesafe.ai/v1/systemone", + config=cfg, + default_endpoint="https://api.typesafe.ai/v1/systemone", + raw_desc=None, + ) + assert v1.approved is False + + # Empty string raw_desc + v2 = check.check_endpoint( + "https://api.typesafe.ai/v1/systemone", + config=cfg, + default_endpoint="https://api.typesafe.ai/v1/systemone", + raw_desc="", + ) + assert v2.approved is False + + # Whitespace raw_desc + v3 = check.check_endpoint( + "https://api.typesafe.ai/v1/systemone", + config=cfg, + default_endpoint="https://api.typesafe.ai/v1/systemone", + raw_desc=" ", + ) + assert v3.approved is False + + +def test_name_only_opt_in_requires_provider_in_desc(): + """Name-only opt-in matches only if the opt-in name appears in raw_desc.""" + opt_typesafe = OptInEntry( + name="TypeSafe — Jev API", + data_residency="eu-central-1", + approved_by="PMC 2026-09-01", + ) + cfg_typesafe = ParsedConfig(path=pathlib.Path("/dev/null"), llm_stack=[], opt_in=[opt_typesafe]) + + v_match = check.check_endpoint( + "https://api.typesafe.ai/v1/systemone", + config=cfg_typesafe, + default_endpoint="https://api.typesafe.ai/v1/systemone", + raw_desc="TypeSafe Jev (https://api.typesafe.ai/v1/systemone)", + ) + assert v_match.approved is True + + # If opt-in is AWS Bedrock, calling default endpoint with TypeSafe desc is rejected + opt_aws = OptInEntry( + name="AWS Bedrock — eu-central-1", + data_residency="eu-central-1", + approved_by="PMC 2026-09-01", + ) + cfg_aws = ParsedConfig(path=pathlib.Path("/dev/null"), llm_stack=[], opt_in=[opt_aws]) + + v_mismatch = check.check_endpoint( + "https://api.typesafe.ai/v1/systemone", + config=cfg_aws, + default_endpoint="https://api.typesafe.ai/v1/systemone", + raw_desc="TypeSafe Jev (https://api.typesafe.ai/v1/systemone)", + ) + assert v_mismatch.approved is False + assert "denied" in v_mismatch.reason + + +def test_name_only_opt_in_does_not_match_inside_endpoint_url(): + """A short opt-in name must not match a fragment of the endpoint URL in raw_desc.""" + for name in ("AI (internal)", "API — internal gateway", "v1 (internal)", "Type (internal)"): + opt = OptInEntry(name=name, data_residency="on-prem", approved_by="PMC 2026-09-01") + cfg = ParsedConfig(path=pathlib.Path("/dev/null"), llm_stack=[], opt_in=[opt]) + v = check.check_endpoint( + "https://api.typesafe.ai/v1/systemone", + config=cfg, + default_endpoint="https://api.typesafe.ai/v1/systemone", + raw_desc="TypeSafe Jev (https://api.typesafe.ai/v1/systemone)", + ) + assert v.approved is False, name + + +def test_extract_opt_in_host_keeps_hyphenated_bare_host(): + """A bare hostname containing hyphens is extracted whole, not cut at the first '-'.""" + assert ( + check._extract_opt_in_host("bedrock-runtime.eu-central-1.amazonaws.com (AWS Bedrock)") + == "bedrock-runtime.eu-central-1.amazonaws.com" + ) + assert check._extract_opt_in_host("my-llm.example.org") == "my-llm.example.org" + # Name-only entries with a spaced dash separator still resolve to no host. + assert check._extract_opt_in_host("TypeSafe - Jev API") is None + + +def test_extract_opt_in_host_rejects_model_version_tokens(): + """Model version tokens (3.5, 4.0, 1.0.0) are not treated as hostnames.""" + assert check._extract_opt_in_host("3.5 Sonnet (AWS Bedrock)") is None + assert check._extract_opt_in_host("4.0 Omni") is None + assert check._extract_opt_in_host("1.0.0 Model") is None + assert check._extract_opt_in_host("api.example.com (Provider)") == "api.example.com" + assert check._extract_opt_in_host("https://api.example.com/v1") == "api.example.com" + + +def test_check_endpoint_rejects_numeric_version_token_as_host(): + """Configure opt-in - 3.5 Sonnet (AWS Bedrock). Call check_endpoint('https://3.5/v1'). Assert approved is False.""" + opt = OptInEntry( + name="3.5 Sonnet (AWS Bedrock)", + data_residency="eu-central-1", + approved_by="PMC 2026-09-01", + ) + cfg = ParsedConfig(path=pathlib.Path("/dev/null"), llm_stack=[], opt_in=[opt]) + + v = check.check_endpoint("https://3.5/v1", config=cfg) + assert v.approved is False + assert "denied" in v.reason + + +def test_check_endpoint_bare_domain_opt_in_extracts_and_approves(): + """Bare valid domain in opt-in entries (e.g. - api.example.com (Provider)) extracts correctly and approves.""" + opt = OptInEntry( + name="api.example.com (Provider)", + data_residency="eu-central-1", + approved_by="PMC 2026-09-01", + ) + cfg = ParsedConfig(path=pathlib.Path("/dev/null"), llm_stack=[], opt_in=[opt]) + v = check.check_endpoint("https://api.example.com/v1", config=cfg) + assert v.approved is True + + +def test_is_valid_hostname_rules(): + """_is_valid_hostname enforces standard hostname syntax and rejects invalid tokens.""" + assert check._is_valid_hostname("api.example.com") is True + assert check._is_valid_hostname("bedrock-runtime.eu-central-1.amazonaws.com") is True + assert check._is_valid_hostname("typesafe.ai") is True + + # Purely numeric or version tokens + assert check._is_valid_hostname("3.5") is False + assert check._is_valid_hostname("4.0") is False + assert check._is_valid_hostname("1.0.0") is False + assert check._is_valid_hostname("192.168.1.1") is False + + # Label syntax violations + assert check._is_valid_hostname("-bad.example.com") is False + assert check._is_valid_hostname("bad-.example.com") is False + assert check._is_valid_hostname("example..com") is False + assert check._is_valid_hostname("singlelabel") is False + + # Userinfo, fragments, slashes + assert check._is_valid_hostname("user@example.com") is False + assert check._is_valid_hostname("example.com/v1") is False + assert check._is_valid_hostname("example.com#frag") is False diff --git a/tools/privacy-llm/checker/tests/test_config.py b/tools/privacy-llm/checker/tests/test_config.py index 017945cc9..a90897357 100644 --- a/tools/privacy-llm/checker/tests/test_config.py +++ b/tools/privacy-llm/checker/tests/test_config.py @@ -33,7 +33,7 @@ def _write(path: pathlib.Path, body: str) -> pathlib.Path: - path.write_text(textwrap.dedent(body)) + path.write_text(textwrap.dedent(body), encoding="utf-8") return path diff --git a/tools/skill-and-tool-validator/src/skill_and_tool_validator/__init__.py b/tools/skill-and-tool-validator/src/skill_and_tool_validator/__init__.py index e083ca654..6fe7bd2f3 100644 --- a/tools/skill-and-tool-validator/src/skill_and_tool_validator/__init__.py +++ b/tools/skill-and-tool-validator/src/skill_and_tool_validator/__init__.py @@ -390,6 +390,7 @@ "contract:scan-format", "contract:project-metadata", "contract:security-cross-ref", + "contract:typed-decision", "substrate:analytics", "substrate:sandbox", "substrate:action-guard", diff --git a/tools/typed-decision/README.md b/tools/typed-decision/README.md new file mode 100644 index 000000000..d8d56ad46 --- /dev/null +++ b/tools/typed-decision/README.md @@ -0,0 +1,66 @@ + + + + +**Table of Contents** *generated with [DocToc](https://github.com/thlorenz/doctoc)* + +- [`tools/typed-decision/`](#toolstyped-decision) + - [Prerequisites](#prerequisites) + - [Operations](#operations) + - [Configuration](#configuration) + - [Fail-Open Contract](#fail-open-contract) + + + +# `tools/typed-decision/` + +**Capability:** contract:typed-decision + +**Kind:** implementation + +**Vendor:** TypeSafe + +**Harness:** agnostic + +Provider-agnostic typed decision adapter contract (`choice`, `score`, `noul`) with TypeSafe's Jev API (`api.typesafe.ai/v1/systemone`) as the initial reference backend. +See [`tool.md`](tool.md) for full contract specifications and Python API usage. + +## Prerequisites + +- **Runtime:** Python 3.11+ via `uv`. + The implementation is stdlib-only (`urllib.request`), introducing zero third-party dependencies. +- **CLIs:** None. +- **Credentials / auth:** `TYPESAFE_API_KEY` (or `JEV_API_KEY`) environment variable or home-directory key at `~/.config/apache-magpie/typesafe.key`. + Outbound prompts are strictly gated through [`tools/privacy-llm/`](../privacy-llm/) and deny unapproved destinations by default. +- **Network:** `api.typesafe.ai` over HTTPS. +- **Secure agent setup:** under [`docs/setup/secure-agent-setup.md`](../../docs/setup/secure-agent-setup.md) the sandbox cannot read `~/`, so the key file is not found and every call resolves to `TypedDecisionUnavailable`. + After the privacy-llm opt-in is signed off, an adopter enables the provider by passing the key through an environment variable the clean-env wrapper forwards, and by adding `api.typesafe.ai` to their own `sandbox.network.allowedDomains`. + The framework's default allowlist does not include it, by design. + +## Operations + +The contract defines three operations (see [`tool.md`](tool.md) for parameter details): + +1. **`choice(prompt, options: list[str]) -> {label, confidence}`** — Discrete option selection. +2. **`score(prompt, scale) -> {value, confidence}`** — Bounded numeric scoring along a range. +3. **`noul(prompt) -> {probability}`** — Semantic null/binary decision probability. + +## Configuration + +Adopters configure the provider and credentials via environment variables: + +| Variable | Description | Default | +|---|---|---| +| `MAGPIE_TYPED_DECISION_PROVIDER` | Selected decision backend (`jev`) | `jev` if credentials configured, else reports unavailable | +| `TYPESAFE_API_KEY` | API key for TypeSafe Jev provider | Unset (can also use `JEV_API_KEY` or `~/.config/apache-magpie/typesafe.key`) | + +Third-party destinations such as `api.typesafe.ai` deny by default per [`tools/privacy-llm/models.md`](../privacy-llm/models.md). +They require an explicit opt-in entry in `/privacy-llm.md` with filled `Data-residency contract` and `Approved-by` sign-offs. + +## Fail-Open Contract + +In adherence with RFC-AI-0004: +- On missing credentials, timeout, network failure, or provider error, every operation raises `TypedDecisionUnavailable`. +- The tool **never fabricates an answer** or returns synthetic defaults. +- Callers must catch `TypedDecisionUnavailable` and fall back to maintainer confirmation or primary LLM reasoning. diff --git a/tools/typed-decision/pyproject.toml b/tools/typed-decision/pyproject.toml new file mode 100644 index 000000000..a0052b37b --- /dev/null +++ b/tools/typed-decision/pyproject.toml @@ -0,0 +1,85 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "typed-decision" +version = "0.1.0" +description = "Provider-agnostic typed decision tool contract (Choice, Score, Noul) for Apache Magpie with TypeSafe Jev API as first backend." +readme = "README.md" +requires-python = ">=3.11" +license = { text = "Apache-2.0" } +dependencies = [ + "checker", +] + +[tool.uv.sources] +checker = { workspace = true } + +[dependency-groups] +dev = ["magpie-dev"] + +[tool.hatch.build.targets.wheel] +packages = ["src/typed_decision"] + +[tool.ruff] +line-length = 110 +target-version = "py311" +src = ["src", "tests"] + +[tool.ruff.lint] +select = [ + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # pyflakes + "I", # isort + "B", # flake8-bugbear + "UP", # pyupgrade + "SIM", # flake8-simplify + "C4", # flake8-comprehensions + "RUF", # ruff-specific +] +ignore = [ + "E501", # line-too-long +] + +[tool.ruff.lint.per-file-ignores] +"tests/**" = ["B", "SIM"] + +[tool.mypy] +python_version = "3.11" +files = ["src", "tests"] +warn_unused_ignores = true +warn_redundant_casts = true +warn_unreachable = true +check_untyped_defs = true +no_implicit_optional = true +disallow_untyped_defs = true +disallow_incomplete_defs = true + +[[tool.mypy.overrides]] +module = "tests.*" +disallow_untyped_defs = false +disallow_incomplete_defs = false + +[tool.pytest.ini_options] +minversion = "8.0" +addopts = "-ra -q" +testpaths = ["tests"] diff --git a/tools/typed-decision/src/typed_decision/__init__.py b/tools/typed-decision/src/typed_decision/__init__.py new file mode 100644 index 000000000..614cd26c2 --- /dev/null +++ b/tools/typed-decision/src/typed_decision/__init__.py @@ -0,0 +1,109 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Provider-agnostic typed decision tool contract (Choice, Score, Noul).""" + +from __future__ import annotations + +from typing import Any + +from typed_decision.exceptions import TypedDecisionUnavailable +from typed_decision.interface import DecisionProvider +from typed_decision.privacy import enforce_privacy_gate +from typed_decision.providers.jev import JevProvider +from typed_decision.registry import get_provider, register_provider + +__all__ = [ + "DecisionProvider", + "JevProvider", + "TypedDecisionClient", + "TypedDecisionUnavailable", + "choice", + "enforce_privacy_gate", + "get_provider", + "noul", + "register_provider", + "score", +] + + +def choice( + prompt: str, + options: list[str], + *, + provider: DecisionProvider | None = None, +) -> dict[str, Any]: + """Select one discrete option label from candidates based on prompt. + + Returns: + {"label": str, "confidence": float} + """ + p = provider or get_provider() + return p.choice(prompt, options) + + +def score( + prompt: str, + scale: tuple[float, float] | list[float] | int | float, + *, + provider: DecisionProvider | None = None, +) -> dict[str, Any]: + """Score prompt along the specified scale. + + Returns: + {"value": float | int, "confidence": float} + """ + p = provider or get_provider() + return p.score(prompt, scale) + + +def noul( + prompt: str, + *, + provider: DecisionProvider | None = None, +) -> dict[str, Any]: + """Evaluate null/binary decision probability for prompt. + + Returns: + {"probability": float} + """ + p = provider or get_provider() + return p.noul(prompt) + + +class TypedDecisionClient: + """Client wrapper for convenient object-oriented typed decision calls.""" + + def __init__(self, provider: DecisionProvider | None = None, **kwargs: Any) -> None: + self._provider = provider or get_provider(**kwargs) + + @property + def provider(self) -> DecisionProvider: + return self._provider + + def choice(self, prompt: str, options: list[str]) -> dict[str, Any]: + return self._provider.choice(prompt, options) + + def score( + self, + prompt: str, + scale: tuple[float, float] | list[float] | int | float, + ) -> dict[str, Any]: + return self._provider.score(prompt, scale) + + def noul(self, prompt: str) -> dict[str, Any]: + return self._provider.noul(prompt) diff --git a/tools/typed-decision/src/typed_decision/exceptions.py b/tools/typed-decision/src/typed_decision/exceptions.py new file mode 100644 index 000000000..a837cdb55 --- /dev/null +++ b/tools/typed-decision/src/typed_decision/exceptions.py @@ -0,0 +1,30 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Exceptions for typed decision operations.""" + +from __future__ import annotations + + +class TypedDecisionUnavailable(Exception): + """Raised when a typed decision backend is unavailable or fails. + + Under the fail-open contract, on missing configuration, timeout, + network error, provider error, or privacy gate rejection, every + operation raises this exception — never a fabricated answer — + so callers can catch it and fall back to their own reasoning. + """ diff --git a/tools/typed-decision/src/typed_decision/interface.py b/tools/typed-decision/src/typed_decision/interface.py new file mode 100644 index 000000000..f1fdb181a --- /dev/null +++ b/tools/typed-decision/src/typed_decision/interface.py @@ -0,0 +1,85 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Abstract provider interface for typed decisions.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import Any + + +class DecisionProvider(ABC): + """Abstract interface that any typed-decision provider backend implements.""" + + @property + @abstractmethod + def name(self) -> str: + """Provider name identifier (e.g. 'jev').""" + + @abstractmethod + def choice(self, prompt: str, options: list[str]) -> dict[str, Any]: + """Select a single discrete label from given options based on prompt. + + Args: + prompt: Question or context for decision. + options: Non-empty list of candidate labels. + + Returns: + dict containing: + "label": str (selected option), + "confidence": float (between 0.0 and 1.0). + + Raises: + TypedDecisionUnavailable: On provider/network failure, timeout, or missing config. + """ + + @abstractmethod + def score( + self, + prompt: str, + scale: tuple[float, float] | list[float] | int | float, + ) -> dict[str, Any]: + """Evaluate a score for the prompt along the specified scale. + + Args: + prompt: Text to be scored. + scale: Range or upper bound (e.g. (1, 5), (0, 10), or numeric max). + + Returns: + dict containing: + "value": float | int (assigned score), + "confidence": float (between 0.0 and 1.0). + + Raises: + TypedDecisionUnavailable: On provider/network failure, timeout, or missing config. + """ + + @abstractmethod + def noul(self, prompt: str) -> dict[str, Any]: + """Compute the null / binary hypothesis decision probability. + + Args: + prompt: Proposition or prompt to evaluate. + + Returns: + dict containing: + "probability": float (between 0.0 and 1.0). + + Raises: + TypedDecisionUnavailable: On provider/network failure, timeout, or missing config. + """ diff --git a/tools/typed-decision/src/typed_decision/privacy.py b/tools/typed-decision/src/typed_decision/privacy.py new file mode 100644 index 000000000..9a17e027a --- /dev/null +++ b/tools/typed-decision/src/typed_decision/privacy.py @@ -0,0 +1,76 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Privacy-LLM gate integration for typed decision providers. + +In accordance with RFC-AI-0004 Principle 6 (Privacy by design) and +tools/privacy-llm/models.md, any outbound prompt sent to an external +model provider must pass through this gate before leaving the process. +The gate strictly denies unapproved endpoints by default. +""" + +from __future__ import annotations + +from checker.check import check_endpoint + +from typed_decision.exceptions import TypedDecisionUnavailable + +DEFAULT_ENDPOINT: str = "https://api.typesafe.ai/v1/systemone" + + +def _check_endpoint_approved(endpoint: str, provider_name: str | None = None) -> tuple[bool, str]: + """Check if the given endpoint is approved per tools/privacy-llm/models.md. + + Denies by default per tools/privacy-llm/models.md ('Anything else -> ✗'). + Binds opt-in checks to the specific destination host: rejects URLs with userinfo + or fragments, requires exact host matching for URL opt-ins, and restricts + name-only opt-ins strictly to DEFAULT_ENDPOINT. + """ + raw_desc = f"{provider_name} ({endpoint})" if provider_name else endpoint + verdict = check_endpoint(endpoint, default_endpoint=DEFAULT_ENDPOINT, raw_desc=raw_desc) + return verdict.approved, verdict.reason + + +def enforce_privacy_gate(prompt: str, endpoint: str, provider_name: str | None = None) -> str: + """Validate that the outbound prompt is destined for an approved endpoint. + + In accordance with RFC-AI-0004 and tools/privacy-llm/models.md, any outbound + prompt must never reach an unapproved endpoint. The gate is conservative and + denies by default: third-party endpoints must be explicitly approved by PMC + members in `/privacy-llm.md` with a non-empty Data-residency + contract and non-placeholder Approved-by line. + + Per tools/privacy-llm/wiring.md, skills processing private source content must + redact PII (using `tools/privacy-llm/redactor`) prior to invoking downstream + tools. This gate strictly guards the network egress boundary. + + Args: + prompt: Raw or pre-redacted prompt text intended for the external provider. + endpoint: Destination API endpoint URL. + provider_name: Optional provider name (e.g. 'TypeSafe Jev') for opt-in matching. + + Returns: + The validated prompt text. + + Raises: + TypedDecisionUnavailable: If the destination endpoint is unapproved. + """ + approved, reason = _check_endpoint_approved(endpoint, provider_name=provider_name) + if not approved: + raise TypedDecisionUnavailable(f"Privacy-LLM gate rejected outbound request to {endpoint}: {reason}") + + return prompt diff --git a/tools/typed-decision/src/typed_decision/providers/__init__.py b/tools/typed-decision/src/typed_decision/providers/__init__.py new file mode 100644 index 000000000..0ec4af08f --- /dev/null +++ b/tools/typed-decision/src/typed_decision/providers/__init__.py @@ -0,0 +1,22 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Typed decision backend providers.""" + +from typed_decision.providers.jev import JevProvider + +__all__ = ["JevProvider"] diff --git a/tools/typed-decision/src/typed_decision/providers/jev.py b/tools/typed-decision/src/typed_decision/providers/jev.py new file mode 100644 index 000000000..b08575496 --- /dev/null +++ b/tools/typed-decision/src/typed_decision/providers/jev.py @@ -0,0 +1,481 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""TypeSafe Jev (SystemOne) typed-decision provider implementation. + +Implements the DecisionProvider contract against api.typesafe.ai/v1/systemone +using the Python standard library (urllib.request) with zero external dependencies. +""" + +from __future__ import annotations + +import json +import math +import os +import pathlib +import socket +import time +import urllib.error +import urllib.parse +import urllib.request +from typing import Any + +from typed_decision.exceptions import TypedDecisionUnavailable +from typed_decision.interface import DecisionProvider +from typed_decision.privacy import DEFAULT_ENDPOINT, enforce_privacy_gate + +# The model version is strictly pinned to an immutable release snapshot, never "latest" +# (per RFC-AI-0004 § Principle 3: Vendor neutrality & reproducible evaluations). +JEV_MODEL: str = "systemone-2026-06-01" +DEFAULT_TIMEOUT_SECONDS: float = 30.0 +DEFAULT_BACKOFF_SECONDS: float = 0.5 + + +class NoAuthRedirectHandler(urllib.request.HTTPRedirectHandler): + """Reject redirects so Authorization credentials are not forwarded to other origins.""" + + def redirect_request( + self, + req: urllib.request.Request, + fp: Any, + code: int, + msg: str, + headers: Any, + newurl: str, + ) -> urllib.request.Request | None: + raise TypedDecisionUnavailable(f"Request redirected to {newurl}; refusing to forward credentials") + + +def _require_https(url: str) -> None: + """Ensure the URL uses HTTPS (permitting HTTP only for localhost in tests).""" + parsed = urllib.parse.urlparse(url) + if parsed.scheme == "https": + return + if parsed.scheme == "http" and parsed.hostname in {"localhost", "127.0.0.1", "::1"}: + return + raise TypedDecisionUnavailable(f"Insecure endpoint URL: '{url}' must use HTTPS") + + +def _read_http_error(exc: urllib.error.HTTPError) -> str: + """Extract a descriptive message from an HTTP error response.""" + try: + body = exc.read().decode("utf-8") + parsed = json.loads(body) + if isinstance(parsed, dict): + for key in ("error", "message", "detail"): + val = parsed.get(key) + if val: + return str(val) + return body.strip() or str(exc) + except Exception: + return str(exc.reason or exc) + + +def _is_timeout(exc: urllib.error.URLError) -> bool: + """Check if URLError was caused by a connection or read timeout.""" + if isinstance(exc.reason, (socket.timeout, TimeoutError)): + return True + reason_str = str(exc.reason).lower() + return "timed out" in reason_str or "timeout" in reason_str + + +def _resolve_api_key(explicit_key: str | None = None) -> str | None: + """Resolve API key from explicit arg, environment variables, or user home config.""" + if explicit_key: + return explicit_key.strip() + + # Adopter environment variables + for var_name in ("TYPESAFE_API_KEY", "JEV_API_KEY"): + val = os.environ.get(var_name) + if val and val.strip(): + return val.strip() + + # Home directory config fallback (~/.config/apache-magpie/typesafe.key) + home = os.environ.get("HOME") or os.environ.get("USERPROFILE") + if home: + key_paths = ( + pathlib.Path(home) / ".config" / "apache-magpie" / "typesafe.key", + pathlib.Path(home) / ".config" / "apache-magpie" / "jev.key", + ) + for kp in key_paths: + if kp.is_file(): + try: + content = kp.read_text(encoding="utf-8").strip() + if content: + return content + except OSError: + # Ignore unreadable or inaccessible key files and try next candidate path. + pass + + return None + + +def _extract_result_dict(resp: dict[str, Any]) -> dict[str, Any]: + """Extract inner result dict from Jev API response envelope if present. + + Validates that if 'result' or 'decision' key exists, its value must be a dict. + Fails closed if the key is present but contains a non-dict value (e.g. error payload). + """ + if "result" in resp: + res = resp["result"] + if not isinstance(res, dict): + raise TypedDecisionUnavailable( + f"Malformed 'result' in Jev API response: expected JSON object, got {type(res).__name__}" + ) + return res + if "decision" in resp: + dec = resp["decision"] + if not isinstance(dec, dict): + raise TypedDecisionUnavailable( + f"Malformed 'decision' in Jev API response: expected JSON object, got {type(dec).__name__}" + ) + return dec + return resp + + +class JevProvider(DecisionProvider): + """TypeSafe Jev decision provider calling api.typesafe.ai/v1/systemone.""" + + def __init__( + self, + api_key: str | None = None, + endpoint: str = DEFAULT_ENDPOINT, + model: str = JEV_MODEL, + timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + backoff_seconds: float = DEFAULT_BACKOFF_SECONDS, + ) -> None: + self._endpoint = endpoint + self._model = model + self._timeout_seconds = timeout_seconds + self._backoff_seconds = backoff_seconds + + resolved_key = _resolve_api_key(api_key) + if not resolved_key: + raise TypedDecisionUnavailable( + "Missing API key for Jev provider: neither TYPESAFE_API_KEY nor JEV_API_KEY is configured" + ) + self._api_key = resolved_key + + @property + def name(self) -> str: + return "jev" + + @property + def endpoint(self) -> str: + return self._endpoint + + @property + def model(self) -> str: + return self._model + + @property + def timeout_seconds(self) -> float: + return self._timeout_seconds + + @property + def backoff_seconds(self) -> float: + return self._backoff_seconds + + def _execute_request(self, payload: dict[str, Any]) -> dict[str, Any]: + """Execute HTTP POST with single retry on timeout, fail-open on any error.""" + _require_https(self._endpoint) + + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + self._endpoint, + data=data, + headers={ + "Content-Type": "application/json", + "Accept": "application/json", + "Authorization": f"Bearer {self._api_key}", + }, + method="POST", + ) + + opener = urllib.request.build_opener(NoAuthRedirectHandler) + max_retries = 1 + + for attempt in range(max_retries + 1): + try: + with opener.open(req, timeout=self._timeout_seconds) as resp: + body = resp.read().decode("utf-8") + parsed = json.loads(body) + if not isinstance(parsed, dict): + raise TypedDecisionUnavailable( + f"Malformed response from Jev API: expected JSON object, got {type(parsed).__name__}" + ) + return parsed + + except TypedDecisionUnavailable: + raise + + except TimeoutError as exc: + if attempt < max_retries: + if self._backoff_seconds > 0: + time.sleep(self._backoff_seconds) + continue + raise TypedDecisionUnavailable( + f"Request to Jev API timed out after retry ({self._timeout_seconds}s limit)" + ) from exc + + except urllib.error.HTTPError as exc: + msg = _read_http_error(exc) + raise TypedDecisionUnavailable(f"Jev API returned HTTP {exc.code}: {msg}") from exc + + except urllib.error.URLError as exc: + if _is_timeout(exc): + if attempt < max_retries: + if self._backoff_seconds > 0: + time.sleep(self._backoff_seconds) + continue + raise TypedDecisionUnavailable( + f"Request to Jev API timed out after retry ({self._timeout_seconds}s limit)" + ) from exc + raise TypedDecisionUnavailable(f"Failed to connect to Jev API: {exc.reason}") from exc + + except json.JSONDecodeError as exc: + raise TypedDecisionUnavailable(f"Failed to parse JSON response from Jev API: {exc}") from exc + + except Exception as exc: + raise TypedDecisionUnavailable(f"Unexpected error communicating with Jev API: {exc}") from exc + + raise TypedDecisionUnavailable("Request to Jev API failed after retry") + + def choice(self, prompt: str, options: list[str]) -> dict[str, Any]: + """Select a single option label from candidate options. + + Returns: + {"label": str, "confidence": float} + """ + if not options: + raise TypedDecisionUnavailable("choice operation requires a non-empty list of options") + + vetted_prompt = enforce_privacy_gate(prompt, self._endpoint, provider_name="TypeSafe Jev") + payload = { + "model": self._model, + "operation": "choice", + "prompt": vetted_prompt, + "options": list(options), + } + + resp = self._execute_request(payload) + + try: + if not isinstance(resp, dict): + raise TypedDecisionUnavailable( + f"Malformed response from Jev API: expected JSON object, got {type(resp).__name__}" + ) + + result = _extract_result_dict(resp) + + label = result.get("label") + if label is None: + raise TypedDecisionUnavailable("Jev API response missing 'label'") + if isinstance(label, bool): + raise TypedDecisionUnavailable(f"Jev API returned boolean for 'label': {label!r}") + label_str = str(label) + if label_str not in options: + raise TypedDecisionUnavailable( + f"Jev API returned label {label_str!r} not in candidate options: {options}" + ) + + raw_conf = result.get("confidence") + if raw_conf is None: + raise TypedDecisionUnavailable("Jev API response missing 'confidence'") + if isinstance(raw_conf, bool): + raise TypedDecisionUnavailable( + f"Jev API returned non-numeric confidence {raw_conf!r}: boolean is not allowed" + ) + try: + confidence = float(raw_conf) + except (ValueError, TypeError) as exc: + raise TypedDecisionUnavailable( + f"Jev API returned non-numeric confidence {raw_conf!r}: {exc}" + ) from exc + + if not 0.0 <= confidence <= 1.0: + raise TypedDecisionUnavailable( + f"Jev API returned confidence {confidence} outside valid range [0.0, 1.0]" + ) + + return { + "label": label_str, + "confidence": confidence, + } + except TypedDecisionUnavailable: + raise + except Exception as exc: + raise TypedDecisionUnavailable(f"Malformed response from Jev API: {exc}") from exc + + def score( + self, + prompt: str, + scale: tuple[float, float] | list[float] | int | float, + ) -> dict[str, Any]: + """Evaluate a score for prompt along given scale. + + Returns: + {"value": float | int, "confidence": float} + """ + norm_scale: tuple[float, float] + if isinstance(scale, bool): + raise TypedDecisionUnavailable(f"Invalid scale: bool ({scale!r}) is not a valid numeric scale") + elif isinstance(scale, (int, float)): + try: + s_max = float(scale) + if not math.isfinite(s_max) or s_max <= 0: + raise TypedDecisionUnavailable( + f"Invalid scalar scale: upper bound ({s_max}) must be finite and strictly positive" + ) + norm_scale = (0.0, s_max) + except (ValueError, TypeError) as exc: + raise TypedDecisionUnavailable(f"Invalid scalar scale {scale!r}: {exc}") from exc + elif isinstance(scale, (tuple, list)): + if len(scale) != 2: + raise TypedDecisionUnavailable("scale must be a (min, max) pair of 2 values") + if isinstance(scale[0], bool) or isinstance(scale[1], bool): + raise TypedDecisionUnavailable("scale bounds must not be bool") + try: + s_min, s_max = float(scale[0]), float(scale[1]) + if not (math.isfinite(s_min) and math.isfinite(s_max)): + raise TypedDecisionUnavailable( + f"Invalid scale: bounds must be finite numbers, got ({s_min}, {s_max})" + ) + if s_min >= s_max: + raise TypedDecisionUnavailable( + f"Invalid scale: min ({s_min}) must be strictly less than max ({s_max})" + ) + norm_scale = (s_min, s_max) + except (ValueError, TypeError) as exc: + raise TypedDecisionUnavailable(f"Invalid scale values {scale!r}: {exc}") from exc + else: + raise TypedDecisionUnavailable(f"Invalid scale type: {type(scale).__name__}") + + vetted_prompt = enforce_privacy_gate(prompt, self._endpoint, provider_name="TypeSafe Jev") + payload = { + "model": self._model, + "operation": "score", + "prompt": vetted_prompt, + "scale": [norm_scale[0], norm_scale[1]], + } + + resp = self._execute_request(payload) + + try: + if not isinstance(resp, dict): + raise TypedDecisionUnavailable( + f"Malformed response from Jev API: expected JSON object, got {type(resp).__name__}" + ) + + result = _extract_result_dict(resp) + + raw_val = result.get("value") + if raw_val is None: + raise TypedDecisionUnavailable("Jev API response missing 'value'") + if isinstance(raw_val, bool): + raise TypedDecisionUnavailable( + f"Jev API returned non-numeric score value {raw_val!r}: boolean is not allowed" + ) + try: + value = float(raw_val) + except (ValueError, TypeError) as exc: + raise TypedDecisionUnavailable( + f"Jev API returned non-numeric score value {raw_val!r}: {exc}" + ) from exc + + if not math.isfinite(value) or not norm_scale[0] <= value <= norm_scale[1]: + raise TypedDecisionUnavailable( + f"Jev API returned score {value} outside scale range [{norm_scale[0]}, {norm_scale[1]}]" + ) + + raw_conf = result.get("confidence") + if raw_conf is None: + raise TypedDecisionUnavailable("Jev API response missing 'confidence'") + if isinstance(raw_conf, bool): + raise TypedDecisionUnavailable( + f"Jev API returned non-numeric confidence {raw_conf!r}: boolean is not allowed" + ) + try: + confidence = float(raw_conf) + except (ValueError, TypeError) as exc: + raise TypedDecisionUnavailable( + f"Jev API returned non-numeric confidence {raw_conf!r}: {exc}" + ) from exc + + if not 0.0 <= confidence <= 1.0: + raise TypedDecisionUnavailable( + f"Jev API returned confidence {confidence} outside valid range [0.0, 1.0]" + ) + + return { + "value": value, + "confidence": confidence, + } + except TypedDecisionUnavailable: + raise + except Exception as exc: + raise TypedDecisionUnavailable(f"Malformed response from Jev API: {exc}") from exc + + def noul(self, prompt: str) -> dict[str, Any]: + """Evaluate null/binary decision probability. + + Returns: + {"probability": float} + """ + vetted_prompt = enforce_privacy_gate(prompt, self._endpoint, provider_name="TypeSafe Jev") + payload = { + "model": self._model, + "operation": "noul", + "prompt": vetted_prompt, + } + + resp = self._execute_request(payload) + + try: + if not isinstance(resp, dict): + raise TypedDecisionUnavailable( + f"Malformed response from Jev API: expected JSON object, got {type(resp).__name__}" + ) + + result = _extract_result_dict(resp) + + raw_prob = result.get("probability") + if raw_prob is None: + raise TypedDecisionUnavailable("Jev API response missing 'probability'") + if isinstance(raw_prob, bool): + raise TypedDecisionUnavailable( + f"Jev API returned non-numeric probability {raw_prob!r}: boolean is not allowed" + ) + try: + probability = float(raw_prob) + except (ValueError, TypeError) as exc: + raise TypedDecisionUnavailable( + f"Jev API returned non-numeric probability {raw_prob!r}: {exc}" + ) from exc + + if not 0.0 <= probability <= 1.0: + raise TypedDecisionUnavailable( + f"Jev API returned probability {probability} outside valid range [0.0, 1.0]" + ) + + return { + "probability": probability, + } + except TypedDecisionUnavailable: + raise + except Exception as exc: + raise TypedDecisionUnavailable(f"Malformed response from Jev API: {exc}") from exc diff --git a/tools/typed-decision/src/typed_decision/registry.py b/tools/typed-decision/src/typed_decision/registry.py new file mode 100644 index 000000000..0f7973225 --- /dev/null +++ b/tools/typed-decision/src/typed_decision/registry.py @@ -0,0 +1,77 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Provider registry and configuration resolution for typed decisions.""" + +from __future__ import annotations + +import os +from typing import Any + +from typed_decision.exceptions import TypedDecisionUnavailable +from typed_decision.interface import DecisionProvider +from typed_decision.providers.jev import JevProvider, _resolve_api_key + +DEFAULT_PROVIDER_ENV = "MAGPIE_TYPED_DECISION_PROVIDER" + +_REGISTRY: dict[str, type[DecisionProvider]] = { + "jev": JevProvider, +} + + +def register_provider(name: str, provider_cls: type[DecisionProvider]) -> None: + """Register a provider class in the registry.""" + _REGISTRY[name.strip().lower()] = provider_cls + + +def _is_jev_configured() -> bool: + """Check if Jev credentials are configured in environment or home directory.""" + return _resolve_api_key() is not None + + +def get_provider(name: str | None = None, **kwargs: Any) -> DecisionProvider: + """Resolve and instantiate a typed decision provider. + + Resolution order: + 1. Explicit `name` argument if provided. + 2. Environment variable `$MAGPIE_TYPED_DECISION_PROVIDER` if set. + 3. If neither is set, defaults to 'jev' IF configured (i.e. credentials exist). + 4. If not configured, raises TypedDecisionUnavailable. + + Raises: + TypedDecisionUnavailable: If the provider is unknown, or if no provider is configured. + """ + selected_name = name or os.environ.get(DEFAULT_PROVIDER_ENV) + + if selected_name: + normalized = selected_name.strip().lower() + provider_cls = _REGISTRY.get(normalized) + if provider_cls is None: + raise TypedDecisionUnavailable( + f"Unknown typed decision provider '{selected_name}'. " + f"Available providers: {', '.join(sorted(_REGISTRY.keys()))}" + ) + return provider_cls(**kwargs) + + # Unset: default to 'jev' if configured, else unavailable + if _is_jev_configured(): + return JevProvider(**kwargs) + + raise TypedDecisionUnavailable( + "No typed decision provider configured. Set MAGPIE_TYPED_DECISION_PROVIDER " + "or provide TYPESAFE_API_KEY credentials for the default 'jev' provider." + ) diff --git a/tools/typed-decision/tests/__init__.py b/tools/typed-decision/tests/__init__.py new file mode 100644 index 000000000..13a83393a --- /dev/null +++ b/tools/typed-decision/tests/__init__.py @@ -0,0 +1,16 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. diff --git a/tools/typed-decision/tests/test_typed_decision.py b/tools/typed-decision/tests/test_typed_decision.py new file mode 100644 index 000000000..5b01c2066 --- /dev/null +++ b/tools/typed-decision/tests/test_typed_decision.py @@ -0,0 +1,1010 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Unit tests for the typed-decision tool contract, Jev provider, and privacy gate.""" + +from __future__ import annotations + +import io +import json +import pathlib +import urllib.error +import urllib.request +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +from typed_decision import ( + DecisionProvider, + JevProvider, + TypedDecisionClient, + TypedDecisionUnavailable, + choice, + get_provider, + noul, + score, +) +from typed_decision.privacy import enforce_privacy_gate +from typed_decision.providers.jev import ( + DEFAULT_ENDPOINT, + JEV_MODEL, + NoAuthRedirectHandler, + _require_https, +) +from typed_decision.registry import _is_jev_configured + + +def _make_mock_response(body: dict[str, Any] | list[Any] | str | int, status: int = 200) -> MagicMock: + """Helper to produce a mock HTTP response object for urllib.""" + mock_resp = MagicMock() + mock_resp.status = status + if isinstance(body, (dict, list, int)): + raw_bytes = json.dumps(body).encode("utf-8") + else: + raw_bytes = body.encode("utf-8") + mock_resp.read.return_value = raw_bytes + mock_resp.__enter__.return_value = mock_resp + mock_resp.__exit__.return_value = None + return mock_resp + + +@pytest.fixture(autouse=True) +def clean_environment(monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path) -> None: + """Ensure tests run with clean typed-decision environment by default. + + ``HOME`` and the working directory point at an empty temp dir, so a real + ``~/.config/apache-magpie/typesafe.key`` or a ``privacy-llm.md`` in the + caller's checkout cannot leak into the tests. + """ + monkeypatch.delenv("MAGPIE_TYPED_DECISION_PROVIDER", raising=False) + monkeypatch.delenv("TYPESAFE_API_KEY", raising=False) + monkeypatch.delenv("JEV_API_KEY", raising=False) + monkeypatch.delenv("PRIVACY_LLM_CONFIG", raising=False) + monkeypatch.delenv("USERPROFILE", raising=False) + monkeypatch.setenv("HOME", str(tmp_path / "home")) + monkeypatch.chdir(tmp_path) + + +@pytest.fixture +def approved_privacy_config(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: + """Provide a valid privacy-llm.md approving the default Jev endpoint.""" + config_file = tmp_path / "privacy-llm.md" + config_file.write_text( + "# Privacy LLM Configuration\n\n" + "## Approved third-party endpoints (opt-in)\n\n" + "- api.typesafe.ai (TypeSafe Jev)\n" + " - Data-residency contract: https://typesafe.ai/legal/dpa-strict\n" + " - Approved-by: JP 2026-09-01\n", + encoding="utf-8", + ) + monkeypatch.setenv("PRIVACY_LLM_CONFIG", str(config_file)) + return config_file + + +# --------------------------------------------------------------------------- +# 1. Construction and Missing Key Tests +# --------------------------------------------------------------------------- + + +def test_missing_key_at_construction_raises_unavailable() -> None: + """Missing API key must raise TypedDecisionUnavailable at provider construction.""" + with pytest.raises(TypedDecisionUnavailable, match="Missing API key"): + JevProvider() + + +def test_explicit_key_at_construction_succeeds() -> None: + """Providing explicit api_key succeeds at construction.""" + provider = JevProvider(api_key="explicit-secret-key") + assert provider.name == "jev" + assert provider.model == JEV_MODEL + + +def test_env_var_key_at_construction_succeeds(monkeypatch: pytest.MonkeyPatch) -> None: + """Setting TYPESAFE_API_KEY environment variable succeeds at construction.""" + monkeypatch.setenv("TYPESAFE_API_KEY", "env-secret-key") + provider = JevProvider() + assert provider.name == "jev" + + +def test_fallback_env_var_key_succeeds(monkeypatch: pytest.MonkeyPatch) -> None: + """Setting JEV_API_KEY environment variable succeeds as fallback.""" + monkeypatch.setenv("JEV_API_KEY", "jev-fallback-key") + provider = JevProvider() + assert provider.name == "jev" + + +def test_home_key_file_is_picked_up() -> None: + """A key at ~/.config/apache-magpie/typesafe.key is used when no env var is set.""" + key_dir = pathlib.Path.home() / ".config" / "apache-magpie" + key_dir.mkdir(parents=True) + (key_dir / "typesafe.key").write_text("file-secret-key\n", encoding="utf-8") + assert _is_jev_configured() is True + provider = JevProvider() + assert provider._api_key == "file-secret-key" + + +# --------------------------------------------------------------------------- +# 2. Pinned Model Version (RFC-AI-0004 § Principle 3) +# --------------------------------------------------------------------------- + + +def test_model_version_is_pinned_constant() -> None: + """Model version must be a pinned constant and never 'latest'.""" + assert JEV_MODEL == "systemone-2026-06-01" + assert "latest" not in JEV_MODEL.lower() + + provider = JevProvider(api_key="test-key") + assert provider.model == JEV_MODEL + assert "latest" not in provider.model.lower() + + +# --------------------------------------------------------------------------- +# 3. Successful Operation Calls (Mocked) +# --------------------------------------------------------------------------- + + +def test_jev_choice_success(approved_privacy_config: pathlib.Path) -> None: + """Mocked successful choice operation returns {label, confidence}.""" + provider = JevProvider(api_key="test-key") + expected_response = { + "label": "bug", + "confidence": 0.96, + } + + mock_resp = _make_mock_response(expected_response) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp) as mock_open: + result = provider.choice("Classify issue description", ["bug", "feature", "question"]) + + assert result == {"label": "bug", "confidence": 0.96} + assert mock_open.call_count == 1 + + req: urllib.request.Request = mock_open.call_args[0][0] + assert req.full_url == DEFAULT_ENDPOINT + assert req.headers["Authorization"] == "Bearer test-key" + assert req.headers["Content-type"] == "application/json" + + assert isinstance(req.data, bytes) + body = json.loads(req.data.decode("utf-8")) + assert body["model"] == JEV_MODEL + assert body["operation"] == "choice" + assert body["prompt"] == "Classify issue description" + assert body["options"] == ["bug", "feature", "question"] + + +def test_jev_score_success(approved_privacy_config: pathlib.Path) -> None: + """Mocked successful score operation returns {value, confidence}.""" + provider = JevProvider(api_key="test-key") + expected_response = { + "value": 4.5, + "confidence": 0.89, + } + + mock_resp = _make_mock_response(expected_response) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp) as mock_open: + result = provider.score("Rate severity of vulnerability", scale=(1, 5)) + + assert result == {"value": 4.5, "confidence": 0.89} + assert mock_open.call_count == 1 + + req: urllib.request.Request = mock_open.call_args[0][0] + assert isinstance(req.data, bytes) + body = json.loads(req.data.decode("utf-8")) + assert body["operation"] == "score" + assert body["scale"] == [1, 5] + + +def test_jev_noul_success(approved_privacy_config: pathlib.Path) -> None: + """Mocked successful noul operation returns {probability}.""" + provider = JevProvider(api_key="test-key") + expected_response = { + "probability": 0.78, + } + + mock_resp = _make_mock_response(expected_response) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp) as mock_open: + result = provider.noul("Is this report actionable?") + + assert result == {"probability": 0.78} + assert mock_open.call_count == 1 + + req: urllib.request.Request = mock_open.call_args[0][0] + assert isinstance(req.data, bytes) + body = json.loads(req.data.decode("utf-8")) + assert body["operation"] == "noul" + assert body["prompt"] == "Is this report actionable?" + + +def test_jev_nested_result_object(approved_privacy_config: pathlib.Path) -> None: + """Provider handles nested 'result' or 'decision' dictionaries from backend.""" + provider = JevProvider(api_key="test-key") + nested_response = { + "result": { + "label": "feature", + "confidence": 0.92, + } + } + mock_resp = _make_mock_response(nested_response) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + res = provider.choice("Request new CLI flag", ["bug", "feature"]) + assert res == {"label": "feature", "confidence": 0.92} + + +# --------------------------------------------------------------------------- +# 4. Timeout and Retry Policy Tests +# --------------------------------------------------------------------------- + + +def test_timeout_retries_once_then_unavailable(approved_privacy_config: pathlib.Path) -> None: + """On timeout, client retries exactly once with backoff, then raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key", backoff_seconds=0.001) + + timeout_exc = TimeoutError("Connection timed out") + + with patch("urllib.request.OpenerDirector.open", side_effect=timeout_exc) as mock_open: + with pytest.raises(TypedDecisionUnavailable, match="timed out after retry"): + provider.choice("Prompt", ["opt1", "opt2"]) + + # Must have attempted exactly 2 times (initial attempt + 1 retry) + assert mock_open.call_count == 2 + + +def test_timeout_on_urlerror_retries_once_then_unavailable(approved_privacy_config: pathlib.Path) -> None: + """URLError caused by timeout retries once, then raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key", backoff_seconds=0.001) + timeout_url_err = urllib.error.URLError("timed out") + + with patch("urllib.request.OpenerDirector.open", side_effect=timeout_url_err) as mock_open: + with pytest.raises(TypedDecisionUnavailable, match="timed out after retry"): + provider.score("Prompt", (1, 10)) + + assert mock_open.call_count == 2 + + +def test_timeout_first_attempt_retry_succeeds(approved_privacy_config: pathlib.Path) -> None: + """First attempt times out, retry succeeds: operation returns response.""" + provider = JevProvider(api_key="test-key", backoff_seconds=0.001) + + mock_success = _make_mock_response({"probability": 0.65}) + side_effects = [TimeoutError("Transient timeout"), mock_success] + + with patch("urllib.request.OpenerDirector.open", side_effect=side_effects) as mock_open: + res = provider.noul("Test prompt") + assert res == {"probability": 0.65} + assert mock_open.call_count == 2 + + +# --------------------------------------------------------------------------- +# 5. Outbound Content Passes Through Privacy-LLM Gate (Deny-by-Default) +# --------------------------------------------------------------------------- + + +def test_outbound_passes_through_privacy_gate(approved_privacy_config: pathlib.Path) -> None: + """Every outbound prompt must be routed through enforce_privacy_gate before dispatch.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"label": "approved", "confidence": 1.0}) + + with ( + patch("typed_decision.providers.jev.enforce_privacy_gate", wraps=enforce_privacy_gate) as mock_gate, + patch("urllib.request.OpenerDirector.open", return_value=mock_resp), + ): + provider.choice("Confidential triage prompt", ["approved", "rejected"]) + + mock_gate.assert_called_once_with( + "Confidential triage prompt", DEFAULT_ENDPOINT, provider_name="TypeSafe Jev" + ) + + +def test_name_only_opt_in_does_not_approve_different_host( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An opt-in matching 'TypeSafe Jev' by name must not approve an arbitrary endpoint host.""" + config_file = tmp_path / "privacy-llm.md" + config_file.write_text( + "# Privacy LLM Configuration\n\n" + "## Approved third-party endpoints (opt-in)\n\n" + "- TypeSafe Jev\n" + " - Data-residency contract: https://typesafe.ai/legal/dpa\n" + " - Approved-by: JP 2026-09-01\n", + encoding="utf-8", + ) + monkeypatch.setenv("PRIVACY_LLM_CONFIG", str(config_file)) + + # DEFAULT_ENDPOINT matches via provider_name + enforce_privacy_gate("Test prompt", DEFAULT_ENDPOINT, provider_name="TypeSafe Jev") + + # Different endpoint must NOT be approved even if provider_name is passed + with pytest.raises(TypedDecisionUnavailable, match="Privacy-LLM gate rejected outbound request"): + enforce_privacy_gate( + "Test prompt", + "https://unapproved.attacker.example.com/v1", + provider_name="TypeSafe Jev", + ) + + +def test_privacy_gate_rejects_url_fragment() -> None: + """Outbound endpoint with a fragment is rejected and cannot trigger free-text rules.""" + with pytest.raises(TypedDecisionUnavailable, match="fragment"): + enforce_privacy_gate("Test prompt", "https://evil.example.com/v1#claude code") + + +def test_privacy_gate_rejects_subdomain_opt_in_bypass( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An opt-in for api.typesafe.ai does not approve api.typesafe.ai.evil.example.""" + config_file = tmp_path / "privacy-llm.md" + config_file.write_text( + "# Privacy LLM Configuration\n\n" + "## Approved third-party endpoints (opt-in)\n\n" + "- https://api.typesafe.ai\n" + " - Data-residency contract: https://typesafe.ai/legal/dpa\n" + " - Approved-by: JP 2026-09-01\n", + encoding="utf-8", + ) + monkeypatch.setenv("PRIVACY_LLM_CONFIG", str(config_file)) + + with pytest.raises(TypedDecisionUnavailable, match="Privacy-LLM gate rejected outbound request"): + enforce_privacy_gate("Test prompt", "https://api.typesafe.ai.evil.example/v1") + + +def test_privacy_gate_rejects_userinfo_opt_in_bypass( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An endpoint with userinfo (@) is rejected by the privacy gate.""" + config_file = tmp_path / "privacy-llm.md" + config_file.write_text( + "# Privacy LLM Configuration\n\n" + "## Approved third-party endpoints (opt-in)\n\n" + "- https://api.typesafe.ai\n" + " - Data-residency contract: https://typesafe.ai/legal/dpa\n" + " - Approved-by: JP 2026-09-01\n", + encoding="utf-8", + ) + monkeypatch.setenv("PRIVACY_LLM_CONFIG", str(config_file)) + + with pytest.raises(TypedDecisionUnavailable, match="userinfo"): + enforce_privacy_gate("Test prompt", "https://api.typesafe.ai@evil.example/v1") + + +def test_privacy_gate_name_only_opt_in_rejects_similar_host( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A name-only opt-in ('TypeSafe — Jev API') does not approve typesafe.evil.example.""" + config_file = tmp_path / "privacy-llm.md" + config_file.write_text( + "# Privacy LLM Configuration\n\n" + "## Approved third-party endpoints (opt-in)\n\n" + "- TypeSafe — Jev API\n" + " - Data-residency contract: https://typesafe.ai/legal/dpa\n" + " - Approved-by: JP 2026-09-01\n", + encoding="utf-8", + ) + monkeypatch.setenv("PRIVACY_LLM_CONFIG", str(config_file)) + + # Reject non-default endpoint + with pytest.raises(TypedDecisionUnavailable, match="Privacy-LLM gate rejected outbound request"): + enforce_privacy_gate("Test prompt", "https://typesafe.evil.example/v1") + + +def test_privacy_gate_rejection_prevents_network_egress() -> None: + """When the privacy gate rejects an endpoint, no HTTP network call is made.""" + provider = JevProvider(api_key="test-key") + + with patch("urllib.request.OpenerDirector.open") as mock_open: + with pytest.raises(TypedDecisionUnavailable, match="Privacy-LLM gate rejected outbound request"): + provider.choice("Test prompt", ["a", "b"]) + + assert mock_open.call_count == 0 + + +def test_privacy_gate_denies_by_default_when_no_config( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Third-party endpoints are denied by default when no privacy-llm config is found.""" + monkeypatch.delenv("PRIVACY_LLM_CONFIG", raising=False) + monkeypatch.chdir(tmp_path) + provider = JevProvider(api_key="test-key") + + with pytest.raises(TypedDecisionUnavailable, match="Privacy-LLM gate rejected outbound request"): + provider.choice("Test prompt", ["a", "b"]) + + with pytest.raises(TypedDecisionUnavailable, match="no privacy-llm config found"): + enforce_privacy_gate("Test prompt", DEFAULT_ENDPOINT, provider_name="TypeSafe Jev") + + +def test_privacy_gate_denies_missing_config_file( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """When PRIVACY_LLM_CONFIG points to a missing file, third-party endpoints are denied.""" + monkeypatch.setenv("PRIVACY_LLM_CONFIG", str(tmp_path / "nonexistent-privacy-llm.md")) + + with pytest.raises(TypedDecisionUnavailable, match="Failed to parse privacy-llm config"): + enforce_privacy_gate("Prompt", DEFAULT_ENDPOINT, provider_name="TypeSafe Jev") + + +def test_privacy_gate_denies_unapproved_third_party_endpoint( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Third-party endpoint not listed in opt-in section is denied.""" + config_file = tmp_path / "privacy-llm.md" + config_file.write_text( + "# Privacy LLM Configuration\n\n" + "## Approved third-party endpoints (opt-in)\n\n" + "- api.other.ai (Other AI Service)\n" + " - Data-residency contract: https://other.ai/legal/dpa\n" + " - Approved-by: JP 2026-09-01\n", + encoding="utf-8", + ) + monkeypatch.setenv("PRIVACY_LLM_CONFIG", str(config_file)) + + with pytest.raises(TypedDecisionUnavailable, match="no opt-in entry was declared"): + enforce_privacy_gate("Prompt", DEFAULT_ENDPOINT, provider_name="TypeSafe Jev") + + +def test_privacy_gate_denies_endpoint_in_currently_configured_stack_only( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Endpoint declared only under 'Currently configured LLM stack' is denied without opt-in sign-off.""" + config_file = tmp_path / "privacy-llm.md" + config_file.write_text( + "# Privacy LLM Configuration\n\n" + "## Currently configured LLM stack\n\n" + "- api.typesafe.ai (TypeSafe Jev)\n\n" + "## Approved third-party endpoints (opt-in)\n\n" + "- none\n", + encoding="utf-8", + ) + monkeypatch.setenv("PRIVACY_LLM_CONFIG", str(config_file)) + + with pytest.raises(TypedDecisionUnavailable, match="no opt-in entry was declared"): + enforce_privacy_gate("Prompt", DEFAULT_ENDPOINT, provider_name="TypeSafe Jev") + + +def test_privacy_gate_denies_placeholder_approved_by( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Opt-in entry with placeholder Approved-by (e.g. ) is denied.""" + config_file = tmp_path / "privacy-llm.md" + config_file.write_text( + "# Privacy LLM Configuration\n\n" + "## Approved third-party endpoints (opt-in)\n\n" + "- api.typesafe.ai (TypeSafe Jev)\n" + " - Data-residency contract: https://typesafe.ai/legal/dpa-strict\n" + " - Approved-by: 2026-09-01\n", + encoding="utf-8", + ) + monkeypatch.setenv("PRIVACY_LLM_CONFIG", str(config_file)) + + with pytest.raises(TypedDecisionUnavailable, match="placeholder text"): + enforce_privacy_gate("Prompt", DEFAULT_ENDPOINT, provider_name="TypeSafe Jev") + + +def test_privacy_gate_denies_missing_data_residency( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Opt-in entry missing Data-residency contract is denied.""" + config_file = tmp_path / "privacy-llm.md" + config_file.write_text( + "# Privacy LLM Configuration\n\n" + "## Approved third-party endpoints (opt-in)\n\n" + "- api.typesafe.ai (TypeSafe Jev)\n" + " - Approved-by: JP 2026-09-01\n", + encoding="utf-8", + ) + monkeypatch.setenv("PRIVACY_LLM_CONFIG", str(config_file)) + + with pytest.raises(TypedDecisionUnavailable, match="missing the 'Data-residency contract'"): + enforce_privacy_gate("Prompt", DEFAULT_ENDPOINT, provider_name="TypeSafe Jev") + + +def test_privacy_gate_approves_valid_opt_in(approved_privacy_config: pathlib.Path) -> None: + """Opt-in entry with contract and real PMC Approved-by succeeds.""" + prompt = "Prompt destined for approved provider" + result = enforce_privacy_gate(prompt, DEFAULT_ENDPOINT, provider_name="TypeSafe Jev") + assert result == prompt + + +def test_privacy_gate_approves_default_rules() -> None: + """Localhost, 127.0.0.1, and *.apache.org endpoints are approved without config.""" + assert enforce_privacy_gate("Test", "http://localhost:8080/v1") == "Test" + assert enforce_privacy_gate("Test", "http://127.0.0.1:9000/v1") == "Test" + assert enforce_privacy_gate("Test", "https://infra.apache.org/v1") == "Test" + + +def test_privacy_gate_denies_carved_out_apache_host() -> None: + """llm.apache.org is carved out and denied by default without explicit opt-in.""" + with pytest.raises(TypedDecisionUnavailable, match=r"carved out of the \*.apache.org default approval"): + enforce_privacy_gate("Test", "https://llm.apache.org/v1") + + +# --------------------------------------------------------------------------- +# 6. Provider Registry and Configuration Resolution +# --------------------------------------------------------------------------- + + +def test_registry_explicit_env_jev(monkeypatch: pytest.MonkeyPatch) -> None: + """MAGPIE_TYPED_DECISION_PROVIDER=jev resolves to JevProvider.""" + monkeypatch.setenv("MAGPIE_TYPED_DECISION_PROVIDER", "jev") + monkeypatch.setenv("TYPESAFE_API_KEY", "test-key") + + provider = get_provider() + assert isinstance(provider, JevProvider) + assert provider.name == "jev" + + +def test_registry_unknown_provider_raises(monkeypatch: pytest.MonkeyPatch) -> None: + """MAGPIE_TYPED_DECISION_PROVIDER=unknown raises TypedDecisionUnavailable.""" + monkeypatch.setenv("MAGPIE_TYPED_DECISION_PROVIDER", "unsupported_provider") + + with pytest.raises(TypedDecisionUnavailable, match="Unknown typed decision provider"): + get_provider() + + +def test_registry_unset_defaults_to_jev_if_configured(monkeypatch: pytest.MonkeyPatch) -> None: + """When MAGPIE_TYPED_DECISION_PROVIDER is unset, defaults to jev if credentials exist.""" + monkeypatch.setenv("TYPESAFE_API_KEY", "test-key") + + provider = get_provider() + assert isinstance(provider, JevProvider) + + +def test_registry_unset_unavailable_if_unconfigured() -> None: + """When MAGPIE_TYPED_DECISION_PROVIDER is unset and no credentials, reports unavailable.""" + with pytest.raises(TypedDecisionUnavailable, match="No typed decision provider configured"): + get_provider() + + +def test_registry_is_jev_configured_deduplicated(monkeypatch: pytest.MonkeyPatch) -> None: + """_is_jev_configured() delegates to _resolve_api_key().""" + assert _is_jev_configured() is False + + monkeypatch.setenv("TYPESAFE_API_KEY", "key-1") + assert _is_jev_configured() is True + + monkeypatch.delenv("TYPESAFE_API_KEY") + monkeypatch.setenv("JEV_API_KEY", "key-2") + assert _is_jev_configured() is True + + +# --------------------------------------------------------------------------- +# 7. Fail-Open, Validation, and Error Handling Tests +# --------------------------------------------------------------------------- + + +def test_http_error_fail_open(approved_privacy_config: pathlib.Path) -> None: + """HTTP 500 error raises TypedDecisionUnavailable without fabricated fallback.""" + provider = JevProvider(api_key="test-key") + err_body = json.dumps({"error": "Internal server error in model execution"}).encode("utf-8") + http_err = urllib.error.HTTPError( + url=DEFAULT_ENDPOINT, + code=500, + msg="Internal Server Error", + hdrs=MagicMock(), + fp=io.BytesIO(err_body), + ) + + with patch("urllib.request.OpenerDirector.open", side_effect=http_err): + with pytest.raises(TypedDecisionUnavailable, match="HTTP 500"): + provider.choice("Test", ["a", "b"]) + + +def test_http_auth_error_fail_open(approved_privacy_config: pathlib.Path) -> None: + """HTTP 401 Unauthorized raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="bad-key") + http_err = urllib.error.HTTPError( + url=DEFAULT_ENDPOINT, + code=401, + msg="Unauthorized", + hdrs=MagicMock(), + fp=io.BytesIO(b'{"error": "Invalid API key"}'), + ) + + with patch("urllib.request.OpenerDirector.open", side_effect=http_err): + with pytest.raises(TypedDecisionUnavailable, match="HTTP 401"): + provider.score("Test", (1, 5)) + + +def test_invalid_json_fail_open(approved_privacy_config: pathlib.Path) -> None: + """Non-JSON response raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response("Bad Gateway (HTML error page)", status=502) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="Failed to parse JSON"): + provider.noul("Test") + + +def test_malformed_response_object_handled_fail_open(approved_privacy_config: pathlib.Path) -> None: + """Non-dictionary JSON responses raise TypedDecisionUnavailable fail-open.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response(["not", "a", "dict"]) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="Malformed response from Jev API"): + provider.choice("Test", ["a", "b"]) + + +def test_choice_empty_options_raises() -> None: + """Passing empty options list raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + with pytest.raises(TypedDecisionUnavailable, match="non-empty list of options"): + provider.choice("Test prompt", []) + + +def test_choice_missing_label_in_response(approved_privacy_config: pathlib.Path) -> None: + """Response missing label raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"confidence": 0.95}) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="missing 'label'"): + provider.choice("Test", ["a", "b"]) + + +def test_choice_rejects_missing_confidence(approved_privacy_config: pathlib.Path) -> None: + """Choice response missing confidence must raise TypedDecisionUnavailable (never default to 1.0).""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"label": "a"}) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="missing 'confidence'"): + provider.choice("Test", ["a", "b"]) + + +def test_choice_rejects_label_not_in_options(approved_privacy_config: pathlib.Path) -> None: + """Choice response returning a label outside the provided options must raise TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"label": "unexpected", "confidence": 0.9}) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="not in candidate options"): + provider.choice("Test", ["a", "b"]) + + +def test_choice_rejects_invalid_confidence_range(approved_privacy_config: pathlib.Path) -> None: + """Confidence outside [0.0, 1.0] must raise TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + + mock_resp_high = _make_mock_response({"label": "a", "confidence": 1.5}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_high): + with pytest.raises(TypedDecisionUnavailable, match="outside valid range"): + provider.choice("Test", ["a", "b"]) + + mock_resp_low = _make_mock_response({"label": "a", "confidence": -0.1}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_low): + with pytest.raises(TypedDecisionUnavailable, match="outside valid range"): + provider.choice("Test", ["a", "b"]) + + +def test_choice_rejects_non_numeric_confidence(approved_privacy_config: pathlib.Path) -> None: + """Non-numeric confidence raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"label": "a", "confidence": "high"}) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="non-numeric confidence"): + provider.choice("Test", ["a", "b"]) + + +def test_score_invalid_scale_argument() -> None: + """Passing invalid scale (min >= max) raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + with pytest.raises(TypedDecisionUnavailable, match="Invalid scale"): + provider.score("Test", scale=(5, 1)) + + +def test_score_scalar_scale_normalized(approved_privacy_config: pathlib.Path) -> None: + """Scalar scale=10 is normalized to (0, 10) and bounds-checked.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"value": 7.5, "confidence": 0.85}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp) as mock_open: + res = provider.score("Test", scale=10) + assert res["value"] == 7.5 + assert res["confidence"] == 0.85 + + req: urllib.request.Request = mock_open.call_args[0][0] + assert isinstance(req.data, bytes) + payload = json.loads(req.data.decode("utf-8")) + assert payload["scale"] == [0.0, 10.0] + + +def test_score_scalar_scale_rejects_out_of_bounds(approved_privacy_config: pathlib.Path) -> None: + """Scalar scale=10 rejects value=42 outside [0, 10].""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"value": 42.0, "confidence": 0.9}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="outside scale range"): + provider.score("Test", scale=10) + + +def test_score_rejects_invalid_scalar_scales() -> None: + """Scalar scale <= 0 or bool is rejected on input.""" + provider = JevProvider(api_key="test-key") + with pytest.raises(TypedDecisionUnavailable, match="Invalid scale: bool"): + provider.score("Test", scale=True) + + with pytest.raises(TypedDecisionUnavailable, match="strictly positive"): + provider.score("Test", scale=0) + + with pytest.raises(TypedDecisionUnavailable, match="strictly positive"): + provider.score("Test", scale=-5) + + +def test_score_rejects_non_finite_scales() -> None: + """Non-finite scales (nan, inf) are rejected on input.""" + provider = JevProvider(api_key="test-key") + + with pytest.raises(TypedDecisionUnavailable, match="finite"): + provider.score("Test", scale=float("nan")) + + with pytest.raises(TypedDecisionUnavailable, match="finite"): + provider.score("Test", scale=float("inf")) + + with pytest.raises(TypedDecisionUnavailable, match="finite"): + provider.score("Test", scale=(0, float("inf"))) + + with pytest.raises(TypedDecisionUnavailable, match="finite"): + provider.score("Test", scale=(float("-inf"), 10)) + + with pytest.raises(TypedDecisionUnavailable, match="finite"): + provider.score("Test", scale=(float("nan"), 5)) + + with pytest.raises(TypedDecisionUnavailable, match="finite"): + provider.score("Test", scale=(0, float("nan"))) + + +def test_score_rejects_non_finite_response_value(approved_privacy_config: pathlib.Path) -> None: + """Score response value that is NaN or Inf raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + + mock_resp_nan = _make_mock_response({"value": float("nan"), "confidence": 0.9}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_nan): + with pytest.raises(TypedDecisionUnavailable, match="outside scale range"): + provider.score("Test", scale=(1, 5)) + + mock_resp_inf = _make_mock_response({"value": float("inf"), "confidence": 0.9}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_inf): + with pytest.raises(TypedDecisionUnavailable, match="outside scale range"): + provider.score("Test", scale=(1, 5)) + + +def test_score_rejects_missing_value(approved_privacy_config: pathlib.Path) -> None: + """Score response missing value raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"confidence": 0.8}) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="missing 'value'"): + provider.score("Test", scale=(1, 5)) + + +def test_score_rejects_value_outside_scale(approved_privacy_config: pathlib.Path) -> None: + """Score response value outside scale raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + + mock_resp_high = _make_mock_response({"value": 6.0, "confidence": 0.9}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_high): + with pytest.raises(TypedDecisionUnavailable, match="outside scale"): + provider.score("Test", scale=(1, 5)) + + mock_resp_low = _make_mock_response({"value": 0.5, "confidence": 0.9}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_low): + with pytest.raises(TypedDecisionUnavailable, match="outside scale"): + provider.score("Test", scale=(1, 5)) + + +def test_score_rejects_missing_confidence(approved_privacy_config: pathlib.Path) -> None: + """Score response missing confidence raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"value": 3.0}) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="missing 'confidence'"): + provider.score("Test", scale=(1, 5)) + + +def test_score_rejects_invalid_confidence_range(approved_privacy_config: pathlib.Path) -> None: + """Score response confidence outside [0.0, 1.0] raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"value": 3.0, "confidence": 1.2}) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="outside valid range"): + provider.score("Test", scale=(1, 5)) + + +def test_noul_rejects_missing_probability(approved_privacy_config: pathlib.Path) -> None: + """Noul response missing probability raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({}) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="missing 'probability'"): + provider.noul("Test") + + +def test_noul_rejects_invalid_probability_range(approved_privacy_config: pathlib.Path) -> None: + """Probability outside [0.0, 1.0] raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + + mock_resp_high = _make_mock_response({"probability": 1.1}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_high): + with pytest.raises(TypedDecisionUnavailable, match="outside valid range"): + provider.noul("Test") + + mock_resp_low = _make_mock_response({"probability": -0.2}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_low): + with pytest.raises(TypedDecisionUnavailable, match="outside valid range"): + provider.noul("Test") + + +def test_noul_rejects_non_numeric_probability(approved_privacy_config: pathlib.Path) -> None: + """Non-numeric probability raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"probability": "likely"}) + + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="non-numeric probability"): + provider.noul("Test") + + +# --------------------------------------------------------------------------- +# 8. High-Level Convenience Functions and Client API +# --------------------------------------------------------------------------- + + +def test_functional_and_client_api() -> None: + """Module-level functions and TypedDecisionClient wrap provider methods.""" + mock_provider = MagicMock(spec=DecisionProvider) + mock_provider.choice.return_value = {"label": "mock_choice", "confidence": 0.9} + mock_provider.score.return_value = {"value": 8, "confidence": 0.85} + mock_provider.noul.return_value = {"probability": 0.42} + + # Functional calls with explicit provider + assert choice("test", ["a", "b"], provider=mock_provider) == {"label": "mock_choice", "confidence": 0.9} + assert score("test", (1, 10), provider=mock_provider) == {"value": 8, "confidence": 0.85} + assert noul("test", provider=mock_provider) == {"probability": 0.42} + + # Client instance + client = TypedDecisionClient(provider=mock_provider) + assert client.choice("test", ["a", "b"]) == {"label": "mock_choice", "confidence": 0.9} + assert client.score("test", (1, 10)) == {"value": 8, "confidence": 0.85} + assert client.noul("test") == {"probability": 0.42} + + +# --------------------------------------------------------------------------- +# 9. Security Guards: HTTPS & Redirect Handler +# --------------------------------------------------------------------------- + + +def test_https_enforced() -> None: + """Insecure remote HTTP URLs must be rejected.""" + with pytest.raises(TypedDecisionUnavailable, match="must use HTTPS"): + _require_https("http://api.typesafe.ai/v1/systemone") + + # Localhost allowed for test fixtures + _require_https("http://127.0.0.1:8080/v1/systemone") + _require_https("https://api.typesafe.ai/v1/systemone") + + +def test_no_auth_redirect_handler_blocks_redirects() -> None: + """Redirect handler must reject redirects to prevent leaking Bearer tokens.""" + handler = NoAuthRedirectHandler() + req = urllib.request.Request("https://api.typesafe.ai/v1/systemone") + + with pytest.raises(TypedDecisionUnavailable, match="refusing to forward credentials"): + handler.redirect_request(req, None, 302, "Found", {}, "https://attacker.example.com") + + +# --------------------------------------------------------------------------- +# 10. Type-Confusion & Envelope Extraction Guards +# --------------------------------------------------------------------------- + + +def test_choice_rejects_boolean_label(approved_privacy_config: pathlib.Path) -> None: + """Boolean for 'label' must raise TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"label": True, "confidence": 0.9}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="boolean"): + provider.choice("Test", ["True", "False"]) + + +def test_choice_rejects_boolean_confidence(approved_privacy_config: pathlib.Path) -> None: + """Boolean for 'confidence' in choice must raise TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"label": "a", "confidence": True}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="boolean is not allowed"): + provider.choice("Test", ["a", "b"]) + + +def test_score_rejects_boolean_response_value(approved_privacy_config: pathlib.Path) -> None: + """Boolean for 'value' in score must raise TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp_false = _make_mock_response({"value": False, "confidence": 0.8}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_false): + with pytest.raises(TypedDecisionUnavailable, match="boolean is not allowed"): + provider.score("Test", scale=(0, 10)) + + mock_resp_true = _make_mock_response({"value": True, "confidence": 0.8}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_true): + with pytest.raises(TypedDecisionUnavailable, match="boolean is not allowed"): + provider.score("Test", scale=(0, 10)) + + +def test_score_rejects_boolean_confidence(approved_privacy_config: pathlib.Path) -> None: + """Boolean for 'confidence' in score must raise TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"value": 5.0, "confidence": True}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises(TypedDecisionUnavailable, match="boolean is not allowed"): + provider.score("Test", scale=(0, 10)) + + +def test_noul_rejects_boolean_probability(approved_privacy_config: pathlib.Path) -> None: + """Boolean for 'probability' in noul must raise TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp_true = _make_mock_response({"probability": True}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_true): + with pytest.raises(TypedDecisionUnavailable, match="boolean is not allowed"): + provider.noul("Test") + + mock_resp_false = _make_mock_response({"probability": False}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp_false): + with pytest.raises(TypedDecisionUnavailable, match="boolean is not allowed"): + provider.noul("Test") + + +def test_response_rejects_non_dict_result_field(approved_privacy_config: pathlib.Path) -> None: + """A non-dict 'result' payload (e.g. error string) raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"result": "unauthorized", "label": "a", "confidence": 0.9}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises( + TypedDecisionUnavailable, + match="Malformed 'result' in Jev API response: expected JSON object, got str", + ): + provider.choice("Test", ["a", "b"]) + + +def test_rejects_non_dict_decision_envelope(approved_privacy_config: pathlib.Path) -> None: + """A non-dict 'decision' payload raises TypedDecisionUnavailable.""" + provider = JevProvider(api_key="test-key") + mock_resp = _make_mock_response({"decision": "unauthorized", "probability": 0.5}) + with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): + with pytest.raises( + TypedDecisionUnavailable, + match="Malformed 'decision' in Jev API response: expected JSON object, got str", + ): + provider.noul("Test") + + +def test_privacy_gate_raw_desc_carries_provider_for_custom_endpoint() -> None: + """_check_endpoint_approved includes provider_name in raw_desc even for custom endpoints.""" + from typed_decision.privacy import _check_endpoint_approved + + with patch("typed_decision.privacy.check_endpoint") as mock_check: + mock_check.return_value = MagicMock(approved=True, reason="ok") + _check_endpoint_approved("https://custom.example.com/v1", provider_name="CustomProvider") + mock_check.assert_called_once() + _, kwargs = mock_check.call_args + assert kwargs["raw_desc"] == "CustomProvider (https://custom.example.com/v1)" diff --git a/tools/typed-decision/tool.md b/tools/typed-decision/tool.md new file mode 100644 index 000000000..8aca1d2bb --- /dev/null +++ b/tools/typed-decision/tool.md @@ -0,0 +1,199 @@ + + + + +**Table of Contents** *generated with [DocToc](https://github.com/thlorenz/doctoc)* + +- [Tool: Typed Decision](#tool-typed-decision) + - [What this tool provides](#what-this-tool-provides) + - [The Contract](#the-contract) + - [`choice(prompt, options) -> {label, confidence}`](#choiceprompt-options---label-confidence) + - [`score(prompt, scale) -> {value, confidence}`](#scoreprompt-scale---value-confidence) + - [`noul(prompt) -> {probability}`](#noulprompt---probability) + - [Fail-Open Contract](#fail-open-contract) + - [Architecture and Providers](#architecture-and-providers) + - [Provider Interface](#provider-interface) + - [TypeSafe Jev API Provider](#typesafe-jev-api-provider) + - [Provider Registry and Configuration](#provider-registry-and-configuration) + - [Privacy-LLM Gate Routing](#privacy-llm-gate-routing) + - [Retry Policy](#retry-policy) + - [Python Usage](#python-usage) + - [Functional API](#functional-api) + - [Client API](#client-api) + - [Handling Failure](#handling-failure) + + + +# Tool: Typed Decision + +This directory documents the **typed-decision** capability contract (`contract:typed-decision`). +It defines the provider-agnostic interface that framework skills use when requesting structured, low-latency, typed decisions (discrete categorization, bounded scoring, and binary/null hypothesis evaluation) from specialized System 1 decision backends. + +Adopting projects configure their chosen backend via `MAGPIE_TYPED_DECISION_PROVIDER` (or credential variables such as `TYPESAFE_API_KEY`). +TypeSafe's Jev API (`api.typesafe.ai/v1/systemone`) serves as the initial reference provider backend. + +## What this tool provides + +Three core decision operations: + +| Operation | Inputs | Output | Use Case | +|---|---|---|---| +| `choice` | `prompt: str`, `options: list[str]` | `{"label": str, "confidence": float}` | Categorical classification (e.g. bug vs feature, reviewer routing, area dispatch). | +| `score` | `prompt: str`, `scale: tuple \| list \| num` | `{"value": num, "confidence": float}` | Numeric evaluation along a scale (e.g. severity rating 1-5, priority scoring 0-10). | +| `noul` | `prompt: str` | `{"probability": float}` | Null hypothesis / binary probability estimation (e.g. is this report actionable?). | + +## The Contract + +### `choice(prompt, options) -> {label, confidence}` + +Selects exactly one label from a non-empty list of candidate strings based on the prompt context. + +- **Inputs:** + - `prompt`: Natural language question, triage snippet, or context. + - `options`: Non-empty list of string labels (e.g. `["bug", "feature-request", "needs-info"]`). +- **Output:** + - A dictionary containing: + - `"label"`: The chosen option. + The returned label must strictly be an element of `options`. + If the provider returns an unexpected label, `TypedDecisionUnavailable` is raised. + - `"confidence"`: Floating-point certainty estimate in `[0.0, 1.0]`. + If confidence is missing, non-numeric, or outside `[0.0, 1.0]`, `TypedDecisionUnavailable` is raised. +- **Preconditions:** `options` must contain at least one string. + +### `score(prompt, scale) -> {value, confidence}` + +Evaluates a score along a specified numeric interval or scale. + +- **Inputs:** + - `prompt`: Text to evaluate. + - `scale`: Numeric range or ceiling (e.g. `(1, 5)`, `(0, 10)`, or upper-bound `10`). +- **Output:** + - A dictionary containing: + - `"value"`: Assigned score (integer or float within scale bounds). + If the provider returns a value outside the bounds of `scale`, `TypedDecisionUnavailable` is raised. + - `"confidence"`: Floating-point certainty estimate in `[0.0, 1.0]`. + If confidence is missing, non-numeric, or outside `[0.0, 1.0]`, `TypedDecisionUnavailable` is raised. + +### `noul(prompt) -> {probability}` + +Computes the semantic probability of a proposition or binary/null decision. + +- **Inputs:** + - `prompt`: Proposition to evaluate (e.g. "Does this issue describe a reproducible defect?"). +- **Output:** + - A dictionary containing: + - `"probability"`: Float in `[0.0, 1.0]` representing the probability of the proposition. + If probability is missing, non-numeric, or outside `[0.0, 1.0]`, `TypedDecisionUnavailable` is raised. + +## Fail-Open Contract + +The typed decision contract is strictly **fail-open**: + +> On missing configuration, timeout, network error, or provider error, every operation raises `TypedDecisionUnavailable` — **never a fabricated answer**. + +This guarantees that: +1. Skills never make decisions based on hallucinated defaults, fallacious random picks, or mock data. +2. Callers can catch `TypedDecisionUnavailable` cleanly and fall back to their primary LLM reasoning, heuristic classifiers, or interactive maintainer confirmation (in line with RFC-AI-0004 Principle 1: Human-in-the-loop and Principle 4: Conversational/correctable skills). + +## Architecture and Providers + +### Provider Interface + +All backends implement the `DecisionProvider` abstract base class defined in [`interface.py`](src/typed_decision/interface.py): + +```python +class DecisionProvider(ABC): + @property + @abstractmethod + def name(self) -> str: + """Provider identifier.""" + + @abstractmethod + def choice(self, prompt: str, options: list[str]) -> dict[str, Any]: + """Select a single option label from candidate options.""" + + @abstractmethod + def score(self, prompt: str, scale: Any) -> dict[str, Any]: + """Evaluate a score for prompt along given scale.""" + + @abstractmethod + def noul(self, prompt: str) -> dict[str, Any]: + """Evaluate null/binary decision probability.""" +``` + +### TypeSafe Jev API Provider + +The reference backend [`providers/jev.py`](src/typed_decision/providers/jev.py) connects to TypeSafe's Jev API: + +- **Endpoint:** `https://api.typesafe.ai/v1/systemone` +- **Pinned Model:** The model version is pinned to an explicit constant (`systemone-2026-06-01`), **never "latest"**, ensuring behavioral stability and reproducible benchmark evaluations (RFC-AI-0004 § Principle 3). +- **Zero Third-Party Dependencies:** Implemented using Python's standard `urllib.request` library. +- **Credential Storage:** Follows framework credential isolation (`AGENTS.md`): reads `TYPESAFE_API_KEY` (or `JEV_API_KEY`) from environment or from home directory storage (`~/.config/apache-magpie/typesafe.key`). +- **Construction Gate:** If credentials are missing at construction time, `JevProvider` raises `TypedDecisionUnavailable` immediately, signaling unavailability to the registry. + +### Provider Registry and Configuration + +Providers are resolved dynamically through [`registry.py`](src/typed_decision/registry.py): + +1. **`MAGPIE_TYPED_DECISION_PROVIDER`** (env var): Explicitly select provider name (e.g. `jev`). + If set to an unknown provider, raises `TypedDecisionUnavailable`. +2. **Default Behavior:** If `MAGPIE_TYPED_DECISION_PROVIDER` is unset: + - Defaults to `"jev"` **if configured** (credentials present). + - If unconfigured, reports unavailable and raises `TypedDecisionUnavailable`. + +## Privacy-LLM Gate Routing + +Per RFC-AI-0004 Principle 6 (Privacy by design) and [`tools/privacy-llm/`](../privacy-llm/), **every outbound prompt must pass through the privacy-llm gate before leaving the process**: + +1. Before an HTTP request is assembled, the prompt is routed through `enforce_privacy_gate(prompt, endpoint)`. +2. The gate strictly validates that the destination host is approved per [`tools/privacy-llm/models.md`](../privacy-llm/models.md) and denies by default: + - Local endpoints (`localhost`, `127.0.0.1`) are default-approved. + - Foundation endpoints (`*.apache.org`, excluding carve-outs) are default-approved. + - Third-party endpoints (`api.typesafe.ai`) require an explicit opt-in entry in `/privacy-llm.md` with non-empty `Data-residency contract` and non-placeholder `Approved-by` sign-offs. + - The gate verifies configuration using [`tools/privacy-llm/checker`](../privacy-llm/checker/). +3. If the gate rejects the destination, `TypedDecisionUnavailable` is raised and no network request occurs. +4. **Caller Redaction Responsibility:** Per [`tools/privacy-llm/wiring.md`](../privacy-llm/wiring.md), skills that process private source content (such as private security reports) must apply redaction via `pii-redact` immediately after fetching, prior to invoking downstream tools. + The typed-decision gate acts strictly as the network egress boundary. + +## Retry Policy + +To guard against transient network fluctuations without unbounded stalls: +- The HTTP client executes with a 30-second socket timeout (`DEFAULT_TIMEOUT_SECONDS = 30.0`). +- On timeout (`socket.timeout`, `TimeoutError`, or timeout-related `URLError`), the client executes **exactly one retry** following a backoff delay (`DEFAULT_BACKOFF_SECONDS = 0.5`). +- If the retry also times out (or on non-timeout network errors / HTTP 4xx/5xx responses), the client raises `TypedDecisionUnavailable`. + +## Python Usage + +### Functional API + +```python +from typed_decision import TypedDecisionUnavailable, choice, noul, score + +try: + decision = choice("Classify issue #412", ["bug", "feature", "question"]) + print(decision["label"], decision["confidence"]) +except TypedDecisionUnavailable: + # Fall back to agentic reasoning or interactive maintainer confirmation + ... +``` + +### Client API + +```python +from typed_decision import TypedDecisionClient, TypedDecisionUnavailable + +client = TypedDecisionClient() +result = client.score("Rate security impact of report", scale=(1, 5)) +print(result["value"], result["confidence"]) +``` + +### Handling Failure + +```python +try: + prob = noul("Is this report actionable?") +except TypedDecisionUnavailable as e: + # Fail-open: do not assume an answer; proceed with maintainer review + logger.info("Typed decision unavailable (%s); falling back to HITL", e) +``` diff --git a/tools/vendor-neutrality-score/src/vendor_neutrality_score/__init__.py b/tools/vendor-neutrality-score/src/vendor_neutrality_score/__init__.py index 1e66e9c9f..d80ddf90c 100644 --- a/tools/vendor-neutrality-score/src/vendor_neutrality_score/__init__.py +++ b/tools/vendor-neutrality-score/src/vendor_neutrality_score/__init__.py @@ -98,6 +98,10 @@ VENDOR_BACKED, "Vulnerability database / cross-reference alias lookup (OSV.dev / NVD)", ), + "contract:typed-decision": ( + VENDOR_BACKED, + "Provider-agnostic typed decision backend (Choice / Score / Noul)", + ), } # Which capability *contract* a skill actually invokes, keyed by diff --git a/uv.lock b/uv.lock index 78129778e..7da2b674e 100644 --- a/uv.lock +++ b/uv.lock @@ -54,6 +54,7 @@ members = [ "spec-status-index", "spec-validator", "symlink-lint", + "typed-decision", "vendor-neutrality-score", "vetted-ops", "vulnogram-api", @@ -2146,6 +2147,25 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/19/97/56608b2249fe206a67cd573bc93cd9896e1efb9e98bce9c163bcdc704b88/truststore-0.10.4-py3-none-any.whl", hash = "sha256:adaeaecf1cbb5f4de3b1959b42d41f6fab57b2b1666adb59e89cb0b53361d981", size = 18660, upload-time = "2025-08-12T18:49:01.46Z" }, ] +[[package]] +name = "typed-decision" +version = "0.1.0" +source = { editable = "tools/typed-decision" } +dependencies = [ + { name = "checker" }, +] + +[package.dev-dependencies] +dev = [ + { name = "magpie-dev" }, +] + +[package.metadata] +requires-dist = [{ name = "checker", editable = "tools/privacy-llm/checker" }] + +[package.metadata.requires-dev] +dev = [{ name = "magpie-dev", editable = "tools/dev" }] + [[package]] name = "typing-extensions" version = "4.16.0"