feat(policy): frozen validated Policy with pinned v0.2 defaults, banding, input cap, and score breakdown

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",
 ]

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,21 @@
     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_with_breakdown,
+    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 +38,52 @@
 
 
 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
+
+        # 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)
@@ -58,7 +99,7 @@ def analyze(self, user_input: str, domain: str = "coding") -> AnalysisResult:
             # 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=[],
@@ -74,28 +115,43 @@ 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)
-        status = get_status(score, self.mode)
+        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, 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
+        # 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()
@@ -130,4 +186,12 @@ def analyze(self, user_input: str, domain: str = "coding") -> AnalysisResult:
                 if probe.heuristic_coverage == COVERAGE_PARTIAL
                 else None
             ),
+            borderline=borderline,
+            truncated=truncated,
+            score_breakdown=breakdown,
         )
+
+
+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)

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)}."
+            )