From 7523fc0c2adcc9ee4069540263a5da9713ec0c86 Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 17 Sep 2026 19:09:17 +0000 Subject: [PATCH 1/4] feat(rule-protocol-registry): typed Rule protocol and in-process registry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Open the engine (spec art_bTvdPdJS §1): rules and domains become registered data instead of code paths. - inputguard/registry.py: runtime-checkable Rule protocol (id, domain, severity, gap + check(text, intent)) and RuleRegistry with register_rule (decorator and imperative forms) and register_domain (priority-ordered intent signals, single empty-terms fallback intent). Duplicate rule ids, unknown severities, and — once a domain exists — undeclared rule domains raise at registration time. - detector.py: the coding priority chain becomes INTENT_SIGNALS data; detect_intent gains an optional signals parameter so registered domains run the same chain; normalize() becomes public so analyze() can honor the "rules receive normalized text" contract. Nothing consumes the registry yet; behavior is unchanged (89 tests green). Zero new dependencies. Co-authored-by: Kalisetti Nihanth Naidu --- inputguard/detector.py | 56 +++++-- inputguard/registry.py | 327 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 370 insertions(+), 13 deletions(-) create mode 100644 inputguard/registry.py diff --git a/inputguard/detector.py b/inputguard/detector.py index 35a17eb..31db70e 100644 --- a/inputguard/detector.py +++ b/inputguard/detector.py @@ -2,6 +2,7 @@ import re +from typing import Iterable, Mapping, Optional, Tuple DEBUG_SIGNALS = { "error", "exception", "traceback", "not working", "isn't working", @@ -50,7 +51,20 @@ } -def _normalize(text: str) -> str: +# The coding intent chain: priority-ordered (intent, terms) pairs. The single +# intent with empty terms ("build") is the fallback. This is the single source +# of truth for detection — the rule modules key their gates off the same sets, +# and the coding domain registers this chain in the rule registry. +INTENT_SIGNALS = ( + ("debug", DEBUG_SIGNALS), + ("optimization", OPTIMIZATION_SIGNALS), + ("explanation", EXPLANATION_SIGNALS), + ("feature", FEATURE_SIGNALS), + ("build", ()), +) + + +def normalize(text: str) -> str: return re.sub(r"\s+", " ", text.strip().lower()) @@ -58,7 +72,10 @@ def _contains_any(text: str, terms) -> bool: return any(term in text for term in terms) -def detect_intent(text: str) -> str: +def detect_intent( + text: str, + signals: Optional[Iterable[Tuple[str, Iterable[str]]]] = None, +) -> str: """ Detect the intent type of a coding input. @@ -70,15 +87,28 @@ def detect_intent(text: str) -> str: feature beats build. Build is the fallback. Ambiguous inputs always resolve to the highest-priority match. + + Pass ``signals`` to run the same chain over a domain's own + priority-ordered ``(intent, terms)`` pairs — the single intent with + empty terms is the fallback. Defaults to :data:`INTENT_SIGNALS`, the + built-in coding chain. """ - normalized = _normalize(text) - - if _contains_any(normalized, DEBUG_SIGNALS): - return "debug" - if _contains_any(normalized, OPTIMIZATION_SIGNALS): - return "optimization" - if _contains_any(normalized, EXPLANATION_SIGNALS): - return "explanation" - if _contains_any(normalized, FEATURE_SIGNALS): - return "feature" - return "build" + normalized = normalize(text) + + chain = INTENT_SIGNALS if signals is None else tuple( + signals.items() if isinstance(signals, Mapping) else signals + ) + + fallback: Optional[str] = None + for intent, terms in chain: + if not terms: + fallback = intent + elif _contains_any(normalized, terms): + return intent + + if fallback is not None: + return fallback + raise ValueError( + "detect_intent requires exactly one fallback intent (an entry with " + "empty signal terms); none was provided." + ) diff --git a/inputguard/registry.py b/inputguard/registry.py new file mode 100644 index 0000000..3762bcc --- /dev/null +++ b/inputguard/registry.py @@ -0,0 +1,327 @@ +"""Typed Rule protocol and the in-process registries that open the engine. + +v0.3 turns InputGuard's closed, coding-only checker into a pluggable clarity +engine: rules and domains are registered data, not code paths. + +- A rule is any object with the four members (``id``, ``domain``, + ``severity``, ``gap``) and a ``check(text, intent)`` method — the + :class:`Rule` protocol. +- A domain is a named analysis scope (``"coding"`` ships built in) that + declares its intent signals in priority order via + :meth:`RuleRegistry.register_domain`. + +Registration happens at import/startup time; ``analyze()`` only reads the +registry, preserving the thread-safe, dependency-free pipeline v0.2 +established. +""" + +from __future__ import annotations + +from typing import ( + Dict, + Iterable, + List, + Mapping, + Optional, + Protocol, + Set, + Tuple, + Union, + runtime_checkable, +) + +from inputguard.types import RuleFinding + +__all__ = [ + "KNOWN_SEVERITIES", + "REGISTRY", + "Rule", + "RuleRegistry", + "register_domain", + "register_rule", +] + +KNOWN_SEVERITIES: Tuple[str, ...] = ("low", "medium", "high") + +# Normalized domain signals: priority-ordered (intent, terms) pairs. The one +# intent with empty terms is the fallback for inputs no other intent matches. +SignalSpec = Tuple[Tuple[str, Tuple[str, ...]], ...] + +_REQUIRED_MEMBERS = ("id", "domain", "severity", "gap", "check") + + +@runtime_checkable +class Rule(Protocol): + """The v0.3 extension contract: four members and one method. + + ``id`` must be unique across the registry, ``severity`` must be one of + ``'low' | 'medium' | 'high'`` (validated at registration), and ``gap`` + groups the rule's findings for scoring dedup (``None`` dedupes by code + instead). ``check`` receives lowercased, whitespace-collapsed text plus + the intent detected for this ``analyze()`` call, and returns at most one + :class:`~inputguard.types.RuleFinding` — ``None`` when the rule does not + fire. + """ + + id: str + domain: str + severity: str + gap: Optional[str] + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + ... # pragma: no cover — protocol body + + +def _instantiate(rule_cls: type) -> Rule: + try: + return rule_cls() + except TypeError as exc: + raise TypeError( + f"Cannot register rule class {rule_cls.__name__!r}: it must be " + f"constructible with no arguments. Register an instance instead: " + f"register_rule({rule_cls.__name__}())." + ) from exc + + +def _normalize_signals( + name: str, + signals: Union[Mapping[str, Iterable[str]], Iterable[Tuple[str, Iterable[str]]]], +) -> SignalSpec: + items = tuple(signals.items() if isinstance(signals, Mapping) else signals) + if not items: + raise ValueError(f"Domain {name!r}: signals must declare at least one intent.") + + normalized: List[Tuple[str, Tuple[str, ...]]] = [] + fallbacks = 0 + for intent, terms in items: + if not isinstance(intent, str) or not intent: + raise ValueError( + f"Domain {name!r}: intent names must be non-empty strings, got {intent!r}." + ) + if isinstance(terms, str) or not isinstance(terms, Iterable): + raise TypeError( + f"Domain {name!r}: terms for intent {intent!r} must be an iterable " + f"of strings, got {terms!r}." + ) + term_tuple = tuple(terms) + for term in term_tuple: + if not isinstance(term, str) or not term: + raise ValueError( + f"Domain {name!r}: terms for intent {intent!r} must be " + f"non-empty strings, got {term!r}." + ) + if not term_tuple: + fallbacks += 1 + normalized.append((intent, term_tuple)) + + if fallbacks != 1: + raise ValueError( + f"Domain {name!r}: exactly one intent must have empty signal terms " + f"(the fallback intent); found {fallbacks}." + ) + return tuple(normalized) + + +class RuleRegistry: + """In-process registry of rules and analysis domains. + + Writes happen at import/startup; ``analyze()`` reads are plain dict + lookups and iteration over registered rules, so parallel ``analyze()`` + calls stay consistent (the v0.2 thread-safety contract). Registering + while another thread is mid-``analyze()`` is not supported. + """ + + def __init__(self) -> None: + self._rules: Dict[str, Rule] = {} + self._domains: Dict[str, SignalSpec] = {} + + # -- registration ---------------------------------------------------- + + def register_rule(self, rule: Union[type, Rule]) -> Union[type, Rule]: + """Register a rule — decorator or imperative form. + + ``@register_rule`` above a rule class registers a zero-argument + instance and returns the class unchanged; ``register_rule(instance)`` + registers the instance and returns it. + + Raises ``ValueError`` on a duplicate rule id, an unknown severity, or + — once any domain is registered — a domain no registered domain + declares as an intent. Raises ``TypeError`` when the object is not + shaped like a :class:`Rule` or the class cannot be constructed with + no arguments. + """ + if isinstance(rule, type): + instance = _instantiate(rule) + self._validate(instance) + self._add(instance) + return rule + self._validate(rule) + self._add(rule) + return rule + + def register_domain( + self, + name: str, + signals: Union[Mapping[str, Iterable[str]], Iterable[Tuple[str, Iterable[str]]]], + rules: Iterable[Union[type, Rule]] = (), + ) -> None: + """Register an analysis domain: intent signals plus its rules. + + ``signals`` maps intent names to trigger terms; insertion order is the + priority chain, and the single intent with empty terms is the fallback + for inputs no other intent matches. Every rule passed in must declare + one of those intents as its ``domain``; rules already registered (for + example via the ``@register_rule`` decorator) are reused, and the rest + are registered through the same path. + + Raises ``ValueError`` on a duplicate domain name, malformed signals + (no fallback intent), or a rule whose domain the domain does not + declare. + """ + if not isinstance(name, str) or not name: + raise ValueError(f"Domain name must be a non-empty string, got {name!r}.") + if name in self._domains: + raise ValueError( + f"Duplicate domain: {name!r}. Domains must be registered once." + ) + normalized_signals = _normalize_signals(name, signals) + declared = {intent for intent, _ in normalized_signals} + + coerced = [self._coerce(entry) for entry in rules] + for rule in coerced: + self._validate(rule) + for rule in coerced: + if rule.domain not in declared: + raise ValueError( + f"Rule {rule.id!r} declares domain {rule.domain!r}, which is not " + f"an intent of domain {name!r}. Declared intents: {sorted(declared)}." + ) + for rule in coerced: + existing = self._rules.get(rule.id) + if existing is not None and existing is not rule: + raise ValueError( + f"Duplicate rule id: {rule.id!r}. Rule ids must be unique." + ) + + # All checks passed — mutate. The domain is stored before its rules so + # the known-domain check in _add sees the intents it declares. + self._domains[name] = normalized_signals + for rule in coerced: + if rule.id not in self._rules: + self._add(rule) + + # -- reads (thread-safe: no mutation happens during analyze) ---------- + + def get_domain_signals(self, name: str) -> SignalSpec: + """Return the domain's priority-ordered ``(intent, terms)`` pairs. + + Raises ``ValueError`` for an unregistered domain — the registry-owned + replacement for v0.2's hardcoded domain whitelist. + """ + try: + return self._domains[name] + except KeyError: + registered = ", ".join(repr(d) for d in self._domains) or "none" + raise ValueError( + f"Unsupported domain: {name!r}. Registered domains: {registered}." + ) from None + + def domain_names(self) -> Tuple[str, ...]: + """Registered domain names, in registration order.""" + return tuple(self._domains) + + def rules_for_intent(self, intent: str) -> List[Rule]: + """Rules whose domain is this intent scope, in registration order.""" + return [rule for rule in self._rules.values() if rule.domain == intent] + + def rule_ids(self) -> Tuple[str, ...]: + """Registered rule ids, in registration order.""" + return tuple(self._rules) + + def rules(self) -> Tuple[Rule, ...]: + """All registered rules, in registration order.""" + return tuple(self._rules.values()) + + def get_rule(self, rule_id: str) -> Optional[Rule]: + """Return the rule registered under ``rule_id``, or ``None``.""" + return self._rules.get(rule_id) + + # -- internals --------------------------------------------------------- + + def _coerce(self, entry: Union[type, Rule]) -> Rule: + """Accept a rule class or instance. + + A class already registered (via the decorator, which instantiates) + resolves to its registered instance; a fresh class is instantiated. + """ + if not isinstance(entry, type): + return entry + probe_id = getattr(entry, "id", None) + existing = self._rules.get(probe_id) if isinstance(probe_id, str) else None + return existing if existing is not None else _instantiate(entry) + + def _validate(self, rule: Rule) -> None: + missing = [m for m in _REQUIRED_MEMBERS if not hasattr(rule, m)] + if missing: + raise TypeError( + f"Rule {rule!r} is missing required member(s): " + f"{', '.join(repr(m) for m in missing)}. A rule needs id, domain, " + "severity, gap, and a check(text, intent) method." + ) + if not isinstance(rule.id, str) or not rule.id: + raise ValueError(f"Rule id must be a non-empty string, got {rule.id!r}.") + if not isinstance(rule.domain, str) or not rule.domain: + raise ValueError( + f"Rule {rule.id!r}: domain must be a non-empty string, got {rule.domain!r}." + ) + if rule.severity not in KNOWN_SEVERITIES: + raise ValueError( + f"Rule {rule.id!r}: unknown severity {rule.severity!r}. " + f"Expected one of: 'low', 'medium', 'high'." + ) + if rule.gap is not None and (not isinstance(rule.gap, str) or not rule.gap): + raise ValueError( + f"Rule {rule.id!r}: gap must be None or a non-empty string, got {rule.gap!r}." + ) + if not callable(rule.check): + raise TypeError( + f"Rule {rule.id!r}: check must be callable — " + "check(text, intent) -> Optional[RuleFinding]." + ) + + def _add(self, rule: Rule) -> None: + if rule.id in self._rules: + raise ValueError(f"Duplicate rule id: {rule.id!r}. Rule ids must be unique.") + # Known-domain check: vacuous until the first domain registers (built-in + # rules register during package import, before any domain exists), then + # enforced for everything registered afterwards. + if self._domains and rule.domain not in self._declared_intents(): + raise ValueError( + f"Rule {rule.id!r} declares domain {rule.domain!r}, which no " + f"registered domain declares as an intent. Registered domains: " + f"{', '.join(repr(d) for d in self._domains)}." + ) + self._rules[rule.id] = rule + + def _declared_intents(self) -> Set[str]: + declared: Set[str] = set() + for signal_spec in self._domains.values(): + declared.update(intent for intent, _ in signal_spec) + return declared + + +REGISTRY = RuleRegistry() + + +def register_rule(rule: Union[type, Rule]) -> Union[type, Rule]: + """Module-level form of :meth:`RuleRegistry.register_rule` — usable as a decorator.""" + return REGISTRY.register_rule(rule) + + +def register_domain( + name: str, + signals: Union[Mapping[str, Iterable[str]], Iterable[Tuple[str, Iterable[str]]]], + rules: Iterable[Union[type, Rule]] = (), +) -> None: + """Module-level form of :meth:`RuleRegistry.register_domain`.""" + REGISTRY.register_domain(name, signals, rules) From dc1f88e244251ec54e2a1543d1cd8d7e0c57514a Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 17 Sep 2026 19:13:47 +0000 Subject: [PATCH 2/4] refactor(built-ins-via-registry): dispatch analyze() through the registry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the hardcoded domain whitelist and the if/elif intent dispatch with registry lookups (spec art_bTvdPdJS §1): - analyzer.py: analyze() resolves domain signals via REGISTRY.get_domain_signals (unknown domain still raises ValueError, now naming every registered domain) and runs rules via REGISTRY.rules_for_intent — the deduped, deterministic-finding-order contract of the v0.2 runners is preserved by the registry's registration-order iteration. - rules/*: each of the 19 built-in rules gains a thin registry adapter class exposing the v0.2 check function through the Rule protocol, registered with @register_rule — the built-ins dogfood the exact extension path users get. The coding catch-all adapter re-runs the built-in build rules internally to preserve its fires-only-when-empty semantics under the frozen check(text, intent) contract. - rules/__init__.py: registers the coding domain (INTENT_SIGNALS chain plus all 19 rule adapters) through register_domain; the v0.2 runner exports are unchanged. Behavior-identical: the 89 existing tests pass unmodified. Co-authored-by: Kalisetti Nihanth Naidu --- inputguard/analyzer.py | 40 ++++----- inputguard/rules/__init__.py | 44 ++++++++-- inputguard/rules/coding.py | 142 +++++++++++++++++++++++++++++++ inputguard/rules/debug.py | 52 +++++++++++ inputguard/rules/explanation.py | 38 +++++++++ inputguard/rules/feature.py | 52 +++++++++++ inputguard/rules/optimization.py | 52 +++++++++++ 7 files changed, 392 insertions(+), 28 deletions(-) diff --git a/inputguard/analyzer.py b/inputguard/analyzer.py index 03c3844..9eb3b2e 100644 --- a/inputguard/analyzer.py +++ b/inputguard/analyzer.py @@ -1,14 +1,11 @@ from __future__ import annotations -from typing import List +from typing import List, Set -from inputguard.detector import detect_intent +import inputguard.rules # noqa: F401 — importing registers the coding domain and the 19 built-in rules +from inputguard.detector import detect_intent, normalize from inputguard.recommender import get_recommendations -from inputguard.rules.coding import run_coding_rules -from inputguard.rules.debug import run_debug_rules -from inputguard.rules.optimization import run_optimization_rules -from inputguard.rules.explanation import run_explanation_rules -from inputguard.rules.feature import run_feature_rules +from inputguard.registry import REGISTRY from inputguard.scorer import calculate_score, get_status from inputguard.types import AnalysisResult, RuleFinding @@ -36,29 +33,26 @@ def analyze(self, user_input: str, domain: str = "coding") -> AnalysisResult: ) if not user_input.strip(): raise ValueError("user_input must be a non-empty, non-whitespace string.") - if domain != "coding": - raise ValueError( - f"Unsupported domain: {domain!r}. Phase 1 only supports 'coding'." - ) - detected_intent = detect_intent(user_input) + domain_signals = REGISTRY.get_domain_signals(domain) + + detected_intent = detect_intent(user_input, domain_signals) + normalized = normalize(user_input) - if detected_intent == "debug": - findings: List[RuleFinding] = run_debug_rules(user_input) - elif detected_intent == "optimization": - findings = run_optimization_rules(user_input) - elif detected_intent == "explanation": - findings = run_explanation_rules(user_input) - elif detected_intent == "feature": - findings = run_feature_rules(user_input) - else: - findings = run_coding_rules(user_input) + findings: List[RuleFinding] = [] + seen_codes: Set[str] = set() + for rule in REGISTRY.rules_for_intent(detected_intent): + finding = rule.check(normalized, detected_intent) + 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) gaps: List[str] = [] - seen = set() + seen: Set[str] = set() for f in findings: if f.gap is not None and f.gap not in seen: gaps.append(f.gap) diff --git a/inputguard/rules/__init__.py b/inputguard/rules/__init__.py index f7c6066..fc32218 100644 --- a/inputguard/rules/__init__.py +++ b/inputguard/rules/__init__.py @@ -1,8 +1,27 @@ -from inputguard.rules.coding import run_coding_rules -from inputguard.rules.debug import run_debug_rules -from inputguard.rules.optimization import run_optimization_rules -from inputguard.rules.explanation import run_explanation_rules -from inputguard.rules.feature import run_feature_rules +"""Built-in rule modules and the registry wiring for the coding domain. + +Importing this package registers the coding domain — its intent signals and +all 19 built-in rules — through the exact same registry path a user rule +takes. The v0.2 ``run_*_rules`` functions stay exported for backward +compatibility, but the analyzer dispatches through the registry now. +""" + +from inputguard.detector import ( + DEBUG_SIGNALS, + EXPLANATION_SIGNALS, + FEATURE_SIGNALS, + INTENT_SIGNALS, + OPTIMIZATION_SIGNALS, +) +from inputguard.registry import REGISTRY +from inputguard.rules.coding import CODING_RULES, run_coding_rules +from inputguard.rules.debug import DEBUG_RULES, run_debug_rules +from inputguard.rules.explanation import EXPLANATION_RULES, run_explanation_rules +from inputguard.rules.feature import FEATURE_RULES, run_feature_rules +from inputguard.rules.optimization import ( + OPTIMIZATION_RULES, + run_optimization_rules, +) __all__ = [ "run_coding_rules", @@ -11,3 +30,18 @@ "run_explanation_rules", "run_feature_rules", ] + +# The coding domain: intent signals in strict priority order (the single +# empty-terms entry, "build", is the fallback) and all 19 built-in rules, +# registered through the same register_domain path a user domain takes. +REGISTRY.register_domain( + "coding", + INTENT_SIGNALS, + rules=( + *CODING_RULES, + *DEBUG_RULES, + *OPTIMIZATION_RULES, + *EXPLANATION_RULES, + *FEATURE_RULES, + ), +) diff --git a/inputguard/rules/coding.py b/inputguard/rules/coding.py index eeb1025..60ca560 100644 --- a/inputguard/rules/coding.py +++ b/inputguard/rules/coding.py @@ -3,6 +3,7 @@ import re from typing import List, Optional, Set +from inputguard.registry import register_rule from inputguard.types import RuleFinding @@ -299,3 +300,144 @@ def run_coding_rules(text: str) -> List[RuleFinding]: seen_codes.add(catchall.code) return findings + + +# Registry adapters: the v0.2 check functions above stay the single home of +# the rule logic; these classes expose it through the v0.3 Rule protocol and +# register it through the same path a user rule takes. + + +@register_rule +class MissingLanguageRule: + """Registry adapter for check_missing_language.""" + + id = "missing_language" + domain = "build" + severity = "high" + gap = "programming language" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_language(text) + + +@register_rule +class MissingApiStructureRule: + """Registry adapter for check_missing_api_structure.""" + + id = "missing_api_structure" + domain = "build" + severity = "high" + gap = "api structure" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_api_structure(text) + + +@register_rule +class MissingDataModelRule: + """Registry adapter for check_missing_data_model.""" + + id = "missing_data_model" + domain = "build" + severity = "high" + gap = "data model" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_data_model(text) + + +@register_rule +class MissingIntegrationSpecificsRule: + """Registry adapter for check_missing_integration_specifics.""" + + id = "missing_integration_specifics" + domain = "build" + severity = "medium" + gap = "integration specifics" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_integration_specifics(text) + + +@register_rule +class MissingAuthTypeRule: + """Registry adapter for check_missing_auth_type.""" + + id = "missing_auth_type" + domain = "build" + severity = "high" + gap = "authentication type" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_auth_type(text) + + +@register_rule +class MissingOutputFormatRule: + """Registry adapter for check_missing_output_format.""" + + id = "missing_output_format" + domain = "build" + severity = "medium" + gap = "output format" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_output_format(text) + + +@register_rule +class IntentWithoutDetailRule: + """Registry adapter for _check_intent_without_detail.""" + + id = "intent_without_language" + domain = "build" + severity = "high" + gap = "programming language" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return _check_intent_without_detail(text) + + +@register_rule +class InsufficientContextRule: + """Catch-all safety net for build-intent input (registry adapter). + + v0.2's run_coding_rules passed this rule the findings collected so far and + it fired only when that list was empty. A Rule sees only (text, intent), + so the adapter re-runs the other built-in build rules — they are pure + functions, so the verdict is identical. Findings from user-registered + rules are not visible here: the catch-all suppresses on the built-in + build rules only. + """ + + id = "insufficient_context" + domain = "build" + severity = "high" + gap = "task context" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + normalized = _normalize(text) + seen_codes: Set[str] = set() + prior: List[RuleFinding] = [] + for check_fn in _CHECKS: + result = check_fn(normalized) + if result is not None and result.code not in seen_codes: + prior.append(result) + seen_codes.add(result.code) + intent_finding = _check_intent_without_detail(normalized) + if intent_finding is not None and intent_finding.code not in seen_codes: + prior.append(intent_finding) + seen_codes.add(intent_finding.code) + return _check_insufficient_context(normalized, prior) + + +CODING_RULES = ( + MissingLanguageRule, + MissingApiStructureRule, + MissingDataModelRule, + MissingIntegrationSpecificsRule, + MissingAuthTypeRule, + MissingOutputFormatRule, + IntentWithoutDetailRule, + InsufficientContextRule, +) diff --git a/inputguard/rules/debug.py b/inputguard/rules/debug.py index 9369619..3067ae7 100644 --- a/inputguard/rules/debug.py +++ b/inputguard/rules/debug.py @@ -4,6 +4,7 @@ from typing import List, Optional from inputguard.detector import DEBUG_SIGNALS +from inputguard.registry import register_rule from inputguard.types import RuleFinding @@ -103,3 +104,54 @@ def run_debug_rules(text: str) -> List[RuleFinding]: if result: findings.append(result) return _dedupe(findings) + + +# Registry adapters: the v0.2 check functions above stay the single home of +# the rule logic; these classes expose it through the v0.3 Rule protocol and +# register it through the same path a user rule takes. + + +@register_rule +class MissingErrorMessageRule: + """Registry adapter for check_missing_error_message.""" + + id = "missing_error_message" + domain = "debug" + severity = "high" + gap = "error description" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_error_message(text) + + +@register_rule +class MissingExpectedVsActualRule: + """Registry adapter for check_missing_expected_vs_actual.""" + + id = "missing_expected_vs_actual" + domain = "debug" + severity = "high" + gap = "expected vs actual behavior" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_expected_vs_actual(text) + + +@register_rule +class MissingDebugCodeContextRule: + """Registry adapter for check_missing_debug_code_context.""" + + id = "missing_debug_code_context" + domain = "debug" + severity = "medium" + gap = "code context" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_debug_code_context(text) + + +DEBUG_RULES = ( + MissingErrorMessageRule, + MissingExpectedVsActualRule, + MissingDebugCodeContextRule, +) diff --git a/inputguard/rules/explanation.py b/inputguard/rules/explanation.py index ef356f4..8a5a5cf 100644 --- a/inputguard/rules/explanation.py +++ b/inputguard/rules/explanation.py @@ -4,6 +4,7 @@ from typing import List, Optional from inputguard.detector import EXPLANATION_SIGNALS +from inputguard.registry import register_rule from inputguard.types import RuleFinding @@ -84,3 +85,40 @@ def run_explanation_rules(text: str) -> List[RuleFinding]: if result: findings.append(result) return _dedupe(findings) + + +# Registry adapters: the v0.2 check functions above stay the single home of +# the rule logic; these classes expose it through the v0.3 Rule protocol and +# register it through the same path a user rule takes. + + +@register_rule +class MissingCodeReferenceRule: + """Registry adapter for check_missing_code_reference.""" + + id = "missing_code_reference" + domain = "explanation" + severity = "high" + gap = "code reference" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_code_reference(text) + + +@register_rule +class MissingExplanationDepthRule: + """Registry adapter for check_missing_explanation_depth.""" + + id = "missing_explanation_depth" + domain = "explanation" + severity = "low" + gap = "explanation depth" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_explanation_depth(text) + + +EXPLANATION_RULES = ( + MissingCodeReferenceRule, + MissingExplanationDepthRule, +) diff --git a/inputguard/rules/feature.py b/inputguard/rules/feature.py index de629fb..7afa551 100644 --- a/inputguard/rules/feature.py +++ b/inputguard/rules/feature.py @@ -4,6 +4,7 @@ from typing import List, Optional from inputguard.detector import FEATURE_SIGNALS +from inputguard.registry import register_rule from inputguard.types import RuleFinding @@ -104,3 +105,54 @@ def run_feature_rules(text: str) -> List[RuleFinding]: if result: findings.append(result) return _dedupe(findings) + + +# Registry adapters: the v0.2 check functions above stay the single home of +# the rule logic; these classes expose it through the v0.3 Rule protocol and +# register it through the same path a user rule takes. + + +@register_rule +class MissingExistingStackRule: + """Registry adapter for check_missing_existing_stack.""" + + id = "missing_existing_stack" + domain = "feature" + severity = "high" + gap = "existing stack" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_existing_stack(text) + + +@register_rule +class MissingFeatureScopeRule: + """Registry adapter for check_missing_feature_scope.""" + + id = "missing_feature_scope" + domain = "feature" + severity = "high" + gap = "feature scope" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_feature_scope(text) + + +@register_rule +class MissingCompletionCriteriaRule: + """Registry adapter for check_missing_completion_criteria.""" + + id = "missing_completion_criteria" + domain = "feature" + severity = "low" + gap = "completion criteria" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_completion_criteria(text) + + +FEATURE_RULES = ( + MissingExistingStackRule, + MissingFeatureScopeRule, + MissingCompletionCriteriaRule, +) diff --git a/inputguard/rules/optimization.py b/inputguard/rules/optimization.py index d89d34b..031346c 100644 --- a/inputguard/rules/optimization.py +++ b/inputguard/rules/optimization.py @@ -4,6 +4,7 @@ from typing import List, Optional from inputguard.detector import OPTIMIZATION_SIGNALS +from inputguard.registry import register_rule from inputguard.types import RuleFinding @@ -103,3 +104,54 @@ def run_optimization_rules(text: str) -> List[RuleFinding]: if result: findings.append(result) return _dedupe(findings) + + +# Registry adapters: the v0.2 check functions above stay the single home of +# the rule logic; these classes expose it through the v0.3 Rule protocol and +# register it through the same path a user rule takes. + + +@register_rule +class MissingOptimizationTargetRule: + """Registry adapter for check_missing_optimization_target.""" + + id = "missing_optimization_target" + domain = "optimization" + severity = "high" + gap = "optimization target" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_optimization_target(text) + + +@register_rule +class MissingPerformanceBaselineRule: + """Registry adapter for check_missing_performance_baseline.""" + + id = "missing_performance_baseline" + domain = "optimization" + severity = "medium" + gap = "performance baseline" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_performance_baseline(text) + + +@register_rule +class MissingOptimizationConstraintRule: + """Registry adapter for check_missing_optimization_constraint.""" + + id = "missing_optimization_constraint" + domain = "optimization" + severity = "low" + gap = "optimization constraint" + + def check(self, text: str, intent: str) -> Optional[RuleFinding]: + return check_missing_optimization_constraint(text) + + +OPTIMIZATION_RULES = ( + MissingOptimizationTargetRule, + MissingPerformanceBaselineRule, + MissingOptimizationConstraintRule, +) From 903b9b87191f81667b62eb0c43f122366b43b642 Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 17 Sep 2026 19:15:03 +0000 Subject: [PATCH 3/4] fix(scorer-loud-failures): unknown severities raise; unknown gaps keep advice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Close the two silent-failure edges the survey pinned (art_CnghyDxp §2.3, §2.4) per spec art_bTvdPdJS §6: - scorer.py: calculate_score validates every finding's severity up front and raises ValueError on an unknown one — the silent zero penalty from .get(severity, 0) is gone (rank comparisons and penalty lookups go through validated paths). - recommender.py: get_recommendations no longer drops gaps without a curated entry; unknown gaps get a documented four-key generic fallback (_fallback_recommendation), keeping advice complete for rule authors who ship a new gap. Existing behavior for known severities and gaps is unchanged: the 89 existing tests pass unmodified. Co-authored-by: Kalisetti Nihanth Naidu --- inputguard/recommender.py | 24 ++++++++++++++++++++++-- inputguard/scorer.py | 25 +++++++++++++++++++++++-- 2 files changed, 45 insertions(+), 4 deletions(-) diff --git a/inputguard/recommender.py b/inputguard/recommender.py index f37275c..56c42bf 100644 --- a/inputguard/recommender.py +++ b/inputguard/recommender.py @@ -115,9 +115,29 @@ } +def _fallback_recommendation(gap: str) -> dict: + """Generic four-key advice for a gap with no curated entry. + + The documented fallback (spec art_bTvdPdJS §6): unknown gaps keep their + advice complete instead of vanishing silently from the recommendations + list — the v0.2 behavior that let rule authors ship empty advice. + """ + return { + "gap": gap, + "what_is_missing": f"You haven't provided the {gap} needed to act on this request.", + "what_to_provide": f"Describe the {gap} explicitly — one or two concrete sentences is enough.", + "why_it_matters": f"Without the {gap}, the request can be read several ways and the answer may miss what you actually need.", + } + + def get_recommendations(gaps: List[str]) -> List[dict]: + """Build one four-key recommendation per gap, in input order. + + A gap without a curated entry gets the documented fallback + (see :func:`_fallback_recommendation`) — never a silent drop. + """ out: List[dict] = [] for gap in gaps: - if gap in _RECOMMENDATIONS: - out.append(dict(_RECOMMENDATIONS[gap])) + entry = _RECOMMENDATIONS.get(gap) + out.append(dict(entry) if entry is not None else _fallback_recommendation(gap)) return out diff --git a/inputguard/scorer.py b/inputguard/scorer.py index 113fba7..d6894ad 100644 --- a/inputguard/scorer.py +++ b/inputguard/scorer.py @@ -18,17 +18,38 @@ _STRICT_NEEDS = 65 +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 + + def calculate_score(findings: List[RuleFinding]) -> 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)`` + behavior). + """ + # 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] = {} 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.get(f.severity, 0) > _SEVERITY_RANK.get(current, 0): + if current is None or _severity_rank(f.severity) > _severity_rank(current): highest_by_gap[key] = f.severity score = 100 for severity in highest_by_gap.values(): - score -= SEVERITY_PENALTIES.get(severity, 0) + score -= SEVERITY_PENALTIES[severity] return max(0, min(100, score)) From adaf65bb43ebddd7ddfbae0f73513ddf4ce7401a Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 17 Sep 2026 19:16:24 +0000 Subject: [PATCH 4/4] test(registry-invariants): registry contract and parity tests New tests/test_registry.py pins the foundation's acceptance points: - dogfooding: all 19 built-in rules registered through the registry, coding domain signals in priority order with a single fallback intent, deterministic build-intent rule order; - validation: duplicate rule id, unknown severity, and unknown domain raise ValueError at registration; decorator form registers and satisfies the Rule protocol; - loud failures: unknown severity raises in calculate_score (including alongside known severities); unknown gaps get the documented four-key fallback, never a silent drop; - end to end: a custom registered rule changes a result (finding, gap, score -15, fallback recommendation); a registered custom domain analyzes; duplicate domain registration raises; - invariants: the registry-walking completeness check (every built-in gap has a complete recommendation entry), dispatch parity with the v0.2 runners across nine inputs, and 65 parallel analyze() calls across 8 threads matching sequential results. Full suite: 114 passed (89 pre-existing, unmodified). Co-authored-by: Kalisetti Nihanth Naidu --- tests/test_registry.py | 306 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 306 insertions(+) create mode 100644 tests/test_registry.py diff --git a/tests/test_registry.py b/tests/test_registry.py new file mode 100644 index 0000000..7e0da55 --- /dev/null +++ b/tests/test_registry.py @@ -0,0 +1,306 @@ +"""Registry contract tests for the v0.3 foundation. + +Covers the acceptance points of the rule-registry PR: registration +validation (duplicate id, unknown severity, unknown domain), the +unknown-gap fallback, a custom rule changing a result end to end, a +registered custom domain, the registry-walking completeness invariant, +dispatch parity with the v0.2 runners, and thread-safe parallel +analyze() consistency. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from inputguard import InputGuard, RuleFinding +from inputguard.recommender import _RECOMMENDATIONS, get_recommendations +from inputguard.registry import REGISTRY, Rule, register_domain, register_rule +from inputguard.rules.coding import run_coding_rules +from inputguard.rules.debug import run_debug_rules +from inputguard.rules.explanation import run_explanation_rules +from inputguard.rules.feature import run_feature_rules +from inputguard.rules.optimization import run_optimization_rules +from inputguard.scorer import calculate_score + +# The 19 built-in rule ids, as documented in the README table. +EXPECTED_BUILTIN_IDS = { + "missing_language", + "missing_api_structure", + "missing_data_model", + "missing_integration_specifics", + "missing_auth_type", + "missing_output_format", + "intent_without_language", + "insufficient_context", + "missing_error_message", + "missing_expected_vs_actual", + "missing_debug_code_context", + "missing_optimization_target", + "missing_performance_baseline", + "missing_optimization_constraint", + "missing_code_reference", + "missing_explanation_depth", + "missing_existing_stack", + "missing_feature_scope", + "missing_completion_criteria", +} + +_V02_INTENT_RUNNERS = { + "build": run_coding_rules, + "debug": run_debug_rules, + "optimization": run_optimization_rules, + "explanation": run_explanation_rules, + "feature": run_feature_rules, +} + + +def _test_rule(rule_id="test_rule", domain="debug", severity="low", gap=None, finding=None): + """Build an instance of a minimal rule class with configurable members.""" + + class TestRule: + pass + + TestRule.id = rule_id + TestRule.domain = domain + TestRule.severity = severity + TestRule.gap = gap + TestRule.check = lambda self, text, intent: finding + return TestRule() + + +@pytest.fixture +def registry_isolation(): + """Snapshot the registry around tests that mutate it.""" + rules_before = dict(REGISTRY._rules) + domains_before = dict(REGISTRY._domains) + yield REGISTRY + REGISTRY._rules.clear() + REGISTRY._rules.update(rules_before) + REGISTRY._domains.clear() + REGISTRY._domains.update(domains_before) + + +# --- the 19 built-ins dogfood the registry path --------------------------- + + +def test_all_builtin_rules_registered_through_registry(): + assert len(REGISTRY.rule_ids()) == 19 + assert set(REGISTRY.rule_ids()) == EXPECTED_BUILTIN_IDS + for rule in REGISTRY.rules(): + assert isinstance(rule, Rule) + assert rule.severity in ("low", "medium", "high") + + +def test_coding_domain_registered_with_priority_signals(): + assert REGISTRY.domain_names() == ("coding",) + chain = REGISTRY.get_domain_signals("coding") + assert [intent for intent, _ in chain] == [ + "debug", + "optimization", + "explanation", + "feature", + "build", + ] + # The single fallback intent carries no terms. + assert [terms for _, terms in chain if not terms] == [()] + + +def test_build_intent_rule_order_is_deterministic(): + ids = [r.id for r in REGISTRY.rules_for_intent("build")] + assert ids == [ + "missing_language", + "missing_api_structure", + "missing_data_model", + "missing_integration_specifics", + "missing_auth_type", + "missing_output_format", + "intent_without_language", + "insufficient_context", + ] + + +# --- registration validation ---------------------------------------------- + + +def test_duplicate_rule_id_raises_value_error(registry_isolation): + with pytest.raises(ValueError, match="Duplicate rule id"): + register_rule(_test_rule(rule_id="missing_error_message")) + + +def test_unknown_severity_raises_value_error_at_registration(registry_isolation): + with pytest.raises(ValueError, match="unknown severity"): + register_rule(_test_rule(severity="critical")) + + +def test_unknown_domain_raises_value_error_at_registration(registry_isolation): + with pytest.raises(ValueError, match="domain"): + register_rule(_test_rule(domain="nosuchintent")) + + +def test_register_rule_decorator_form(registry_isolation): + @register_rule + class AlwaysAskForDeadlineRule: + id = "test_always_ask_deadline" + domain = "debug" + severity = "medium" + gap = "deadline" + + def check(self, text, intent): + return RuleFinding( + code=self.id, + message="Test rule fired.", + severity=self.severity, + gap=self.gap, + ) + + registered = REGISTRY.get_rule("test_always_ask_deadline") + assert registered is not None + assert isinstance(registered, AlwaysAskForDeadlineRule) + assert isinstance(registered, Rule) + + +# --- scorer / recommender loud failures ------------------------------------ + + +def test_scorer_unknown_severity_raises_value_error(): + finding = RuleFinding(code="x", message="m", severity="catastrophic") + with pytest.raises(ValueError, match="Unknown severity"): + calculate_score([finding]) + + +def test_scorer_unknown_severity_raises_even_alongside_known_severities(): + known = RuleFinding(code="y", message="m", severity="high") + unknown = RuleFinding(code="x", message="m", severity="typo") + with pytest.raises(ValueError, match="Unknown severity"): + calculate_score([known, unknown]) + + +def test_unknown_gap_gets_documented_fallback(): + recs = get_recommendations(["totally new gap"]) + assert len(recs) == 1 + entry = recs[0] + assert set(entry) == {"gap", "what_is_missing", "what_to_provide", "why_it_matters"} + assert entry["gap"] == "totally new gap" + assert all(isinstance(v, str) and v for v in entry.values()) + + +def test_unknown_gap_fallback_preserves_order_and_known_entries(): + recs = get_recommendations(["programming language", "totally new gap"]) + assert [r["gap"] for r in recs] == ["programming language", "totally new gap"] + # The known gap still gets its curated entry (copied, not shared). + assert recs[0]["why_it_matters"] == _RECOMMENDATIONS["programming language"]["why_it_matters"] + + +# --- custom rules and domains end to end ----------------------------------- + + +def test_custom_rule_changes_result(registry_isolation): + text = "fix the bug in my app" + before = InputGuard().analyze(text) + assert not any(f.code == "test_always_ask_deadline" for f in before.findings) + + finding = RuleFinding( + code="test_always_ask_deadline", + message="Test rule fired.", + severity="medium", + gap="deadline", + ) + register_rule(_test_rule("test_always_ask_deadline", finding=finding)) + + result = InputGuard().analyze(text) + codes = [f.code for f in result.findings] + assert "test_always_ask_deadline" in codes + assert "deadline" in result.gaps + assert result.clarity_score == before.clarity_score - 15 + deadline_rec = next(r for r in result.recommendations if r["gap"] == "deadline") + assert all(str(v) for v in deadline_rec.values()) + + +def test_custom_domain_analyzes_end_to_end(registry_isolation): + register_domain( + "testdom", + {"thing": ("widget",), "fallback": ()}, + rules=[], + ) + result = InputGuard().analyze("widget please", domain="testdom") + assert result.detected_intent == "thing" + assert result.findings == [] + assert result.clarity_score == 100 + assert result.status == "ready" + + +def test_duplicate_domain_registration_raises(registry_isolation): + with pytest.raises(ValueError, match="Duplicate domain"): + register_domain("coding", {"x": ("term",), "fallback": ()}) + + +# --- completeness invariant ------------------------------------------------- + + +def test_every_builtin_gap_has_recommendation_and_complete_entry(): + gaps = {rule.gap for rule in REGISTRY.rules() if rule.gap is not None} + assert gaps, "built-in rules must declare gaps" + for gap in gaps: + assert gap in _RECOMMENDATIONS, f"gap {gap!r} has no recommendation entry" + entry = _RECOMMENDATIONS[gap] + assert set(entry) == {"gap", "what_is_missing", "what_to_provide", "why_it_matters"} + assert all(isinstance(v, str) and v for v in entry.values()) + + +# --- dispatch parity with the v0.2 runners ---------------------------------- + + +def _ordered_unique_gaps(findings): + seen = set() + out = [] + for f in findings: + if f.gap is not None and f.gap not in seen: + out.append(f.gap) + seen.add(f.gap) + return out + + +@pytest.mark.parametrize( + "text", + [ + "Build a REST API using FastAPI. Store users in PostgreSQL with fields for name and email. Add email and password login.", + "make me something good", + "I need a mobile app", + "fix the bug in my app", + "the deploy fails with a TypeError: cannot read property of undefined, it should return the cached value instead", + "make my database query faster", + "explain recursion in this function", + "add search to my app", + "what does this code do", + ], +) +def test_registry_dispatch_matches_v02_runner_output(text): + result = InputGuard().analyze(text) + legacy = _V02_INTENT_RUNNERS[result.detected_intent](text) + assert [f.code for f in result.findings] == [f.code for f in legacy] + assert result.gaps == _ordered_unique_gaps(legacy) + assert result.clarity_score == calculate_score(legacy) + + +# --- thread safety ----------------------------------------------------------- + + +def test_parallel_analyze_stays_consistent(): + guard = InputGuard() + texts = [ + "fix the bug in my app", + "make my database query faster", + "what does this code do", + "add search to my app", + "Build a REST API using FastAPI. Store users in PostgreSQL. Add login.", + ] + sequential = [guard.analyze(t).to_dict() for t in texts] + + with ThreadPoolExecutor(max_workers=8) as pool: + parallel = list(pool.map(lambda t: guard.analyze(t).to_dict(), texts * 13)) + + assert len(parallel) == 65 + for offset in range(0, len(parallel), len(texts)): + assert parallel[offset : offset + len(texts)] == sequential