You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@magpie.apache.org by po...@apache.org on 2026/10/04 00:17:04 UTC
(magpie) branch main updated: feat(tools): provider-agnostic typed decision contract with Jev API backend (#1402)
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/magpie.git
The following commit(s) were added to refs/heads/main by this push:
new 0b8a1f80 feat(tools): provider-agnostic typed decision contract with Jev API backend (#1402)
0b8a1f80 is described below
commit 0b8a1f8052f51f950c9ee69a6aeeb3e85ec1c31d
Author: Arnav <im...@gmail.com>
AuthorDate: Sun Oct 4 05:46:59 2026 +0530
feat(tools): provider-agnostic typed decision contract with Jev API backend (#1402)
* feat(tools): provider-agnostic typed decision contract with Jev API backend
Add the provider-agnostic typed decision tool contract (contract:typed-decision)
under tools/typed-decision/ supporting Choice, Score, and Noul decision
operations, with TypeSafe's Jev API (systemone) as the initial backend.
Key features and invariants:
- Contract operations:
* choice(prompt, options: list[str]) -> {label, confidence}
* score(prompt, scale) -> {value, confidence}
* noul(prompt) -> {probability}
- Fail-open contract: On missing configuration, timeout, network error,
or provider error, raise TypedDecisionUnavailable — never hallucinate or
fabricate an answer.
- Zero external HTTP dependencies: Standard library only (urllib.request).
- Pinned model version: Pinned to constant JEV_MODEL = "systemone-2026-06-01",
never "latest".
- Privacy-LLM gate: All outbound prompts are routed through the tools/privacy-llm/
gate check and redaction pipeline before leaving the process.
- Credential resolution: TYPESAFE_API_KEY (or fallback JEV_API_KEY) and
~/.config/apache-magpie/typesafe.key.
- Network resilience: Exactly one retry with backoff on timeout before fail-open.
- Workspace integration: Added to pyproject.toml workspace members, Axis 2
capabilities in docs/labels-and-capabilities.md, validator TOOL_CAPABILITIES,
and CONTRACT_POLICY in tools/vendor-neutrality-score.
- Pytest suite covering all operations, timeout retries, error conditions,
privacy gate routing, and registration.
Fixes #1370
Generated-by: Antigravity
* fix(typed-decision): address CodeQL alerts on ellipsis and empty except
- In interface.py, remove redundant `...` (ellipsis) expression statements
following docstrings in abstract method definitions, resolving CodeQL
"Statement has no effect" (py/useless-expression) alerts.
- In providers/jev.py, add an explanatory comment inside the `except OSError:`
clause when scanning candidate key paths, resolving the CodeQL
"Empty except" (py/empty-except) alert.
Refs #1370
Generated-by: Antigravity
* fix(typed-decision): address maintainer review feedback on privacy gate and validation
- Privacy gate: Enforce deny-by-default for third-party endpoints when no privacy-llm.md config is found. Integrate tools/privacy-llm/checker canonical parser (parse_config, _approve_by_default_rules, _approve_by_opt_in) with strict opt-in verification (non-empty Data-residency contract, non-placeholder Approved-by sign-offs).
- Responsibility split: Clarify in tool.md and README.md that callers (skills) are responsible for redacting PII before calling typed-decision, while the privacy gate enforces network egress boundary.
- Jev provider strict validation: Raise TypedDecisionUnavailable on missing or non-numeric confidence (never default to 1.0), validate confidence and probability in [0.0, 1.0], validate label in options, and validate score scale bounds upfront. Wrap all response JSON extraction in try/except to prevent KeyError/TypeError leaks.
- Cleanup: Remove dead leftover file tools/typed-decision/providers/jev.py.
- Registry deduplication: Simplify _is_jev_configured() to delegate directly to _resolve_api_key() is not None.
- Documentation: Apply Semantic Line Breaks (SemBr) and remove duplicate SPDX headers in README.md and tool.md. Add tools/typed-decision to docs/adapters/registry.md and docs/vendor-neutrality.md (Table 1, Table 2, and extension points).
- Test suite: Add comprehensive unit tests covering deny-by-default behavior, opt-in requirements, scale pre-validation, and response validation.
Refs #1370
Refs #1402
Generated-by: Antigravity
* fix(typed-decision): address workspace-mypy in prek and empty except CodeQL alert
- Remove unused type: ignore comments from checker imports in privacy.py
- Add mypy override module checker.* with ignore_missing_imports in tools/typed-decision/pyproject.toml so workspace-mypy passes cleanly under prek.
- Remove redundant empty try/except block in JevProvider.score() response parsing, resolving CodeQL alert #97.
Refs #1370
Refs #1402
Generated-by: Antigravity
* fix(typed-decision): address maintainer review on privacy binding, dependency wiring, and scale validation
- Privacy gate: Bind opt-in checks to URL/host and only apply provider name label when endpoint equals DEFAULT_ENDPOINT, preventing name-only opt-ins from approving arbitrary destination hosts
- Checker dependency: Declare checker as a workspace dependency in tools/typed-decision/pyproject.toml and root pyproject.toml, add public check_endpoint() helper in checker.check and py.typed marker in checker package, and drop sys.path hack and mypy override
- Gate hook: Remove production custom gate hook and ensure deny-by-default cannot be bypassed in-process
- Scale validation: Pre-validate scalar and tuple scale arguments in JevProvider.score(), normalize scalars to (0, n), and bounds-check returned scores across all scale formats
- Logging: Drop full response repr ({resp!r}) from error messages to avoid echoing prompt content into logs
- Documentation & registries: Cite dedicated issue #1431 for the local-model typed-decision backend in docs/adapters/registry.md and docs/vendor-neutrality.md Table 2, and fix reference adapter column formatting
- Tests: Add unit tests for name-only opt-in host binding, scalar scale normalization and bounds checking, and clean up strict gate environment variables
Refs #1370
Refs #1402
Refs #1431
Generated-by: Antigravity
* style: update vendor-neutrality score to 10/13 (77%) and trim trailing newlines
Generated-by: Antigravity
* style(docs): fix markdownlint MD012 multiple blanks in vendor-neutrality.md
Generated-by: Antigravity
* fix(typed-decision): enforce host-bound privacy check, reject userinfo/fragments, and validate finite scales
Address maintainer review feedback on PR #1402:
- Host-bind endpoint privacy check in tools/privacy-llm/checker:
- Reject URLs containing userinfo (@) or fragments (#)
- Drop Claude Code free-text matching for URL endpoint checks
- Require exact host matching for opt-in URLs: host_of(endpoint) == host_of(opt_url)
- Restrict name-only opt-ins strictly to default_endpoint matching
- Update tools/typed-decision/src/typed_decision/privacy.py:
- Pass default_endpoint=DEFAULT_ENDPOINT to check_endpoint
- Clarify host-bound validation in docstring
- Deduplicate DEFAULT_ENDPOINT: import from privacy.py into providers/jev.py
- Add math.isfinite checks on scalar and interval scale bounds and returned score values
- Add license field and clean up blank lines in pyproject.toml
- Make no-config tests hermetic using monkeypatch.chdir(tmp_path)
- Add comprehensive test coverage for security attack URLs and non-finite scales
Generated-by: Antigravity
* fix(privacy-llm,typed-decision): fix host extraction bypasses, response type-confusion, and test encoding
- checker/check.py: validate candidate hostnames with _is_valid_hostname to reject decimal/version tokens (e.g. 3.5 Sonnet) and require non-empty raw_desc for name-only opt-in matching
- typed_decision/privacy.py: always include provider name in raw_desc
- typed_decision/providers/jev.py: enforce strict non-boolean checks on response fields and require dict envelopes for 'result' and 'decision'
- checker/tests: specify encoding='utf-8' in _write helper to prevent Windows cp1252 decode errors
- Add regression tests covering all bypasses and error envelope cases
* test(privacy-llm,typed-decision): align regression test cases with review specifications
* fixup: tighten name-only opt-in matching and close review follow-ups
- check.py: match name-only opt-ins on whole words against the provider
description with URLs stripped, so a short opt-in name such as
"AI (internal)" can no longer match inside https://api.typesafe.ai/...
and approve a provider nobody opted in to.
- check.py: stop splitting bare opt-in hosts on '-', so
bedrock-runtime.eu-central-1.amazonaws.com is extracted whole.
- Tests for both, plus HOME/cwd isolation in the typed-decision autouse
fixture and a test for the ~/.config/apache-magpie/typesafe.key path.
- Restore the checker package docstring above the new re-exports.
- README: how the provider behaves and is enabled under the secure
agent setup.
Generated-by: Claude Opus 5
---------
Co-authored-by: Jarek Potiuk <po...@apache.org>
---
.github/labeler.yml | 6 +
docs/adapters/registry.md | 1 +
docs/labels-and-capabilities.md | 2 +
docs/vendor-neutrality.md | 10 +-
pyproject.toml | 3 +
tools/privacy-llm/checker/src/checker/__init__.py | 11 +
tools/privacy-llm/checker/src/checker/check.py | 213 +++++
tools/privacy-llm/checker/src/checker/py.typed | 1 +
tools/privacy-llm/checker/tests/test_check.py | 253 ++++-
tools/privacy-llm/checker/tests/test_config.py | 2 +-
.../src/skill_and_tool_validator/__init__.py | 1 +
tools/typed-decision/README.md | 66 ++
tools/typed-decision/pyproject.toml | 85 ++
.../typed-decision/src/typed_decision/__init__.py | 109 +++
.../src/typed_decision/exceptions.py} | 22 +-
.../typed-decision/src/typed_decision/interface.py | 85 ++
tools/typed-decision/src/typed_decision/privacy.py | 76 ++
.../src/typed_decision/providers}/__init__.py | 14 +-
.../src/typed_decision/providers/jev.py | 481 ++++++++++
.../typed-decision/src/typed_decision/registry.py | 77 ++
.../checker => typed-decision/tests}/__init__.py | 14 -
tools/typed-decision/tests/test_typed_decision.py | 1010 ++++++++++++++++++++
tools/typed-decision/tool.md | 199 ++++
.../src/vendor_neutrality_score/__init__.py | 4 +
uv.lock | 20 +
25 files changed, 2726 insertions(+), 39 deletions(-)
diff --git a/.github/labeler.yml b/.github/labeler.yml
index 47eda92a..b0020074 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 b85d37aa..b30633a8 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](a [...]
| 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 6f5ff38c..5c4854eb 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 cf90995a..05952420 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.
<!-- BEGIN vendor-neutrality-score — generated by `uv run --project tools/vendor-neutrality-score vendor-neutrality-score --markdown`; do not edit by hand -->
-**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 beebf78c..f9e4939b 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 c83a5dcc..9402d96d 100644
--- a/tools/privacy-llm/checker/src/checker/__init__.py
+++ b/tools/privacy-llm/checker/src/checker/__init__.py
@@ -27,4 +27,15 @@ stderr explanation otherwise.
Skills shell out to this command at Step 0 (pre-flight) when they
may read ``<private-list>`` 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 1b45e7e9..3e210d52 100644
--- a/tools/privacy-llm/checker/src/checker/check.py
+++ b/tools/privacy-llm/checker/src/checker/check.py
@@ -35,12 +35,15 @@ from __future__ import annotations
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"(?<!\w){re.escape(candidate)}(?!\w)", desc_lc):
+ return True
+ return False
+
+
+def _approve_endpoint_by_opt_in(
+ entry: LLMEntry,
+ ep_host: str,
+ opt_in: list[OptInEntry],
+ *,
+ default_endpoint: str | None = None,
+ raw_desc: str | None = None,
+) -> 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 <project-config>/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 ``<project-config>/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 00000000..1242d432
--- /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 7f23b387..0c9c5ad7 100644
--- a/tools/privacy-llm/checker/tests/test_check.py
+++ b/tools/privacy-llm/checker/tests/test_check.py
@@ -27,7 +27,7 @@ from checker.config import LLMEntry, OptInEntry, ParsedConfig, parse_config
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 017945cc..a9089735 100644
--- a/tools/privacy-llm/checker/tests/test_config.py
+++ b/tools/privacy-llm/checker/tests/test_config.py
@@ -33,7 +33,7 @@ from checker.config import (
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 e083ca65..6fe7bd2f 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 @@ TOOL_CAPABILITIES = {
"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 00000000..d8d56ad4
--- /dev/null
+++ b/tools/typed-decision/README.md
@@ -0,0 +1,66 @@
+<!-- SPDX-License-Identifier: Apache-2.0
+ https://www.apache.org/licenses/LICENSE-2.0 -->
+
+<!-- START doctoc generated TOC please keep comment here to allow auto update -->
+<!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->
+**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)
+
+<!-- END doctoc generated TOC please keep comment here to allow auto update -->
+
+# `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 `<project-config>/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 00000000..a0052b37
--- /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 00000000..614cd26c
--- /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/privacy-llm/checker/src/checker/__init__.py b/tools/typed-decision/src/typed_decision/exceptions.py
similarity index 56%
copy from tools/privacy-llm/checker/src/checker/__init__.py
copy to tools/typed-decision/src/typed_decision/exceptions.py
index c83a5dcc..a837cdb5 100644
--- a/tools/privacy-llm/checker/src/checker/__init__.py
+++ b/tools/typed-decision/src/typed_decision/exceptions.py
@@ -14,17 +14,17 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Approved-LLM gate-check for the apache-magpie privacy-llm tool.
-Single console script: ``privacy-llm-check`` (:mod:`checker.check`).
+"""Exceptions for typed decision operations."""
-Parses ``<project-config>/privacy-llm.md``, extracts the active LLM
-stack and the opt-in third-party-endpoint registry, applies the
-approval rules from ``tools/privacy-llm/models.md``, and exits 0
-if every entry in the active stack is approved — exit 1 with a
-stderr explanation otherwise.
+from __future__ import annotations
-Skills shell out to this command at Step 0 (pre-flight) when they
-may read ``<private-list>`` content. The contract is documented
-in ``tools/privacy-llm/wiring.md``.
-"""
+
+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 00000000..f1fdb181
--- /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 00000000..9a17e027
--- /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 `<project-config>/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/privacy-llm/checker/src/checker/__init__.py b/tools/typed-decision/src/typed_decision/providers/__init__.py
similarity index 56%
copy from tools/privacy-llm/checker/src/checker/__init__.py
copy to tools/typed-decision/src/typed_decision/providers/__init__.py
index c83a5dcc..0ec4af08 100644
--- a/tools/privacy-llm/checker/src/checker/__init__.py
+++ b/tools/typed-decision/src/typed_decision/providers/__init__.py
@@ -14,17 +14,9 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Approved-LLM gate-check for the apache-magpie privacy-llm tool.
-Single console script: ``privacy-llm-check`` (:mod:`checker.check`).
+"""Typed decision backend providers."""
-Parses ``<project-config>/privacy-llm.md``, extracts the active LLM
-stack and the opt-in third-party-endpoint registry, applies the
-approval rules from ``tools/privacy-llm/models.md``, and exits 0
-if every entry in the active stack is approved — exit 1 with a
-stderr explanation otherwise.
+from typed_decision.providers.jev import JevProvider
-Skills shell out to this command at Step 0 (pre-flight) when they
-may read ``<private-list>`` content. The contract is documented
-in ``tools/privacy-llm/wiring.md``.
-"""
+__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 00000000..b0857549
--- /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 00000000..0f797322
--- /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/privacy-llm/checker/src/checker/__init__.py b/tools/typed-decision/tests/__init__.py
similarity index 56%
copy from tools/privacy-llm/checker/src/checker/__init__.py
copy to tools/typed-decision/tests/__init__.py
index c83a5dcc..13a83393 100644
--- a/tools/privacy-llm/checker/src/checker/__init__.py
+++ b/tools/typed-decision/tests/__init__.py
@@ -14,17 +14,3 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Approved-LLM gate-check for the apache-magpie privacy-llm tool.
-
-Single console script: ``privacy-llm-check`` (:mod:`checker.check`).
-
-Parses ``<project-config>/privacy-llm.md``, extracts the active LLM
-stack and the opt-in third-party-endpoint registry, applies the
-approval rules from ``tools/privacy-llm/models.md``, and exits 0
-if every entry in the active stack is approved — exit 1 with a
-stderr explanation otherwise.
-
-Skills shell out to this command at Step 0 (pre-flight) when they
-may read ``<private-list>`` content. The contract is documented
-in ``tools/privacy-llm/wiring.md``.
-"""
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 00000000..5b01c206
--- /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. <pmc-member-initials>) 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: <pmc-member-initials> 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 00000000..8aca1d2b
--- /dev/null
+++ b/tools/typed-decision/tool.md
@@ -0,0 +1,199 @@
+<!-- SPDX-License-Identifier: Apache-2.0
+ https://www.apache.org/licenses/LICENSE-2.0 -->
+
+<!-- START doctoc generated TOC please keep comment here to allow auto update -->
+<!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->
+**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)
+
+<!-- END doctoc generated TOC please keep comment here to allow auto update -->
+
+# 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 `<project-config>/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 1e66e9c9..d80ddf90 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 @@ CONTRACT_POLICY: dict[str, tuple[str, str]] = {
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 78129778..7da2b674 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"