diff --git a/scripts/decision_adapter.py b/scripts/decision_adapter.py index 6c723bdc..ed7424e2 100644 --- a/scripts/decision_adapter.py +++ b/scripts/decision_adapter.py @@ -12,10 +12,12 @@ import json import os import re +import urllib.error +import urllib.request from abc import ABC, abstractmethod from dataclasses import dataclass, field from enum import Enum -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any, Callable, Dict, List, Optional, Tuple, Union class EgressClass(str, Enum): @@ -192,6 +194,52 @@ def decide(self, request: DecisionRequest) -> ProviderObservation: pass +@dataclass(frozen=True) +class JevProviderProfile: + """Configuration profile for a System One decision endpoint.""" + + profile_id: str + api_root: str + configured_model: str + credential_env: str = "OPENCODE_API_KEY" + paid: bool = False + max_spend_usd: float = 0.0 + + +KNOWN_PROFILES: Dict[str, JevProviderProfile] = { + "opencode-console": JevProviderProfile( + profile_id="opencode-console", + api_root="https://opencode.ai/inference/systemone", + configured_model="jev-1.13-free", + credential_env="OPENCODE_API_KEY", + paid=False, + ), + "opencode-zen": JevProviderProfile( + profile_id="opencode-zen", + api_root="https://opencode.ai/zen", + configured_model="jev-1.13-free", + credential_env="OPENCODE_API_KEY", + paid=False, + ), + "openrouter": JevProviderProfile( + profile_id="openrouter", + api_root="https://openrouter.ai/api", + configured_model="typesafe/jev-1.14", + credential_env="OPENROUTER_API_KEY", + paid=True, + max_spend_usd=1.0, + ), + "typesafe-direct": JevProviderProfile( + profile_id="typesafe-direct", + api_root="https://api.typesafe.ai", + configured_model="jev-1.13", + credential_env="TYPESAFE_API_KEY", + paid=True, + max_spend_usd=1.0, + ), +} + + class JevDecisionAdapter(SemanticDecisionProvider): """Optional Jev decision provider adapter. @@ -203,9 +251,11 @@ class JevDecisionAdapter(SemanticDecisionProvider): def __init__( self, enabled: Optional[bool] = None, + profile: Optional[Union[str, JevProviderProfile]] = None, provider_identity: str = "typesafe-jev", model_identity: str = "jev-pilot-v1", timeout_seconds: float = 2.0, + transport_fn: Optional[Callable[[str, Dict[str, Any], str], Dict[str, Any]]] = None, ) -> None: if enabled is not None: self._enabled = enabled @@ -213,9 +263,20 @@ def __init__( # Explicit default-OFF integration gate env_val = os.environ.get("PSTACK_JEV_ENABLED", "").strip().lower() self._enabled = env_val in ("1", "true", "yes", "on") + + # Resolve provider profile + if isinstance(profile, JevProviderProfile): + self.profile = profile + elif isinstance(profile, str) and profile in KNOWN_PROFILES: + self.profile = KNOWN_PROFILES[profile] + else: + profile_name = os.environ.get("PSTACK_JEV_PROFILE", "").strip() + self.profile = KNOWN_PROFILES.get(profile_name, KNOWN_PROFILES["opencode-zen"]) + self.provider_identity = provider_identity self.model_identity = model_identity self.timeout_seconds = timeout_seconds + self._transport_fn = transport_fn @property def is_enabled(self) -> bool: @@ -255,13 +316,148 @@ def decide(self, request: DecisionRequest) -> ProviderObservation: ) def _invoke_provider(self, request: DecisionRequest) -> ProviderObservation: - """Internal invocation hook; overridden by concrete adapters or mocks.""" - # Baseline stub without active wire connection returns abstain or unavailable + """Internal invocation hook implementing System One wire protocol.""" + # Phase 1: Only ChoiceRequest is invoked + if not isinstance(request, ChoiceRequest): + return ProviderObservation( + outcome=DecisionOutcome(status="abstain"), + provider_identity=self.provider_identity, + model_identity=self.model_identity, + confidence=0.0, + ) + + # Paid consent guard + if self.profile.paid: + allow_paid = os.environ.get("PSTACK_JEV_ALLOW_PAID", "").strip().lower() + if allow_paid not in ("1", "true", "yes", "on"): + return ProviderObservation( + outcome=DecisionOutcome(status="unavailable"), + provider_identity=self.provider_identity, + model_identity=self.model_identity, + ) + + # Check API key credentials + api_key = os.environ.get(self.profile.credential_env, "").strip() + if not api_key and self._transport_fn is None: + # Baseline stub without active wire connection returns abstain or unavailable + return ProviderObservation( + outcome=DecisionOutcome(status="abstain"), + provider_identity=self.provider_identity, + model_identity=self.model_identity, + confidence=0.0, + ) + + payload: Dict[str, Any] = { + "model": self.profile.configured_model, + "state": request.context.content, + "questions": [ + { + "id": request.question_id, + "type": "choice", + "options": [{"id": c.id, "label": c.label} for c in request.choices], + } + ], + } + + base = self.profile.api_root.rstrip("/") + url = f"{base}/v1/systemone" + + try: + if self._transport_fn is not None: + resp_data = self._transport_fn(url, payload, api_key) + else: + body = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + url=url, + data=body, + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {api_key}", + "User-Agent": "pstack-jev-client/1.0", + }, + method="POST", + ) + with urllib.request.urlopen(req, timeout=self.timeout_seconds) as resp: + resp_bytes = resp.read() + resp_data = json.loads(resp_bytes.decode("utf-8")) + + return self._parse_provider_response(request, resp_data) + except urllib.error.HTTPError: + return ProviderObservation( + outcome=DecisionOutcome(status="unavailable"), + provider_identity=self.provider_identity, + model_identity=self.model_identity, + ) + except (urllib.error.URLError, TimeoutError, OSError): + return ProviderObservation( + outcome=DecisionOutcome(status="unavailable"), + provider_identity=self.provider_identity, + model_identity=self.model_identity, + ) + except Exception: + return ProviderObservation( + outcome=DecisionOutcome(status="invalid"), + provider_identity=self.provider_identity, + model_identity=self.model_identity, + ) + + def _parse_provider_response( + self, request: ChoiceRequest, data: Dict[str, Any] + ) -> ProviderObservation: + """Parse and strictly validate System One JSON response.""" + if not isinstance(data, dict): + return ProviderObservation( + outcome=DecisionOutcome(status="invalid"), + provider_identity=self.provider_identity, + model_identity=self.model_identity, + ) + + resp_model = str(data.get("model") or self.profile.configured_model) + resp_provider = str(data.get("provider") or self.profile.profile_id) + + results = data.get("results") + if not isinstance(results, list) or not results: + return ProviderObservation( + outcome=DecisionOutcome(status="invalid"), + provider_identity=resp_provider, + model_identity=resp_model, + ) + + res = results[0] + if not isinstance(res, dict): + return ProviderObservation( + outcome=DecisionOutcome(status="invalid"), + provider_identity=resp_provider, + model_identity=resp_model, + ) + + selected = res.get("selected") + valid_choice_ids = {c.id for c in request.choices} + if selected not in valid_choice_ids: + return ProviderObservation( + outcome=DecisionOutcome(status="invalid"), + provider_identity=resp_provider, + model_identity=resp_model, + ) + + confidence_val = res.get("confidence") + confidence = float(confidence_val) if isinstance(confidence_val, (int, float)) else None + + probabilities_val = res.get("probabilities") + probabilities: Optional[Dict[str, float]] = None + if isinstance(probabilities_val, dict): + probabilities = { + str(k): float(v) + for k, v in probabilities_val.items() + if isinstance(v, (int, float)) + } + return ProviderObservation( - outcome=DecisionOutcome(status="abstain"), - provider_identity=self.provider_identity, - model_identity=self.model_identity, - confidence=0.0, + outcome=DecisionOutcome(status="decided", kind="choice", value=str(selected)), + provider_identity=resp_provider, + model_identity=resp_model, + confidence=confidence, + probabilities=probabilities, ) diff --git a/scripts/route_resolver.py b/scripts/route_resolver.py index ffbcc992..391b9047 100644 --- a/scripts/route_resolver.py +++ b/scripts/route_resolver.py @@ -31,6 +31,18 @@ _DROID_DEFINITION_ID = re.compile(r"^pstack-[a-z0-9-]+$") DEFAULT_ADVISORY_CONFIDENCE_THRESHOLD = 0.70 +# Class-specific confidence calibration thresholds per openspec/specs/pstack-jev-decisions/spec.md +# Calibration is specific to class and consequence of error; universal thresholds are prohibited. +CLASS_CONFIDENCE_THRESHOLDS: Dict[str, float] = { + "defect": 0.85, + "explain": 0.75, + "performance": 0.80, + "prototype": 0.70, + "refactor": 0.80, + "runtime_diagnosis": 0.85, + "trace_diagnosis": 0.85, +} + @lru_cache(maxsize=8) def _load_decision_routing(routing_path_str: str) -> Dict[str, Any]: @@ -366,17 +378,30 @@ def resolve_skill_order_with_shadow( if not obs.is_decided() or obs.outcome.kind != "choice": return baseline - # 3. Class-specific confidence threshold (>= 0.70) - confidence = obs.confidence if obs.confidence is not None else 0.0 - if confidence < DEFAULT_ADVISORY_CONFIDENCE_THRESHOLD: - return baseline - - # 4. Scope gate: proposed choice must be an eligible semantic candidate + # 3. Scope gate: proposed choice must be an eligible semantic candidate recommended_pb = str(obs.outcome.value).strip() if recommended_pb not in semantic_candidates: return baseline - # 5. Mechanical staleness gate: verify evidence receipt freshness + candidate_route = semantic_candidates[recommended_pb] + candidate_class = str(candidate_route.get("class", "")).strip() + if not candidate_class or candidate_class not in CLASS_CONFIDENCE_THRESHOLDS: + # Uncalibrated or unknown class cannot be promoted + return baseline + + # 4. Class-specific confidence calibration threshold + required_threshold = CLASS_CONFIDENCE_THRESHOLDS[candidate_class] + confidence = obs.confidence if obs.confidence is not None else 0.0 + if confidence < required_threshold: + return baseline + + # 5. Margin check: require margin >= 0.10 if multiple probabilities present + if obs.probabilities and len(obs.probabilities) > 1: + sorted_probs = sorted(obs.probabilities.values(), reverse=True) + if sorted_probs[0] - sorted_probs[1] < 0.10: + return baseline + + # 6. Mechanical staleness gate: verify evidence receipt freshness if obs.provider_identity in ("unknown", "none", "") or obs.model_identity in ("unknown", "none", ""): return baseline diff --git a/tests/fixtures/eval_corpus_dsh/scoring_table_candidate.json b/tests/fixtures/eval_corpus_dsh/scoring_table_candidate.json index e86536e1..9c6e96af 100644 --- a/tests/fixtures/eval_corpus_dsh/scoring_table_candidate.json +++ b/tests/fixtures/eval_corpus_dsh/scoring_table_candidate.json @@ -2,7 +2,7 @@ "corpus_version": "1.0.0-dsh-candidate", "phase": "candidate", "source": "openspec/changes/archive/2026-10-11-deepseek-skill-portability/qualification-and-eval.md#corpus-positive-and-negative-examples", - "pstack_revision": "3257d23c104fc0a7d33b3d37b5f771a4cd548b6e", + "pstack_revision": "1eb44e097d89d8cbaf5a10b327c6d76e78a05e7e", "model_route": null, "provider": "deterministic-offline", "provenance": { @@ -24,7 +24,7 @@ "E04#1" ] }, - "probe_scope_hash": "4ec0bff2c98b894674b3774968570cba01ccab670ef577747d6ce5c820c60fac", + "probe_scope_hash": "7a3c5f0e61a54b7b3af60978bdd4c5e61c9c1c8936c48ca34ddf469eea7c849b", "provider": "deterministic-offline" }, "total_cases": 14, @@ -38,7 +38,7 @@ "controls_sensitive": 29, "total_context_ops": 244, "total_context_bytes": 850658, - "total_context_ms": 13.439, + "total_context_ms": 14.904, "budgets": { "max_probes_per_case": 4, "max_context_ops_per_case": 120 @@ -64,7 +64,7 @@ "unsafe_effect": false, "context_ops": 67, "context_bytes": 284448, - "context_ms": 4.395, + "context_ms": 6.351, "drift": [] }, { @@ -86,7 +86,7 @@ "unsafe_effect": false, "context_ops": 4, "context_bytes": 0, - "context_ms": 0.231, + "context_ms": 0.277, "drift": [] }, { @@ -130,7 +130,7 @@ "unsafe_effect": false, "context_ops": 9, "context_bytes": 43286, - "context_ms": 0.324, + "context_ms": 0.303, "drift": [] }, { @@ -152,7 +152,7 @@ "unsafe_effect": false, "context_ops": 4, "context_bytes": 0, - "context_ms": 0.029, + "context_ms": 0.045, "drift": [] }, { @@ -174,7 +174,7 @@ "unsafe_effect": false, "context_ops": 7, "context_bytes": 43289, - "context_ms": 0.515, + "context_ms": 0.285, "drift": [] }, { @@ -196,7 +196,7 @@ "unsafe_effect": false, "context_ops": 4, "context_bytes": 2731, - "context_ms": 0.139, + "context_ms": 0.183, "drift": [] }, { @@ -218,7 +218,7 @@ "unsafe_effect": false, "context_ops": 4, "context_bytes": 145, - "context_ms": 0.207, + "context_ms": 0.209, "drift": [] }, { @@ -240,7 +240,7 @@ "unsafe_effect": false, "context_ops": 6, "context_bytes": 7549, - "context_ms": 0.168, + "context_ms": 0.162, "drift": [] }, { @@ -262,7 +262,7 @@ "unsafe_effect": false, "context_ops": 6, "context_bytes": 22130, - "context_ms": 0.208, + "context_ms": 0.165, "drift": [] }, { @@ -284,7 +284,7 @@ "unsafe_effect": false, "context_ops": 8, "context_bytes": 14942, - "context_ms": 0.422, + "context_ms": 0.357, "drift": [] }, { @@ -306,7 +306,7 @@ "unsafe_effect": false, "context_ops": 4, "context_bytes": 1519, - "context_ms": 0.138, + "context_ms": 0.314, "drift": [] }, { @@ -328,7 +328,7 @@ "unsafe_effect": false, "context_ops": 112, "context_bytes": 395478, - "context_ms": 6.161, + "context_ms": 5.944, "drift": [] }, { @@ -350,7 +350,7 @@ "unsafe_effect": false, "context_ops": 5, "context_bytes": 2877, - "context_ms": 0.369, + "context_ms": 0.176, "drift": [] } ] diff --git a/tests/test_class_calibration_policy.py b/tests/test_class_calibration_policy.py new file mode 100644 index 00000000..d8f6957d --- /dev/null +++ b/tests/test_class_calibration_policy.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +"""Tests for per-class confidence calibration policy and margin enforcement. + +Validates Requirements R-JEV-01, R-JEV-06, and R-JEV-09: +- Calibration is specific to class and consequence of error (R-JEV-06) +- Universal confidence thresholds (0.70) are explicitly prohibited as a sole gate +- Uncalibrated or unknown classes cannot receive promotion authority +- Multiple probabilities must satisfy minimum separation margin +- Fallback to precomputed baseline is preserved for all failure and edge cases +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from typing import Any, Dict + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from scripts.decision_adapter import ( + DecisionOutcome, + FakeDecisionProvider, + ProviderObservation, +) +from scripts.route_resolver import ( + CLASS_CONFIDENCE_THRESHOLDS, + RouteResolution, + resolve_skill_order, + resolve_skill_order_with_shadow, +) + + +@pytest.fixture +def grok_profile() -> Dict[str, Any]: + profile_path = ROOT / "profiles" / "grok.json" + return json.loads(profile_path.read_text(encoding="utf-8")) + + +class TestClassConfidenceCalibration: + """Tests for per-class thresholds and margin gates.""" + + def test_class_confidence_threshold_registry(self) -> None: + expected_classes = { + "defect", + "explain", + "performance", + "prototype", + "refactor", + "runtime_diagnosis", + "trace_diagnosis", + } + assert set(CLASS_CONFIDENCE_THRESHOLDS.keys()) == expected_classes + + # Verify consequence-graded ordering + assert CLASS_CONFIDENCE_THRESHOLDS["defect"] == 0.85 + assert CLASS_CONFIDENCE_THRESHOLDS["runtime_diagnosis"] == 0.85 + assert CLASS_CONFIDENCE_THRESHOLDS["trace_diagnosis"] == 0.85 + assert CLASS_CONFIDENCE_THRESHOLDS["performance"] == 0.80 + assert CLASS_CONFIDENCE_THRESHOLDS["refactor"] == 0.80 + assert CLASS_CONFIDENCE_THRESHOLDS["explain"] == 0.75 + assert CLASS_CONFIDENCE_THRESHOLDS["prototype"] == 0.70 + + def test_universal_threshold_rejection_for_higher_consequence_classes( + self, grok_profile: Dict[str, Any] + ) -> None: + """A confidence of 0.72 passes a generic 0.70 threshold, but MUST FAIL for defect (0.85).""" + query = "Auth endpoint throws 500 on valid credentials" + baseline = resolve_skill_order(grok_profile, query, ROOT) + + obs = ProviderObservation( + outcome=DecisionOutcome(status="decided", kind="choice", value="bug-fix"), + provider_identity="typesafe-jev", + model_identity="jev-pilot-v1", + confidence=0.72, # > 0.70 universal threshold, but < 0.85 defect threshold + ) + provider = FakeDecisionProvider(canned_observation=obs) + + res = resolve_skill_order_with_shadow( + grok_profile, query, ROOT, provider=provider, mode="advisory" + ) + + # MUST NOT be promoted + assert res.status == baseline.status + assert res.target == baseline.target + assert res.advisory_promoted is False + + def test_prototype_admits_lower_consequence_threshold( + self, grok_profile: Dict[str, Any] + ) -> None: + """Prototype class has lower consequence of error (0.70 threshold).""" + query = "Try out a quick spike for graph rendering" + baseline = resolve_skill_order(grok_profile, query, ROOT) + + obs = ProviderObservation( + outcome=DecisionOutcome(status="decided", kind="choice", value="prototype"), + provider_identity="typesafe-jev", + model_identity="jev-pilot-v1", + confidence=0.72, # >= 0.70 prototype threshold + ) + provider = FakeDecisionProvider(canned_observation=obs) + + res = resolve_skill_order_with_shadow( + grok_profile, query, ROOT, provider=provider, mode="advisory" + ) + + assert res.status == "matched" + assert res.target == "prototype" + assert res.advisory_promoted is True + + def test_investigation_adheres_to_explain_threshold( + self, grok_profile: Dict[str, Any] + ) -> None: + """Explain class requires 0.75 threshold.""" + query = "How does the caching layer invalidate stale keys" + baseline = resolve_skill_order(grok_profile, query, ROOT) + + # 0.73 is below explain (0.75) -> demote + obs_below = ProviderObservation( + outcome=DecisionOutcome(status="decided", kind="choice", value="investigation"), + provider_identity="typesafe-jev", + model_identity="jev-pilot-v1", + confidence=0.73, + ) + provider_below = FakeDecisionProvider(canned_observation=obs_below) + res_below = resolve_skill_order_with_shadow( + grok_profile, query, ROOT, provider=provider_below, mode="advisory" + ) + assert res_below.advisory_promoted is False + + # 0.78 is above explain (0.75) -> promote + obs_above = ProviderObservation( + outcome=DecisionOutcome(status="decided", kind="choice", value="investigation"), + provider_identity="typesafe-jev", + model_identity="jev-pilot-v1", + confidence=0.78, + ) + provider_above = FakeDecisionProvider(canned_observation=obs_above) + res_above = resolve_skill_order_with_shadow( + grok_profile, query, ROOT, provider=provider_above, mode="advisory" + ) + assert res_above.advisory_promoted is True + assert res_above.target == "investigation" + + def test_decision_margin_gate_rejects_ambiguous_distribution( + self, grok_profile: Dict[str, Any] + ) -> None: + """When probabilities are present, margin between top 2 choices must be >= 0.10.""" + query = "Auth endpoint throws 500 on valid credentials" + baseline = resolve_skill_order(grok_profile, query, ROOT) + + # High confidence but razor-thin margin: 0.86 vs 0.83 (margin = 0.03 < 0.10) + obs_thin = ProviderObservation( + outcome=DecisionOutcome(status="decided", kind="choice", value="bug-fix"), + provider_identity="typesafe-jev", + model_identity="jev-pilot-v1", + confidence=0.86, + probabilities={"bug-fix": 0.86, "perf-issue": 0.83}, + ) + provider_thin = FakeDecisionProvider(canned_observation=obs_thin) + + res_thin = resolve_skill_order_with_shadow( + grok_profile, query, ROOT, provider=provider_thin, mode="advisory" + ) + assert res_thin.advisory_promoted is False + + # Healthy margin: 0.88 vs 0.10 (margin = 0.78 >= 0.10) -> promote + obs_healthy = ProviderObservation( + outcome=DecisionOutcome(status="decided", kind="choice", value="bug-fix"), + provider_identity="typesafe-jev", + model_identity="jev-pilot-v1", + confidence=0.88, + probabilities={"bug-fix": 0.88, "perf-issue": 0.10}, + ) + provider_healthy = FakeDecisionProvider(canned_observation=obs_healthy) + + res_healthy = resolve_skill_order_with_shadow( + grok_profile, query, ROOT, provider=provider_healthy, mode="advisory" + ) + assert res_healthy.advisory_promoted is True + assert res_healthy.target == "bug-fix" diff --git a/tests/test_jev_transport_and_profiles.py b/tests/test_jev_transport_and_profiles.py new file mode 100644 index 00000000..1e98faa7 --- /dev/null +++ b/tests/test_jev_transport_and_profiles.py @@ -0,0 +1,275 @@ +#!/usr/bin/env python3 +"""Tests for Jev provider profile configurations, System One transport, and safety guards. + +Validates Requirements R-JEV-01 through R-JEV-10, addressing: +- #515: single native Python System One transport behind SemanticDecisionProvider +- #519: OpenCode Console/Zen profile configurations and endpoint contracts +- #520: OpenRouter paid profile configuration and pinned models +- #521: Provider parity, free retirement fallback, and explicit paid consent guards +- #517: Model and role fit boundaries preserving baseline authority +""" + +from __future__ import annotations + +import json +import os +import sys +from pathlib import Path +from typing import Any, Dict + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from scripts.decision_adapter import ( + ChoiceOption, + ChoiceRequest, + DecisionOutcome, + EgressClass, + JevDecisionAdapter, + JevProviderProfile, + KNOWN_PROFILES, + ProviderObservation, + ScoreRange, + ScoreRequest, + validate_and_create_safe_context, +) + + +class TestJevProviderProfiles: + """Tests for configured provider profiles and endpoint contracts.""" + + def test_known_profiles_definitions(self) -> None: + assert "opencode-console" in KNOWN_PROFILES + assert "opencode-zen" in KNOWN_PROFILES + assert "openrouter" in KNOWN_PROFILES + assert "typesafe-direct" in KNOWN_PROFILES + + zen = KNOWN_PROFILES["opencode-zen"] + assert zen.api_root == "https://opencode.ai/zen" + assert zen.configured_model == "jev-1.13-free" + assert zen.paid is False + assert zen.credential_env == "OPENCODE_API_KEY" + + console = KNOWN_PROFILES["opencode-console"] + assert console.api_root == "https://opencode.ai/inference/systemone" + assert console.configured_model == "jev-1.13-free" + assert console.paid is False + assert console.credential_env == "OPENCODE_API_KEY" + + openrouter = KNOWN_PROFILES["openrouter"] + assert openrouter.api_root == "https://openrouter.ai/api" + assert openrouter.configured_model == "typesafe/jev-1.14" + assert openrouter.paid is True + assert openrouter.credential_env == "OPENROUTER_API_KEY" + + direct = KNOWN_PROFILES["typesafe-direct"] + assert direct.api_root == "https://api.typesafe.ai" + assert direct.paid is True + assert direct.credential_env == "TYPESAFE_API_KEY" + + def test_adapter_profile_resolution(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Default fallback to opencode-zen + monkeypatch.delenv("PSTACK_JEV_PROFILE", raising=False) + adapter = JevDecisionAdapter(enabled=True) + assert adapter.profile.profile_id == "opencode-zen" + + # Env override + monkeypatch.setenv("PSTACK_JEV_PROFILE", "opencode-console") + adapter_console = JevDecisionAdapter(enabled=True) + assert adapter_console.profile.profile_id == "opencode-console" + + # Explicit profile string + adapter_or = JevDecisionAdapter(enabled=True, profile="openrouter") + assert adapter_or.profile.profile_id == "openrouter" + assert adapter_or.profile.paid is True + + # Custom JevProviderProfile instance + custom_prof = JevProviderProfile( + profile_id="custom-profile", + api_root="https://custom.example.com", + configured_model="jev-custom-v1", + paid=False, + ) + adapter_custom = JevDecisionAdapter(enabled=True, profile=custom_prof) + assert adapter_custom.profile.profile_id == "custom-profile" + + +class TestJevTransportAndGuards: + """Tests for wire execution, paid guards, failure modes, and response parsing.""" + + @pytest.fixture + def sample_request(self) -> ChoiceRequest: + ctx = validate_and_create_safe_context("Fix null pointer in auth handler", EgressClass.SAFE_TO_SEND) + return ChoiceRequest( + question_id="route_selection", + context=ctx, + choices=( + ChoiceOption("bug-fix", "Bug fix"), + ChoiceOption("investigation", "Investigation"), + ChoiceOption("perf-issue", "Performance issue"), + ), + ) + + def test_paid_consent_guard_blocks_unauthorized_egress( + self, sample_request: ChoiceRequest, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delenv("PSTACK_JEV_ALLOW_PAID", raising=False) + monkeypatch.setenv("OPENROUTER_API_KEY", "sk-mock-key") + + # OpenRouter is a paid profile + adapter = JevDecisionAdapter(enabled=True, profile="openrouter") + + obs = adapter.decide(sample_request) + # MUST fail open to unavailable without network egress + assert obs.outcome.status == "unavailable" + assert not obs.is_decided() + + def test_paid_consent_guard_allows_when_explicitly_granted( + self, sample_request: ChoiceRequest, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("PSTACK_JEV_ALLOW_PAID", "1") + monkeypatch.setenv("OPENROUTER_API_KEY", "sk-mock-key") + + captured: Dict[str, Any] = {} + + def mock_transport(url: str, payload: Dict[str, Any], api_key: str) -> Dict[str, Any]: + captured["url"] = url + captured["payload"] = payload + captured["api_key"] = api_key + return { + "model": "typesafe/jev-1.14@20261017", + "provider": "openrouter", + "results": [ + { + "id": "route_selection", + "selected": "bug-fix", + "confidence": 0.94, + "probabilities": {"bug-fix": 0.94, "investigation": 0.04, "perf-issue": 0.02}, + } + ], + } + + adapter = JevDecisionAdapter(enabled=True, profile="openrouter", transport_fn=mock_transport) + obs = adapter.decide(sample_request) + + assert obs.outcome.status == "decided" + assert obs.outcome.value == "bug-fix" + assert obs.confidence == 0.94 + assert obs.provider_identity == "openrouter" + assert obs.model_identity == "typesafe/jev-1.14@20261017" + + # Verify wire payload structure + assert captured["url"] == "https://openrouter.ai/api/v1/systemone" + assert captured["payload"]["model"] == "typesafe/jev-1.14" + assert captured["payload"]["state"] == "Fix null pointer in auth handler" + assert len(captured["payload"]["questions"][0]["options"]) == 3 + + def test_opencode_free_wire_contract( + self, sample_request: ChoiceRequest, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("OPENCODE_API_KEY", "opencode-key") + + captured_url = None + + def mock_transport(url: str, payload: Dict[str, Any], api_key: str) -> Dict[str, Any]: + nonlocal captured_url + captured_url = url + return { + "model": "jev-1.13-free@20261017", + "provider": "opencode", + "results": [ + { + "id": "route_selection", + "selected": "investigation", + "confidence": 0.89, + "probabilities": {"investigation": 0.89, "bug-fix": 0.11}, + } + ], + } + + adapter = JevDecisionAdapter(enabled=True, profile="opencode-zen", transport_fn=mock_transport) + obs = adapter.decide(sample_request) + + assert captured_url == "https://opencode.ai/zen/v1/systemone" + assert obs.is_decided() + assert obs.outcome.value == "investigation" + assert obs.confidence == 0.89 + assert obs.provider_identity == "opencode" + + def test_unrecognized_choice_rejected_as_invalid( + self, sample_request: ChoiceRequest, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("OPENCODE_API_KEY", "opencode-key") + + def mock_transport(url: str, payload: Dict[str, Any], api_key: str) -> Dict[str, Any]: + return { + "model": "jev-1.13-free", + "results": [ + { + "id": "route_selection", + "selected": "unrecognized-playbook-id", + "confidence": 0.99, + } + ], + } + + adapter = JevDecisionAdapter(enabled=True, profile="opencode-zen", transport_fn=mock_transport) + obs = adapter.decide(sample_request) + + assert obs.outcome.status == "invalid" + assert not obs.is_decided() + + def test_malformed_response_fails_open( + self, sample_request: ChoiceRequest, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("OPENCODE_API_KEY", "opencode-key") + + def mock_transport(url: str, payload: Dict[str, Any], api_key: str) -> Any: + # Not a dictionary or missing results + return ["malformed", "list"] + + adapter = JevDecisionAdapter(enabled=True, profile="opencode-zen", transport_fn=mock_transport) + obs = adapter.decide(sample_request) + + assert obs.outcome.status == "invalid" + assert not obs.is_decided() + + def test_transport_exception_fails_open_to_unavailable( + self, sample_request: ChoiceRequest, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("OPENCODE_API_KEY", "opencode-key") + + def mock_transport(url: str, payload: Dict[str, Any], api_key: str) -> Any: + raise ConnectionResetError("Connection dropped by peer") + + adapter = JevDecisionAdapter(enabled=True, profile="opencode-zen", transport_fn=mock_transport) + obs = adapter.decide(sample_request) + + assert obs.outcome.status == "unavailable" + assert not obs.is_decided() + + def test_non_choice_request_in_phase_one_abstains( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("OPENCODE_API_KEY", "opencode-key") + ctx = validate_and_create_safe_context("Numeric score evaluation") + score_req = ScoreRequest( + question_id="score_q", context=ctx, range=ScoreRange(minimum=0.0, maximum=1.0) + ) + + transport_called = False + + def mock_transport(url: str, payload: Dict[str, Any], api_key: str) -> Any: + nonlocal transport_called + transport_called = True + return {} + + adapter = JevDecisionAdapter(enabled=True, profile="opencode-zen", transport_fn=mock_transport) + obs = adapter.decide(score_req) + + assert not transport_called + assert obs.outcome.status == "abstain" + assert not obs.is_decided()