From b775f323f079ef4b48a15126db37bb03187629eb Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 17 Sep 2026 19:45:32 +0000 Subject: [PATCH 1/6] feat(policy-object): frozen validated Policy with rule filters and word floor Every scoring tunable moves into a frozen, validated Policy dataclass whose defaults are the v0.2 constants: ready_at 85, usable_at 60, strict_clarify_at 65, penalties 5/15/25, min_words 3, max_chars 10000, borderline_at 74. Validation rejects mis-ordered bands, out-of-range values, uncompilable allow_patterns, and disabled rule ids that are not registered. InputGuard and analyze() take an optional policy (a per-call policy overrides the guard's). disabled_rules skips registered rules by id; allow_patterns regex matches skip flagging entirely; min_words gates the insufficient_context vague rule below its floor. Penalties flow from Policy through calculate_score(findings, policy=None). Closes review concern C5 (art_hC18m78C): one severity vocabulary (policy.SEVERITIES) shared by registration and a new single choke point, scorer.require_known_severity, which the analyzer applies to every emitted finding; registry.KNOWN_SEVERITIES is cross-pinned to it by test. Zero new dependencies; stdlib only. Co-authored-by: Kalisetti Nihanth Naidu --- inputguard/__init__.py | 4 + inputguard/analyzer.py | 98 +++++++++++----- inputguard/policy.py | 180 +++++++++++++++++++++++++++++ inputguard/scorer.py | 55 +++++++-- tests/test_policy.py | 257 +++++++++++++++++++++++++++++++++++++++++ 5 files changed, 556 insertions(+), 38 deletions(-) create mode 100644 inputguard/policy.py create mode 100644 tests/test_policy.py diff --git a/inputguard/__init__.py b/inputguard/__init__.py index f62271f..0ad497f 100644 --- a/inputguard/__init__.py +++ b/inputguard/__init__.py @@ -1,4 +1,5 @@ from inputguard.analyzer import InputGuard +from inputguard.policy import Policy from inputguard.registry import REGISTRY, Rule, register_domain, register_rule from inputguard.types import AnalysisResult, RuleFinding @@ -16,4 +17,7 @@ "Rule", "register_rule", "register_domain", + # v0.3 policy calibration (additive): the frozen tuning data for scoring, + # status bands, rule filters, and the input cap. + "Policy", ] diff --git a/inputguard/analyzer.py b/inputguard/analyzer.py index d518443..3dc1311 100644 --- a/inputguard/analyzer.py +++ b/inputguard/analyzer.py @@ -1,6 +1,7 @@ from __future__ import annotations -from typing import List, Set +import re +from typing import List, Optional, Set import inputguard.rules # noqa: F401 — importing registers the coding domain and the 19 built-in rules from inputguard.detector import detect_intent, normalize @@ -14,13 +15,17 @@ partial_coverage_note, probe_script, ) +from inputguard.policy import Policy from inputguard.recommender import get_recommendations from inputguard.registry import REGISTRY -from inputguard.scorer import calculate_score, get_status +from inputguard.scorer import calculate_score, get_status, require_known_severity from inputguard.types import AnalysisResult, RuleFinding _VALID_MODES = {"warning", "strict"} +# The one built-in rule that flags input as vague. Policy.min_words gates it: +# below the floor, input is valid-but-short — never "vague" (spec art_bTvdPdJS §2). +_VAGUE_RULE_ID = "insufficient_context" _INTERPRETATION_NOTE = ( "This input is ambiguous in multiple ways. Addressing each gap below " "before sending will prevent the AI from making assumptions that lead " @@ -29,20 +34,45 @@ class InputGuard: - def __init__(self, mode: str = "warning") -> None: + """The clarity engine: ``analyze(user_input, domain=...) -> AnalysisResult``. + + ``mode`` selects the status banding (warning never blocks, strict does). + ``policy`` is optional :class:`~inputguard.Policy` calibration — status + bands, severity penalties, rule filters, and the input cap. It defaults to + the v0.2 constants; every guard and every ``analyze()`` call may carry its + own (a per-call ``policy`` overrides the guard's). + """ + + def __init__(self, mode: str = "warning", policy: Optional[Policy] = None) -> None: if mode not in _VALID_MODES: raise ValueError( f"Invalid mode: {mode!r}. Expected one of: 'warning', 'strict'." ) + if policy is not None and not isinstance(policy, Policy): + raise TypeError( + f"policy must be a Policy instance, got {type(policy).__name__}." + ) self.mode = mode - - def analyze(self, user_input: str, domain: str = "coding") -> AnalysisResult: + self.policy: Policy = Policy() if policy is None else policy + + def analyze( + self, + user_input: str, + domain: str = "coding", + policy: Optional[Policy] = None, + ) -> AnalysisResult: if not isinstance(user_input, str): raise TypeError( f"user_input must be a string, got {type(user_input).__name__}." ) if not user_input.strip(): raise ValueError("user_input must be a non-empty, non-whitespace string.") + if policy is not None and not isinstance(policy, Policy): + raise TypeError( + f"policy must be a Policy instance, got {type(policy).__name__}." + ) + + effective_policy = self.policy if policy is None else policy domain_signals = REGISTRY.get_domain_signals(domain) normalized = normalize(user_input) @@ -74,27 +104,38 @@ def analyze(self, user_input: str, domain: str = "coding") -> AnalysisResult: findings: List[RuleFinding] = [] seen_codes: Set[str] = set() - for rule in REGISTRY.rules_for_intent(detected_intent): - try: - finding = rule.check(normalized) - except Exception as exc: - # Loud failure with attribution (review C2-lite): a rule - # exception is never swallowed or silently degraded around — - # it aborts analyze(), naming the rule and where it was - # registered, with the original traceback chained. - raise RuntimeError( - f"inputguard rule {rule.id!r} " - f"(registered {REGISTRY.rule_origin(rule.id)}) raised " - f"{type(exc).__name__}: {exc}. A rule exception aborts " - f"analyze() by design — rules are the author's " - f"responsibility after registration; fix or remove the rule." - ) from exc - if finding is None or finding.code in seen_codes: - continue - seen_codes.add(finding.code) - findings.append(finding) - - score = calculate_score(findings) + if not _matches_allow_pattern(user_input, effective_policy): + word_count = len(normalized.split()) + for rule in REGISTRY.rules_for_intent(detected_intent): + if rule.id in effective_policy.disabled_rules: + continue + if rule.id == _VAGUE_RULE_ID and word_count < effective_policy.min_words: + continue + try: + finding = rule.check(normalized) + except Exception as exc: + # Loud failure with attribution (review C2-lite): a rule + # exception is never swallowed or silently degraded around — + # it aborts analyze(), naming the rule and where it was + # registered, with the original traceback chained. + raise RuntimeError( + f"inputguard rule {rule.id!r} " + f"(registered {REGISTRY.rule_origin(rule.id)}) raised " + f"{type(exc).__name__}: {exc}. A rule exception aborts " + f"analyze() by design — rules are the author's " + f"responsibility after registration; fix or remove the rule." + ) from exc + if finding is None: + continue + # C5 choke point: every emitted finding's severity must be in + # the same Policy-owned vocabulary registration validated. + require_known_severity(finding.severity) + if finding.code in seen_codes: + continue + seen_codes.add(finding.code) + findings.append(finding) + + score = calculate_score(findings, effective_policy) status = get_status(score, self.mode) gaps: List[str] = [] @@ -131,3 +172,8 @@ def analyze(self, user_input: str, domain: str = "coding") -> AnalysisResult: else None ), ) + + +def _matches_allow_pattern(text: str, policy: Policy) -> bool: + """True when the analyzed input matches any allowlist pattern — such input is never flagged.""" + return any(re.search(pattern, text) for pattern in policy.allow_patterns) diff --git a/inputguard/policy.py b/inputguard/policy.py new file mode 100644 index 0000000..ea302de --- /dev/null +++ b/inputguard/policy.py @@ -0,0 +1,180 @@ +"""Policy: InputGuard calibration as data, not constants. + +A :class:`Policy` bundles every tunable the ``analyze()`` pipeline consumes — +status bands, severity penalties, the input cap, rule filters, and the +borderline near-miss band — in one frozen, validated object. Every default is +the v0.2 constant, pinned byte-exactly by ``tests/test_policy_defaults.py`` so +existing behavior cannot drift. + +Two layers stay separate: per-rule severity decides *what fires*; the policy's +bands decide *what happens* to the score. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import FrozenSet, Optional, Tuple + +__all__ = ["Policy"] + + +SEVERITIES: Tuple[str, ...] = ("low", "medium", "high") +"""The one severity vocabulary, owned here and shared by both validation sites +(review concern C5, art_hC18m78C): registration validates a rule's *declared* +severity against it (``registry.KNOWN_SEVERITIES``, cross-pinned by +``tests/test_policy.py``), and scoring validates every *emitted* finding's +severity against it (``scorer.require_known_severity``) — so the two checks +cannot drift apart. Deliberately not a ``Policy`` field: a per-instance +vocabulary would be silently ignored by the penalty lookup. Extend only +additively, with a matching ``penalty_*`` field.""" + + +@dataclass(frozen=True) +class Policy: + """Calibration for one :class:`~inputguard.InputGuard`. + + Every default is the v0.2 constant (pinned by ``tests/test_policy_defaults.py``). + Instances are frozen, validated at construction, and safe to share across + threads and guards. + + Fields + ------ + ready_at: + A score at or above this is ``ready`` in both modes. + usable_at: + Warning-mode floor for ``usable_with_warnings``. + strict_clarify_at: + Strict-mode floor for ``needs_clarification``; below it, strict blocks. + penalty_low / penalty_medium / penalty_high: + Points deducted per distinct gap, by the highest-severity finding in + that gap. Must satisfy ``penalty_low <= penalty_medium <= penalty_high``. + min_words: + Inputs with fewer words are never flagged as vague by the built-in + ``insufficient_context`` rule — short input is valid-but-short, not + vague. That rule keeps a structural minimum of 3 words, so values + below 3 cannot force shorter input to be flagged. + max_chars: + Input cap. ``analyze()`` inspects at most this many characters; + truncation is reported on the result (``AnalysisResult.truncated``), + never silent. + borderline_at: + Near-miss band: a score in ``[borderline_at, ready_at)`` sets the + result's ``borderline`` signal — distinct "worth one more pass" + messaging just below the ready floor. Set ``borderline_at`` equal to + ``ready_at`` to disable the signal. + disabled_rules: + Rule ids to skip entirely during ``analyze()``. Every id must be a + registered rule id (validated at construction). + allow_patterns: + Regex patterns; analyzed input matching any of them is never flagged — + rules are skipped and the result is ``ready``. Patterns are matched + against the analyzed (length-capped) input via ``re.search``. + """ + + ready_at: int = 85 + usable_at: int = 60 + strict_clarify_at: int = 65 + penalty_low: int = 5 + penalty_medium: int = 15 + penalty_high: int = 25 + min_words: int = 3 + max_chars: int = 10_000 + borderline_at: int = 74 + disabled_rules: FrozenSet[str] = frozenset() + allow_patterns: Tuple[str, ...] = () + + def __post_init__(self) -> None: + # A bare string would silently explode into characters under + # frozenset()/tuple() — reject it before coercion. + if isinstance(self.disabled_rules, str): + raise ValueError( + "Policy.disabled_rules must be a collection of rule ids, e.g. " + f"frozenset({{'missing_language'}}), got the string {self.disabled_rules!r}." + ) + if isinstance(self.allow_patterns, str): + raise ValueError( + "Policy.allow_patterns must be a collection of regex strings, " + f"e.g. ('^re:',), got the string {self.allow_patterns!r}." + ) + # Coerce to immutable containers so a later mutation of the caller's + # set or list cannot change a "frozen" policy mid-flight. + object.__setattr__(self, "disabled_rules", frozenset(self.disabled_rules)) + object.__setattr__(self, "allow_patterns", tuple(self.allow_patterns)) + _validate(self) + + +def _check_int(policy: Policy, name: str, *, minimum: int, maximum: Optional[int]) -> None: + value = getattr(policy, name) + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError(f"Policy.{name} must be an int, got {value!r}.") + if maximum is None: + if value < minimum: + raise ValueError(f"Policy.{name} must be >= {minimum}, got {value}.") + elif not minimum <= value <= maximum: + raise ValueError(f"Policy.{name} must be in [{minimum}, {maximum}], got {value}.") + + +def _validate(policy: Policy) -> None: + for name in ("ready_at", "usable_at", "strict_clarify_at", "borderline_at"): + _check_int(policy, name, minimum=0, maximum=100) + if not policy.usable_at <= policy.strict_clarify_at < policy.ready_at: + raise ValueError( + "Policy bands are mis-ordered: they must satisfy " + "usable_at <= strict_clarify_at < ready_at, got " + f"usable_at={policy.usable_at}, strict_clarify_at={policy.strict_clarify_at}, " + f"ready_at={policy.ready_at}." + ) + if not policy.usable_at <= policy.borderline_at <= policy.ready_at: + raise ValueError( + "Policy.borderline_at must satisfy usable_at <= borderline_at <= ready_at, got " + f"usable_at={policy.usable_at}, borderline_at={policy.borderline_at}, " + f"ready_at={policy.ready_at}. Set borderline_at equal to ready_at to " + "disable the borderline signal." + ) + + for name in ("penalty_low", "penalty_medium", "penalty_high"): + _check_int(policy, name, minimum=0, maximum=None) + if not policy.penalty_low <= policy.penalty_medium <= policy.penalty_high: + raise ValueError( + "Policy penalties are mis-ordered: they must satisfy " + "penalty_low <= penalty_medium <= penalty_high, got " + f"penalty_low={policy.penalty_low}, penalty_medium={policy.penalty_medium}, " + f"penalty_high={policy.penalty_high}." + ) + + _check_int(policy, "min_words", minimum=0, maximum=None) + _check_int(policy, "max_chars", minimum=1, maximum=None) + + for pattern in policy.allow_patterns: + if not isinstance(pattern, str): + raise ValueError( + f"Policy.allow_patterns entries must be regex strings, got {pattern!r}." + ) + try: + re.compile(pattern) + except re.error as exc: + raise ValueError( + f"Policy.allow_patterns entry {pattern!r} is not a valid regex: {exc}" + ) from exc + + for rule_id in policy.disabled_rules: + if not isinstance(rule_id, str): + raise ValueError( + f"Policy.disabled_rules entries must be rule id strings, got {rule_id!r}." + ) + + if policy.disabled_rules: + # Local import: the registry is populated as a side effect of importing + # inputguard.rules, so the known-id check must run at construction time, + # not module-import time (and registry imports nothing from this module). + from inputguard.registry import REGISTRY + + registered = REGISTRY.rule_ids() + unknown = sorted(set(policy.disabled_rules) - set(registered)) + if unknown: + raise ValueError( + "Policy.disabled_rules references unknown rule ids: " + f"{', '.join(repr(r) for r in unknown)}. Registered rule ids: " + f"{', '.join(repr(r) for r in registered)}." + ) diff --git a/inputguard/scorer.py b/inputguard/scorer.py index d6894ad..1cdfc8d 100644 --- a/inputguard/scorer.py +++ b/inputguard/scorer.py @@ -1,13 +1,17 @@ from __future__ import annotations -from typing import Dict, List +from typing import Dict, List, Optional +from inputguard.policy import Policy, SEVERITIES from inputguard.types import RuleFinding SEVERITY_PENALTIES = {"low": 5, "medium": 15, "high": 25} +"""The v0.2 default penalty table — now mirrored by :class:`~inputguard.Policy`'s +penalty defaults (cross-pinned by ``tests/test_policy_defaults.py``). Kept for +backward compatibility of the module surface.""" -_SEVERITY_RANK = {"low": 0, "medium": 1, "high": 2} +_SEVERITY_RANK = {name: rank for rank, name in enumerate(SEVERITIES)} # Warning mode thresholds _WARN_READY = 85 @@ -18,23 +22,37 @@ _STRICT_NEEDS = 65 +def require_known_severity(severity: str) -> None: + """The single choke point validating a severity against ``Policy.SEVERITIES``. + + Two validations share this one vocabulary (review concern C5, + art_hC18m78C): registration validates a rule's *declared* severity + (``registry.KNOWN_SEVERITIES``, cross-pinned by ``tests/test_policy.py``), + and the analyzer/scorer validate every *emitted* finding's severity here — + so the two checks cannot drift apart. + """ + if severity not in _SEVERITY_RANK: + raise ValueError( + f"Unknown severity: {severity!r}. Expected one of: " + f"{', '.join(repr(s) for s in SEVERITIES)}." + ) + + def _severity_rank(severity: str) -> int: """Rank lookup that fails loudly: an unknown severity must never silently score zero.""" - try: - return _SEVERITY_RANK[severity] - except KeyError: - raise ValueError( - f"Unknown severity: {severity!r}. Expected one of: 'low', 'medium', 'high'." - ) from None + require_known_severity(severity) + return _SEVERITY_RANK[severity] -def calculate_score(findings: List[RuleFinding]) -> int: +def calculate_score(findings: List[RuleFinding], policy: Optional[Policy] = None) -> int: """100 minus one penalty per distinct gap (or code, when gap is None), clamped to [0, 100]. - Unknown severities raise ``ValueError`` — a typo'd severity would - otherwise distort every score silently (the v0.2 ``.get(severity, 0)`` + Penalties come from ``policy`` (defaults reproduce the v0.2 constants + byte-exactly). Unknown severities raise ``ValueError`` — a typo'd severity + would otherwise distort every score silently (the v0.2 ``.get(severity, 0)`` behavior). """ + p = Policy() if policy is None else policy # Validate every severity before scoring so a bad finding fails loudly # even when a later finding would otherwise mask it in the dedup loop. for f in findings: @@ -49,11 +67,24 @@ def calculate_score(findings: List[RuleFinding]) -> int: score = 100 for severity in highest_by_gap.values(): - score -= SEVERITY_PENALTIES[severity] + score -= _penalty_for(severity, p) return max(0, min(100, score)) +def _penalty_for(severity: str, policy: Policy) -> int: + """Per-distinct-gap penalty for a severity already validated by ``_severity_rank``.""" + if severity == "low": + return policy.penalty_low + if severity == "medium": + return policy.penalty_medium + if severity == "high": + return policy.penalty_high + raise ValueError( + f"Unknown severity: {severity!r}. Expected one of: 'low', 'medium', 'high'." + ) + + def get_status(score: int, mode: str) -> str: if mode == "warning": if score >= _WARN_READY: diff --git a/tests/test_policy.py b/tests/test_policy.py new file mode 100644 index 0000000..0fc1622 --- /dev/null +++ b/tests/test_policy.py @@ -0,0 +1,257 @@ +"""Policy object tests: validation, filtering, severity vocabulary, and plumbing. + +Covers the policy-object slice of the v0.3 spec (art_bTvdPdJS §2) plus the C5 +closure from the adversarial API review (art_hC18m78C): one Policy-owned +severity vocabulary shared by registration and scoring. Byte-exact default +pinning lives in tests/test_policy_defaults.py. +""" + +from __future__ import annotations + +import dataclasses +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from inputguard import InputGuard, Policy, RuleFinding +from inputguard.policy import SEVERITIES +from inputguard.registry import KNOWN_SEVERITIES, register_rule +from inputguard.scorer import require_known_severity +from test_registry import _test_rule, registry_isolation # noqa: F401 — pytest fixture + +# Observed v0.2 / release-branch behavior (runtime probes), reused as fixtures. +VAGUE_INPUT = "do something now" # build intent, one high finding: insufficient_context +SINGLE_HIGH_INPUT = "build a REST API using Python" # build intent, one high finding + + +# --- construction, coercion, and validation -------------------------------- + + +def test_policy_is_frozen(): + policy = Policy() + with pytest.raises(dataclasses.FrozenInstanceError): + policy.ready_at = 90 # type: ignore[misc] + + +def test_disabled_rules_coerced_to_frozenset_and_isolated(): + source = {"missing_language"} + policy = Policy(disabled_rules=source) # type: ignore[arg-type] + assert isinstance(policy.disabled_rules, frozenset) + source.add("missing_api_structure") # mutating the source must not leak in + assert policy.disabled_rules == frozenset({"missing_language"}) + + +def test_allow_patterns_coerced_to_tuple_and_isolated(): + source = ["^re:"] + policy = Policy(allow_patterns=source) # type: ignore[arg-type] + assert isinstance(policy.allow_patterns, tuple) + source.append("x") + assert policy.allow_patterns == ("^re:",) + + +def test_string_disabled_rules_rejected(): + # A bare string would silently become a set of characters. + with pytest.raises(ValueError, match="collection of rule ids"): + Policy(disabled_rules="missing_language") # type: ignore[arg-type] + + +def test_string_allow_patterns_rejected(): + with pytest.raises(ValueError, match="collection of regex"): + Policy(allow_patterns="^re:") # type: ignore[arg-type] + + +@pytest.mark.parametrize( + "kwargs, match", + [ + ({"usable_at": 70, "strict_clarify_at": 65}, "mis-ordered"), + ({"strict_clarify_at": 90}, "mis-ordered"), + ({"usable_at": 90}, "mis-ordered"), + ({"ready_at": 101}, "ready_at"), + ({"usable_at": -1}, "usable_at"), + ({"borderline_at": 90, "ready_at": 85}, "borderline_at"), + ({"borderline_at": 50}, "borderline_at"), + ({"penalty_high": -1}, "penalty_high"), + ({"penalty_low": 30}, "penalties are mis-ordered"), + ({"penalty_medium": 30, "penalty_high": 25}, "penalties are mis-ordered"), + ({"min_words": -1}, "min_words"), + ({"max_chars": 0}, "max_chars"), + ({"ready_at": "85"}, "must be an int"), + ({"disabled_rules": {"no_such_rule_id"}}, "unknown rule ids"), + ({"allow_patterns": ("(",)}, "not a valid regex"), + ({"allow_patterns": (5,)}, "must be regex strings"), + ({"disabled_rules": {5}}, "must be rule id strings"), + ], +) +def test_invalid_policy_rejected_with_value_error(kwargs, match): + with pytest.raises(ValueError, match=match): + Policy(**kwargs) + + +def test_band_edges_are_valid(): + policy = Policy(usable_at=0, strict_clarify_at=0, ready_at=100, borderline_at=0) + assert policy.ready_at == 100 + + +def test_borderline_disabled_by_setting_equal_to_ready_at(): + policy = Policy(borderline_at=85) + assert policy.borderline_at == policy.ready_at + + +def test_known_rule_ids_accepted_in_disabled_rules(): + policy = Policy( + disabled_rules=frozenset({"missing_error_message", "missing_language"}) + ) + assert policy.disabled_rules == frozenset( + {"missing_error_message", "missing_language"} + ) + + +# --- C5: one severity vocabulary, shared by registration and scoring -------- + + +def test_registry_severity_vocabulary_cross_pins_to_policy(): + # The registration check (registry.KNOWN_SEVERITIES) and the scoring choke + # point (policy.SEVERITIES) must never drift apart. + assert KNOWN_SEVERITIES == SEVERITIES + assert SEVERITIES == ("low", "medium", "high") + + +def test_require_known_severity_rejects_unknown(): + with pytest.raises(ValueError, match="Unknown severity: 'typo'"): + require_known_severity("typo") + + +def test_rule_emitting_unknown_severity_fails_loudly_at_analyze(registry_isolation): + # Declared severity is valid; the EMITTED finding's severity is not. The + # analyzer choke point must reject it — the tie registration could not give us. + bad = _test_rule( + rule_id="bad_severity_rule", + domain="debug", + severity="high", + gap="error description", + finding=RuleFinding(code="bad_finding", message="m", severity="critical"), + ) + register_rule(bad) + with pytest.raises(ValueError, match="Unknown severity: 'critical'"): + InputGuard().analyze("fix the bug in my app") + + +# --- disabled_rules filtering ------------------------------------------------ + + +def test_disabled_rule_is_skipped_during_analyze(): + guard = InputGuard( + policy=Policy(disabled_rules=frozenset({"missing_api_structure"})) + ) + result = guard.analyze(SINGLE_HIGH_INPUT) + assert result.findings == [] # the only finding this input produces + assert result.clarity_score == 100 + assert result.status == "ready" + + +def test_disabled_rules_default_is_no_filtering(): + result = InputGuard().analyze(SINGLE_HIGH_INPUT) + assert [f.code for f in result.findings] == ["missing_api_structure"] + + +def test_disabled_rules_apply_per_call_too(): + guard = InputGuard() + result = guard.analyze( + SINGLE_HIGH_INPUT, + policy=Policy(disabled_rules=frozenset({"missing_api_structure"})), + ) + assert result.findings == [] + + +# --- allow_patterns filtering ------------------------------------------------- + + +def test_allow_pattern_skips_flagging(): + guard = InputGuard(policy=Policy(allow_patterns=(r"\bfixture\b",))) + result = guard.analyze("I love this fixture in the test suite, what does it do") + assert result.status == "ready" + assert result.clarity_score == 100 + assert result.findings == [] + assert result.gaps == [] + + +def test_allow_pattern_no_match_leaves_analysis_unchanged(): + matched = InputGuard(policy=Policy(allow_patterns=("fix",))).analyze( + "fix the bug in my app" + ) + unmatched = InputGuard(policy=Policy(allow_patterns=("deploy",))).analyze( + "fix the bug in my app" + ) + assert matched.status == "ready" + assert matched.findings == [] + assert unmatched.findings # still analyzed + assert unmatched.status != "ready" + + +# --- min_words: short input is valid-but-short, never "vague" ----------------- + + +def test_min_words_below_floor_never_vague(): + result = InputGuard().analyze(VAGUE_INPUT) + assert [f.code for f in result.findings] == ["insufficient_context"] + assert result.clarity_score == 75 + + raised = InputGuard(policy=Policy(min_words=10)).analyze(VAGUE_INPUT) + assert raised.findings == [] + assert raised.clarity_score == 100 + assert raised.status == "ready" + + +def test_min_words_default_matches_explicit_default_policy(): + default_guard = InputGuard() + pinned = InputGuard(policy=Policy()) + assert default_guard.analyze(VAGUE_INPUT).to_dict() == pinned.analyze( + VAGUE_INPUT + ).to_dict() + + +def test_min_words_above_floor_still_flags_long_vague_input(): + result = InputGuard(policy=Policy(min_words=4)).analyze("do something now please") + assert [f.code for f in result.findings] == ["insufficient_context"] + + +# --- plumbing: guard attribute, per-call override, type errors ---------------- + + +def test_guard_exposes_its_policy(): + policy = Policy(penalty_high=30) + assert InputGuard(policy=policy).policy is policy + assert InputGuard().policy == Policy() + + +def test_per_call_policy_overrides_guard_policy(): + guard = InputGuard(policy=Policy(min_words=10)) + assert guard.analyze(VAGUE_INPUT).findings == [] # guard policy applies + allowed = guard.analyze(VAGUE_INPUT, policy=Policy()) # per-call override + assert [f.code for f in allowed.findings] == ["insufficient_context"] + + +def test_constructor_rejects_non_policy(): + with pytest.raises(TypeError, match="Policy instance"): + InputGuard(policy=Policy) # class, not instance + with pytest.raises(TypeError, match="Policy instance"): + InputGuard(policy={"ready_at": 85}) # type: ignore[dict-item] + + +def test_analyze_rejects_non_policy(): + guard = InputGuard() + with pytest.raises(TypeError, match="Policy instance"): + guard.analyze(SINGLE_HIGH_INPUT, policy=85) # type: ignore[arg-type] + + +# --- thread safety with a shared custom policy -------------------------------- + + +def test_shared_policy_parallel_analyze_consistent(): + policy = Policy(disabled_rules=frozenset({"missing_debug_code_context"})) + guard = InputGuard(policy=policy) + text = "fix the bug in my app" + expected = guard.analyze(text).to_dict() + with ThreadPoolExecutor(max_workers=8) as pool: + results = list(pool.map(lambda _: guard.analyze(text).to_dict(), range(64))) + assert results == [expected] * 64 From 49a2dee3778e3a95424114922ad5eec716540460 Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 17 Sep 2026 19:47:42 +0000 Subject: [PATCH 2/6] feat(two-layer-banding): policy-driven status bands and a borderline near-miss signal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_status(score, mode, policy=None) now reads its bands from Policy (ready_at 85, usable_at 60, strict_clarify_at 65 — the v0.2 constants, pinned byte-exactly in tests/test_banding.py), replacing the four hard-coded scorer thresholds. Severity still decides what fires; bands decide what happens. AnalysisResult gains an additive borderline: bool field (default False, in to_dict) set when the score lands in [policy.borderline_at, ready_at) — the distinct near-miss "worth one more pass" signal from the spec's result states. interpretation_note is untouched, so existing serialized results are byte-identical under default policy. Co-authored-by: Kalisetti Nihanth Naidu --- inputguard/analyzer.py | 9 ++- inputguard/scorer.py | 28 +++++---- inputguard/types.py | 5 ++ tests/test_banding.py | 139 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 166 insertions(+), 15 deletions(-) create mode 100644 tests/test_banding.py diff --git a/inputguard/analyzer.py b/inputguard/analyzer.py index 3dc1311..9b75a7e 100644 --- a/inputguard/analyzer.py +++ b/inputguard/analyzer.py @@ -88,7 +88,7 @@ def analyze( # modes; the note says honestly what the tool does not know. score = max(0, 100 - DEGRADATION_PENALTY) return AnalysisResult( - status=get_status(score, self.mode), + status=get_status(score, self.mode, effective_policy), clarity_score=score, detected_intent=DEGRADED_INTENT, gaps=[], @@ -136,7 +136,11 @@ def analyze( findings.append(finding) score = calculate_score(findings, effective_policy) - status = get_status(score, self.mode) + status = get_status(score, self.mode, effective_policy) + # Two layers: severity decided what fired; the policy's bands decide + # what happens. The borderline band is the near-miss signal just below + # ready — distinct "worth one more pass" messaging (spec art_bTvdPdJS). + borderline = effective_policy.borderline_at <= score < effective_policy.ready_at gaps: List[str] = [] seen: Set[str] = set() @@ -171,6 +175,7 @@ def analyze( if probe.heuristic_coverage == COVERAGE_PARTIAL else None ), + borderline=borderline, ) diff --git a/inputguard/scorer.py b/inputguard/scorer.py index 1cdfc8d..3b1d31a 100644 --- a/inputguard/scorer.py +++ b/inputguard/scorer.py @@ -13,14 +13,6 @@ _SEVERITY_RANK = {name: rank for rank, name in enumerate(SEVERITIES)} -# Warning mode thresholds -_WARN_READY = 85 -_WARN_USABLE = 60 - -# Strict mode thresholds -_STRICT_READY = 85 -_STRICT_NEEDS = 65 - def require_known_severity(severity: str) -> None: """The single choke point validating a severity against ``Policy.SEVERITIES``. @@ -85,17 +77,27 @@ def _penalty_for(severity: str, policy: Policy) -> int: ) -def get_status(score: int, mode: str) -> str: +def get_status(score: int, mode: str, policy: Optional[Policy] = None) -> str: + """Map a score to a status using the mode's policy bands. + + Bands come from ``policy`` (defaults reproduce the v0.2 constants + byte-exactly): warning mode — ``ready`` at ``ready_at`` (85), then + ``usable_with_warnings`` at ``usable_at`` (60); strict mode — ``ready`` + at ``ready_at`` (85), then ``needs_clarification`` at ``strict_clarify_at`` + (65), else ``blocked``. Two layers stay separate: severity decided what + fired; these bands decide what happens. + """ + p = Policy() if policy is None else policy if mode == "warning": - if score >= _WARN_READY: + if score >= p.ready_at: return "ready" - if score >= _WARN_USABLE: + if score >= p.usable_at: return "usable_with_warnings" return "needs_clarification" if mode == "strict": - if score >= _STRICT_READY: + if score >= p.ready_at: return "ready" - if score >= _STRICT_NEEDS: + if score >= p.strict_clarify_at: return "needs_clarification" return "blocked" raise ValueError(f"Unknown mode: {mode!r}. Expected 'warning' or 'strict'.") diff --git a/inputguard/types.py b/inputguard/types.py index 573046b..15329ea 100644 --- a/inputguard/types.py +++ b/inputguard/types.py @@ -32,6 +32,10 @@ class AnalysisResult: detected_language: str = "en" heuristic_coverage: str = "full" degradation_note: Optional[str] = None + # v0.3 policy calibration (additive): near-miss signal — True when the + # score lands in [policy.borderline_at, policy.ready_at), the "worth one + # more pass" band just below ready. + borderline: bool = False def to_dict(self) -> dict: return { @@ -46,6 +50,7 @@ def to_dict(self) -> dict: "detected_language": self.detected_language, "heuristic_coverage": self.heuristic_coverage, "degradation_note": self.degradation_note, + "borderline": self.borderline, } def is_clear(self) -> bool: diff --git a/tests/test_banding.py b/tests/test_banding.py new file mode 100644 index 0000000..f183cda --- /dev/null +++ b/tests/test_banding.py @@ -0,0 +1,139 @@ +"""Two-layer banding tests: policy-driven status bands and the borderline signal. + +Covers the banding slice of the v0.3 spec (art_bTvdPdJS §2 and Result states): +severity decides what fires, Policy bands decide what happens. Byte-exact +default pinning lives in tests/test_policy_defaults.py. +""" + +from __future__ import annotations + +import dataclasses + +import pytest + +from inputguard import InputGuard, Policy +from inputguard.scorer import get_status + +# Observed via runtime probe: "do something now" scores 75 (one high finding). +SINGLE_HIGH_INPUT = "do something now" + + +# --- default bands reproduce v0.2 warning mode ------------------------------- + + +def test_default_warning_bands_byte_exact(): + assert get_status(100, "warning") == "ready" + assert get_status(85, "warning") == "ready" + assert get_status(84, "warning") == "usable_with_warnings" + assert get_status(60, "warning") == "usable_with_warnings" + assert get_status(59, "warning") == "needs_clarification" + assert get_status(0, "warning") == "needs_clarification" + + +def test_default_strict_bands_byte_exact(): + assert get_status(100, "strict") == "ready" + assert get_status(85, "strict") == "ready" + assert get_status(84, "strict") == "needs_clarification" + assert get_status(65, "strict") == "needs_clarification" + assert get_status(64, "strict") == "blocked" + assert get_status(0, "strict") == "blocked" + + +def test_default_policy_get_status_matches_no_policy(): + for mode in ("warning", "strict"): + for score in (0, 59, 60, 64, 65, 74, 84, 85, 100): + assert get_status(score, mode) == get_status(score, mode, Policy()) + + +# --- bands are tunable data --------------------------------------------------- + + +def test_custom_bands_change_warning_mode(): + policy = Policy(ready_at=90, usable_at=40) + assert get_status(85, "warning", policy) == "usable_with_warnings" + assert get_status(90, "warning", policy) == "ready" + assert get_status(39, "warning", policy) == "needs_clarification" + + +def test_custom_bands_change_strict_mode(): + policy = Policy(ready_at=95, strict_clarify_at=75) + assert get_status(80, "strict", policy) == "needs_clarification" + assert get_status(95, "strict", policy) == "ready" + assert get_status(74, "strict", policy) == "blocked" + + +def test_relaxed_bands_never_block_in_warning_mode(): + # Warning mode never returns blocked for any score, whatever the bands. + policy = Policy(usable_at=0, strict_clarify_at=0) + for score in range(0, 101): + assert get_status(score, "warning", policy) != "blocked" + + +# --- the borderline near-miss signal ------------------------------------------ + + +def test_default_borderline_band_is_74_to_84(): + guard = InputGuard() + result = guard.analyze(SINGLE_HIGH_INPUT) + assert result.clarity_score == 75 + assert result.borderline is True + assert result.status == "usable_with_warnings" + + +def test_scores_outside_the_band_are_not_borderline(): + guard = InputGuard() + below = guard.analyze("fix the bug in my app") # 35 + assert below.clarity_score == 35 + assert below.borderline is False + clean = guard.analyze( + "Build a web app using React and FastAPI. " + "It needs user authentication with JWT. " + "Store tasks in a PostgreSQL database with title, description, and due date fields. " + "Expose REST API endpoints: GET /tasks, POST /tasks, DELETE /tasks/{id}." + ) # 100 + assert clean.clarity_score == 100 + assert clean.borderline is False + + +def test_borderline_at_bounds(): + # Band floor inclusive: borderline_at == score -> borderline. + assert ( + InputGuard(policy=Policy(borderline_at=75)) + .analyze(SINGLE_HIGH_INPUT) + .borderline + is True + ) + # Disabled: borderline_at == ready_at means no score can be borderline. + disabled = InputGuard(policy=Policy(borderline_at=85)) + result = disabled.analyze(SINGLE_HIGH_INPUT) + assert result.borderline is False + + +def test_borderline_works_in_strict_mode_too(): + # Score 75 in strict mode is needs_clarification (65-84) and inside [74, 85). + result = InputGuard(mode="strict").analyze(SINGLE_HIGH_INPUT) + assert result.status == "needs_clarification" + assert result.borderline is True + + +def test_borderline_field_is_additive_on_to_dict(): + result = InputGuard().analyze(SINGLE_HIGH_INPUT).to_dict() + assert result["borderline"] is True + # Every v0.2 key is still present. + assert set(result).issuperset( + { + "status", + "clarity_score", + "detected_intent", + "gaps", + "recommendations", + "findings", + "interpretation_note", + } + ) + + +def test_frozen_result_rejects_mutation(): + result = InputGuard().analyze(SINGLE_HIGH_INPUT) + with pytest.raises(dataclasses.FrozenInstanceError): + result.borderline = False # type: ignore[misc] From 7f211ec241d8db7022cf7576b34d3f4c7e2f5bab Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 17 Sep 2026 19:48:31 +0000 Subject: [PATCH 3/6] feat(input-cap): enforce Policy.max_chars with visible truncation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit analyze() now slices input to policy.max_chars before intent detection and rule execution — analysis cost is bounded regardless of input size (v0.2 took ~3.8 s on a 1.6 MB input; capped analysis of the same input runs in milliseconds). Truncation is never silent: AnalysisResult gains an additive truncated: bool field (default False, in to_dict), and the allowlist is evaluated against the same capped input so no scan escapes the budget. Default cap 10_000 matches the Policy default, pinned by test. Co-authored-by: Kalisetti Nihanth Naidu --- inputguard/analyzer.py | 8 ++++ inputguard/types.py | 2 + tests/test_input_cap.py | 90 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 100 insertions(+) create mode 100644 tests/test_input_cap.py diff --git a/inputguard/analyzer.py b/inputguard/analyzer.py index 9b75a7e..5779960 100644 --- a/inputguard/analyzer.py +++ b/inputguard/analyzer.py @@ -74,6 +74,13 @@ def analyze( effective_policy = self.policy if policy is None else policy + # Enforce the input cap: analysis is bounded, always. Truncation is + # visible on the result (AnalysisResult.truncated), never silent. + max_chars = effective_policy.max_chars + truncated = len(user_input) > max_chars + if truncated: + user_input = user_input[:max_chars] + domain_signals = REGISTRY.get_domain_signals(domain) normalized = normalize(user_input) @@ -176,6 +183,7 @@ def analyze( else None ), borderline=borderline, + truncated=truncated, ) diff --git a/inputguard/types.py b/inputguard/types.py index 15329ea..94f01d5 100644 --- a/inputguard/types.py +++ b/inputguard/types.py @@ -36,6 +36,7 @@ class AnalysisResult: # score lands in [policy.borderline_at, policy.ready_at), the "worth one # more pass" band just below ready. borderline: bool = False + truncated: bool = False def to_dict(self) -> dict: return { @@ -51,6 +52,7 @@ def to_dict(self) -> dict: "heuristic_coverage": self.heuristic_coverage, "degradation_note": self.degradation_note, "borderline": self.borderline, + "truncated": self.truncated, } def is_clear(self) -> bool: diff --git a/tests/test_input_cap.py b/tests/test_input_cap.py new file mode 100644 index 0000000..12b1800 --- /dev/null +++ b/tests/test_input_cap.py @@ -0,0 +1,90 @@ +"""Input-cap tests: enforced max_chars truncation, visible in the result. + +Covers the input-length slice of the v0.3 spec (art_bTvdPdJS §2): analyzed +input is bounded, always, and truncation is reported on the result — never +silent. The v0.2 code had no cap at all (a 1.6 MB input took ~3.8 s). +""" + +from __future__ import annotations + +import time + +from inputguard import InputGuard, Policy + +CLEAN_INPUT = ( + "Build a web app using React and FastAPI. " + "It needs user authentication with JWT. " + "Store tasks in a PostgreSQL database with title, description, and due date fields. " + "Expose REST API endpoints: GET /tasks, POST /tasks, DELETE /tasks/{id}." +) + + +def test_input_at_cap_is_not_truncated(): + guard = InputGuard() + assert guard.analyze("a" * 10_000).truncated is False + + +def test_input_one_char_over_cap_is_truncated(): + guard = InputGuard() + assert guard.analyze("a" * 10_001).truncated is True + + +def test_truncation_visible_in_to_dict(): + assert InputGuard().analyze("x" * 10_001).to_dict()["truncated"] is True + + +def test_default_cap_is_the_v02_compatible_10k(): + assert Policy().max_chars == 10_000 + + +def test_custom_cap_truncates_at_its_boundary(): + guard = InputGuard(policy=Policy(max_chars=50)) + assert guard.analyze("x" * 51).truncated is True + assert guard.analyze("x" * 50).truncated is False + + +def test_truncated_analysis_equals_analyzing_the_kept_prefix(): + # Only the truncation flag may differ — the cap is a pure prefix slice. + prefix = "fix the bug in my app" + huge = prefix + " and some filler that would change nothing " * 400 + guard = InputGuard() + full = guard.analyze(huge).to_dict() + head = guard.analyze(huge[:10_000]).to_dict() + assert full["truncated"] is True + assert head["truncated"] is False + full.pop("truncated") + head.pop("truncated") + assert full == head + + +def test_findings_come_from_the_kept_prefix(): + # Flag-worthy content inside the cap still fires after truncation. + guard = InputGuard(policy=Policy(max_chars=100)) + result = guard.analyze("fix the bug in my app" + " z" * 200) + assert result.truncated is True + assert result.findings[0].code == "missing_error_message" + + +def test_allowlist_evaluated_against_the_capped_input(): + # The allow pattern must match within the analyzed (capped) region. + guard = InputGuard(policy=Policy(max_chars=10, allow_patterns=("^fix",))) + result = guard.analyze("fix the bug in my app" + "z" * 500) + assert result.truncated is True + assert result.findings == [] + assert result.status == "ready" + + +def test_clean_short_input_reports_no_truncation(): + result = InputGuard().analyze(CLEAN_INPUT) + assert result.truncated is False + assert result.status == "ready" + + +def test_huge_input_analysis_is_bounded_by_the_cap(): + # v0.2 took ~3.8 s on a 1.6 MB input; the cap must bound the scan. + huge = ("make my database query faster " * 55_000)[:1_600_000] + start = time.perf_counter() + result = InputGuard().analyze(huge) + elapsed = time.perf_counter() - start + assert result.truncated is True + assert elapsed < 1.0 From fbb5890813513d68ddfb5a4b1172a9c732fcabc2 Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 17 Sep 2026 19:54:11 +0000 Subject: [PATCH 4/6] feat(score-breakdown): additive score_breakdown audit field on the result MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit calculate_score_with_breakdown(findings, policy=None) scores findings and returns the audit trail — {"base": 100, "penalties": [{"code", "severity", "points"(negative)}...], "final"} — one entry per distinct gap (or code when gap is None) in first-occurrence order, matching the result's gaps order. calculate_score keeps its v0.2 signature and delegates to it, so both stay in lockstep by construction. AnalysisResult gains an additive score_breakdown field (default None, deep- copied in to_dict so a serialized snapshot can never be rewritten by later mutation). With default policy the penalties mirror the v0.2 deductions exactly, so serialized defaults differ only by the new keys. Co-authored-by: Kalisetti Nihanth Naidu --- inputguard/analyzer.py | 9 ++- inputguard/scorer.py | 37 +++++++-- inputguard/types.py | 13 ++- tests/test_score_breakdown.py | 147 ++++++++++++++++++++++++++++++++++ 4 files changed, 195 insertions(+), 11 deletions(-) create mode 100644 tests/test_score_breakdown.py diff --git a/inputguard/analyzer.py b/inputguard/analyzer.py index 5779960..5b68c44 100644 --- a/inputguard/analyzer.py +++ b/inputguard/analyzer.py @@ -18,7 +18,11 @@ from inputguard.policy import Policy from inputguard.recommender import get_recommendations from inputguard.registry import REGISTRY -from inputguard.scorer import calculate_score, get_status, require_known_severity +from inputguard.scorer import ( + calculate_score_with_breakdown, + get_status, + require_known_severity, +) from inputguard.types import AnalysisResult, RuleFinding @@ -142,7 +146,7 @@ def analyze( seen_codes.add(finding.code) findings.append(finding) - score = calculate_score(findings, effective_policy) + score, breakdown = calculate_score_with_breakdown(findings, effective_policy) status = get_status(score, self.mode, effective_policy) # Two layers: severity decided what fired; the policy's bands decide # what happens. The borderline band is the near-miss signal just below @@ -184,6 +188,7 @@ def analyze( ), borderline=borderline, truncated=truncated, + score_breakdown=breakdown, ) diff --git a/inputguard/scorer.py b/inputguard/scorer.py index 3b1d31a..19e7226 100644 --- a/inputguard/scorer.py +++ b/inputguard/scorer.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from inputguard.policy import Policy, SEVERITIES from inputguard.types import RuleFinding @@ -44,24 +44,45 @@ def calculate_score(findings: List[RuleFinding], policy: Optional[Policy] = None would otherwise distort every score silently (the v0.2 ``.get(severity, 0)`` behavior). """ + score, _ = calculate_score_with_breakdown(findings, policy) + return score + + +def calculate_score_with_breakdown( + findings: List[RuleFinding], policy: Optional[Policy] = None +) -> Tuple[int, Dict[str, Any]]: + """Score the findings and also return the additive audit breakdown. + + The breakdown is ``{"base": 100, "penalties": [...], "final": score}`` + where each penalty records ``{"code", "severity", "points"}`` (negative) — + one entry per distinct gap (or code, when gap is None), in first-occurrence + order. Ops teams use it to audit exactly which finding cost which points. + Unknown severities raise ``ValueError`` before any scoring happens. + """ p = Policy() if policy is None else policy # Validate every severity before scoring so a bad finding fails loudly # even when a later finding would otherwise mask it in the dedup loop. for f in findings: _severity_rank(f.severity) - highest_by_gap: Dict[str, str] = {} + highest_by_gap: Dict[str, RuleFinding] = {} for f in findings: key = f.gap if f.gap is not None else f.code current = highest_by_gap.get(key) - if current is None or _severity_rank(f.severity) > _severity_rank(current): - highest_by_gap[key] = f.severity + if current is None or _severity_rank(f.severity) > _severity_rank( + current.severity + ): + highest_by_gap[key] = f score = 100 - for severity in highest_by_gap.values(): - score -= _penalty_for(severity, p) - - return max(0, min(100, score)) + penalties: List[Dict[str, Any]] = [] + for f in highest_by_gap.values(): + points = _penalty_for(f.severity, p) + penalties.append({"code": f.code, "severity": f.severity, "points": -points}) + score -= points + + final = max(0, min(100, score)) + return final, {"base": 100, "penalties": penalties, "final": final} def _penalty_for(severity: str, policy: Policy) -> int: diff --git a/inputguard/types.py b/inputguard/types.py index 94f01d5..73bde72 100644 --- a/inputguard/types.py +++ b/inputguard/types.py @@ -1,7 +1,7 @@ from __future__ import annotations from dataclasses import dataclass, field, asdict -from typing import List, Optional +from typing import Any, Dict, List, Optional @dataclass(frozen=True) @@ -12,6 +12,15 @@ class RuleFinding: gap: Optional[str] = None +def _copy_breakdown(breakdown: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: + """Copy the breakdown so a result and its to_dict never share mutable state.""" + if breakdown is None: + return None + copied = dict(breakdown) + copied["penalties"] = [dict(p) for p in breakdown["penalties"]] + return copied + + @dataclass(frozen=True) class AnalysisResult: status: str @@ -37,6 +46,7 @@ class AnalysisResult: # more pass" band just below ready. borderline: bool = False truncated: bool = False + score_breakdown: Optional[Dict[str, Any]] = None def to_dict(self) -> dict: return { @@ -53,6 +63,7 @@ def to_dict(self) -> dict: "degradation_note": self.degradation_note, "borderline": self.borderline, "truncated": self.truncated, + "score_breakdown": _copy_breakdown(self.score_breakdown), } def is_clear(self) -> bool: diff --git a/tests/test_score_breakdown.py b/tests/test_score_breakdown.py new file mode 100644 index 0000000..89096ff --- /dev/null +++ b/tests/test_score_breakdown.py @@ -0,0 +1,147 @@ +"""Score-breakdown tests: the additive audit trail on AnalysisResult. + +Covers the breakdown slice of the v0.3 spec (art_bTvdPdJS §2/§5): base, +per-distinct-gap penalties (negative points), final — owned by the +policy-calibration PR. All expected values below were observed via runtime +probes against the release branch. +""" + +from __future__ import annotations + +import json + +import pytest + +from inputguard import InputGuard, Policy, RuleFinding +from inputguard.scorer import calculate_score, calculate_score_with_breakdown + + +def test_breakdown_shape_end_to_end(): + result = InputGuard().analyze("fix the bug in my app") + assert result.clarity_score == 35 + assert result.score_breakdown == { + "base": 100, + "penalties": [ + {"code": "missing_error_message", "severity": "high", "points": -25}, + { + "code": "missing_expected_vs_actual", + "severity": "high", + "points": -25, + }, + { + "code": "missing_debug_code_context", + "severity": "medium", + "points": -15, + }, + ], + "final": 35, + } + + +def test_breakdown_final_matches_clarity_score(): + for text in ("build a REST API", "do something now", "fix the bug in my app"): + result = InputGuard().analyze(text) + assert result.score_breakdown is not None + assert result.score_breakdown["final"] == result.clarity_score + + +def test_breakdown_dedupes_shared_gap(): + # missing_language and intent_without_language share gap + # "programming language" — one penalty entry for that gap, first wins. + result = InputGuard().analyze("I need to build an app") + assert result.clarity_score == 60 + assert result.score_breakdown == { + "base": 100, + "penalties": [ + {"code": "missing_language", "severity": "high", "points": -25}, + {"code": "missing_output_format", "severity": "medium", "points": -15}, + ], + "final": 60, + } + + +def test_breakdown_penalty_order_matches_findings_order(): + result = InputGuard().analyze("fix the bug in my app") + assert result.score_breakdown is not None + # Every finding here has a distinct gap, so penalty order == findings order. + assert [p["code"] for p in result.score_breakdown["penalties"]] == [ + f.code for f in result.findings + ] + + +def test_breakdown_penalties_follow_policy_penalties(): + guard = InputGuard(policy=Policy(penalty_low=1, penalty_medium=3, penalty_high=10)) + result = guard.analyze("build a REST API") # two high findings + assert result.score_breakdown == { + "base": 100, + "penalties": [ + {"code": "missing_language", "severity": "high", "points": -10}, + {"code": "missing_api_structure", "severity": "high", "points": -10}, + ], + "final": 80, + } + assert result.clarity_score == 80 + + +def test_breakdown_clean_input_has_no_penalties(): + result = InputGuard().analyze( + "Build a web app using React and FastAPI. " + "It needs user authentication with JWT. " + "Store tasks in a PostgreSQL database with title, description, and due date fields. " + "Expose REST API endpoints: GET /tasks, POST /tasks, DELETE /tasks/{id}." + ) + assert result.score_breakdown == {"base": 100, "penalties": [], "final": 100} + + +def test_breakdown_present_in_to_dict_and_json_serializable(): + d = InputGuard().analyze("build a REST API").to_dict() + assert d["score_breakdown"]["final"] == 50 + assert json.dumps(d) # must not raise + + +def test_to_dict_breakdown_is_a_copy(): + # The dict to_dict() returns must be detached from the result's own + # breakdown object: mutating the result afterwards cannot rewrite history. + result = InputGuard().analyze("build a REST API") + snapshot = result.to_dict() + assert result.score_breakdown is not None + result.score_breakdown["penalties"][0]["points"] = 0 # type: ignore[index] + assert snapshot["score_breakdown"]["penalties"][0]["points"] == -25 + assert result.to_dict()["score_breakdown"]["penalties"][0]["points"] == 0 + + +def test_breakdown_clamps_at_zero(): + findings = [ + RuleFinding(code=f"c{i}", message="m", severity="high") for i in range(5) + ] + score, breakdown = calculate_score_with_breakdown(findings) + assert score == 0 + assert breakdown["final"] == 0 + assert len(breakdown["penalties"]) == 5 + assert sum(p["points"] for p in breakdown["penalties"]) == -125 + + +def test_breakdown_gap_none_dedupes_by_code(): + findings = [ + RuleFinding(code="a", message="m", severity="low"), + RuleFinding(code="a", message="m", severity="high"), + ] + score, breakdown = calculate_score_with_breakdown(findings) + assert score == 75 + assert breakdown["penalties"] == [ + {"code": "a", "severity": "high", "points": -25} + ] + + +def test_calculate_score_keeps_v02_signature_and_matches_breakdown(): + findings = [RuleFinding(code="a", message="m", severity="medium", gap="g")] + assert calculate_score(findings) == 85 + score, breakdown = calculate_score_with_breakdown(findings) + assert score == 85 + assert breakdown["final"] == 85 + + +def test_breakdown_unknown_severity_raises_before_scoring(): + findings = [RuleFinding(code="a", message="m", severity="typo")] + with pytest.raises(ValueError, match="Unknown severity: 'typo'"): + calculate_score_with_breakdown(findings) From ae2f59329f9775cd3468f832b8bbd0568ef166fe Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 17 Sep 2026 19:56:51 +0000 Subject: [PATCH 5/6] test(pin-v02-defaults): pin every Policy default byte-exactly to v0.2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dedicated pinning tests guard the calibration-drift risk the spec flags: each Policy default (ready_at 85, usable_at 60, strict_clarify_at 65, penalties 5/15/25, min_words 3, max_chars 10000, borderline_at 74) is asserted three ways — dataclass field defaults, constructed instance values, and the literal source line (underscore separators normalized) — plus a no-unpinned-fields guard so future fields cannot ship uncalibrated. Behavioral pins reproduce the observed v0.2 reference outputs under default policy (75/usable_with_warnings, 35/needs_clarification, 50, strict blocked/needs_clarification) and the v0.2 result and recommendation key shapes. Co-authored-by: Kalisetti Nihanth Naidu --- tests/test_policy_defaults.py | 120 ++++++++++++++++++++++++++++++++++ 1 file changed, 120 insertions(+) create mode 100644 tests/test_policy_defaults.py diff --git a/tests/test_policy_defaults.py b/tests/test_policy_defaults.py new file mode 100644 index 0000000..8aca8c7 --- /dev/null +++ b/tests/test_policy_defaults.py @@ -0,0 +1,120 @@ +"""Byte-exact pinning of every Policy default against the v0.2 constants. + +The v0.2 constants are load-bearing for existing users (spec art_bTvdPdJS, +"Calibration drift" risk): these tests fail if any default drifts. Sources: +SEVERITY_PENALTIES and the four threshold constants in the v0.2 scorer, and +the three-word structural floor of the insufficient_context rule. +""" + +from __future__ import annotations + +import dataclasses +import inspect +import re + +import pytest + +from inputguard import InputGuard, Policy + +V02_THRESHOLD_CONSTANTS = { + "ready_at": 85, # warning and strict ready floor + "usable_at": 60, # warning-mode band floor + "strict_clarify_at": 65, # strict-mode needs_clarification floor + "penalty_low": 5, + "penalty_medium": 15, + "penalty_high": 25, + "min_words": 3, # insufficient_context structural floor + "max_chars": 10_000, # v0.3 default cap (v0.2 had none) + "borderline_at": 74, # near-miss band just below ready +} + + +def test_every_policy_default_is_pinned(): + fields = {f.name: f.default for f in dataclasses.fields(Policy)} + for name, value in V02_THRESHOLD_CONSTANTS.items(): + assert fields[name] == value, f"Policy.{name} drifted from v0.2 ({value})" + + +def test_no_unpinned_numeric_fields(): + # Every int field must appear in the pin table so future fields cannot + # ship uncalibrated. + int_fields = { + f.name + for f in dataclasses.fields(Policy) + if f.type in ("int",) or getattr(f.type, "__name__", "") == "int" + } + assert int_fields == set(V02_THRESHOLD_CONSTANTS) + + +def test_dataclass_defaults_are_the_source_of_truth(): + # Constructing with zero arguments must equal the pinned table — no + # hidden overrides inside __post_init__. + policy = Policy() + for name, value in V02_THRESHOLD_CONSTANTS.items(): + assert getattr(policy, name) == value + + +def test_defaults_are_frozen(): + policy = Policy() + with pytest.raises(dataclasses.FrozenInstanceError): + policy.ready_at = 90 # type: ignore[misc] + + +def test_source_defaults_match_the_pinned_table(): + # Guard against the dataclass defaults themselves drifting (the pinning + # tests above read instantiated values; this reads the source). Underscore + # digit separators are normalized before comparison. + source = inspect.getsource(Policy) + for name, value in V02_THRESHOLD_CONSTANTS.items(): + match = re.search(rf"^\s*{name}: int = ([0-9_]+),?\s*$", source, re.MULTILINE) + assert match is not None, f"source default for {name} missing or reformatted" + assert int(match.group(1).replace("_", "")) == value + + +# --- behavioral pins: v0.2 reference outputs under default policy -------------- + + +def _result(text, **kwargs): + return InputGuard(**kwargs).analyze(text).to_dict() + + +def test_reference_outputs_unchanged_warning_mode(): + # Scores observed via runtime probes on the release branch. + assert _result("do something now")["clarity_score"] == 75 + assert _result("do something now")["status"] == "usable_with_warnings" + assert _result("fix the bug in my app")["clarity_score"] == 35 + assert _result("fix the bug in my app")["status"] == "needs_clarification" + assert _result("build a REST API")["clarity_score"] == 50 + + +def test_reference_outputs_unchanged_strict_mode(): + # 50 < 65 strict floor -> blocked; 75 in [65, 85) -> needs_clarification. + assert _result("build a REST API", mode="strict")["status"] == "blocked" + assert _result("do something now", mode="strict")["status"] == "needs_clarification" + + +def test_default_result_keys_are_a_superset_of_v02(): + result = _result("build a REST API") + assert set(result).issuperset( + { + "status", + "clarity_score", + "detected_intent", + "gaps", + "recommendations", + "findings", + "interpretation_note", + } + ) + + +def test_v02_recommendation_keys_unchanged(): + recs = _result("build a REST API")["recommendations"] + assert recs, "expected at least one recommendation" + for rec in recs: + assert set(rec) == { + "gap", + "what_is_missing", + "what_to_provide", + "why_it_matters", + } From 93380c9bc21cf305bddda2e3cf498d54fb070d0d Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 17 Sep 2026 20:02:12 +0000 Subject: [PATCH 6/6] test: integrate policy keys into sibling additive-contract tests The followups parity test and the language ordered-key-list test pin the exhaustive to_dict key set as of their own merge; the policy-calibration fields (borderline, truncated, score_breakdown) are additive per the compat contract, so the expected sets extend rather than the fields disappearing. Co-authored-by: Kalisetti Nihanth Naidu --- tests/test_followups.py | 4 ++++ tests/test_language.py | 6 +++++- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/tests/test_followups.py b/tests/test_followups.py index 2f93711..f468deb 100644 --- a/tests/test_followups.py +++ b/tests/test_followups.py @@ -298,6 +298,10 @@ def test_english_score_parity_with_v02(text): "detected_language", "heuristic_coverage", "degradation_note", + # policy-calibration PR: borderline band, input cap, score breakdown + "borderline", + "truncated", + "score_breakdown", } v02_shaped = AnalysisResult( status=result.status, diff --git a/tests/test_language.py b/tests/test_language.py index d235d0f..997a1ab 100644 --- a/tests/test_language.py +++ b/tests/test_language.py @@ -321,7 +321,8 @@ def test_to_dict_includes_additive_fields(): assert isinstance(d["degradation_note"], str) # Merged additive contract: v0.2 keys keep their names and relative # order, follow_ups slots in after recommendations (questions engine), - # and the three probe keys trail. + # the three probe keys trail, and the policy-calibration keys append + # after those (borderline band, input cap, score breakdown). assert list(d) == [ "status", "clarity_score", @@ -334,6 +335,9 @@ def test_to_dict_includes_additive_fields(): "detected_language", "heuristic_coverage", "degradation_note", + "borderline", + "truncated", + "score_breakdown", ] json.dumps(d, ensure_ascii=False) # JSON-serializable as before