fix(inference): remove ghost bridge and unowned KB rule loading

engine/inference_rule_registry.py

@@ -13,8 +13,8 @@
 
 Previously _RULE_REGISTRY was empty at startup, causing every `derived_from`
 inference rule that referenced a named rule to block with NO_RULE.
-This module registers all production inference functions and wires them
-into the DerivationGraph execution engine.
+This module registers the supported in-code inference functions and exposes
+them through the explicit execute_rule() registry boundary.
 
 Registration pattern:
     @register_inference_rule("rule_name")
@@ -47,7 +47,8 @@ class InferenceContext:
     domain_id: str
     pass_number: int
     known_fields: dict[str, Any] = field(default_factory=dict)
-    domain_kb: dict[str, Any] = field(default_factory=dict)  # from domain spec KB injection
+    # Optional caller-supplied tuning context. DomainSpec/DomainPackLoader do not populate it.
+    domain_kb: dict[str, Any] = field(default_factory=dict)
     confidence_floor: float = 0.55
 
 
@@ -132,8 +133,9 @@ def execute_rule(
 
 # ===========================================================================
 # PRODUCTION INFERENCE RULES
-# Register all domain-agnostic rules here. Domain-specific rules are
-# loaded from domain KB via load_domain_rules() below.
+# Register supported in-code inference rules here. Domain configuration remains
+# owned by the typed DomainSpec/DomainPackLoader boundary; raw rule dictionaries
+# are not loaded into this registry.
 # ===========================================================================
 
 
@@ -416,87 +418,5 @@ def infer_buyer_persona(entity: dict, ctx: InferenceContext) -> InferenceResult
     )
 
 
-# ---------------------------------------------------------------------------
-# Dynamic domain rule loader (Gap-3: KB injection pathway)
-# ---------------------------------------------------------------------------
-
-
-def load_domain_rules(domain_kb: dict[str, Any]) -> int:
-    """
-    Load domain-specific inference rules from domain KB.
-    Returns count of rules registered.
-
-    The KB may contain a 'inference_rules' list of:
-        {name: str, field: str, conditions: [...], value: Any, confidence: float}
-
-    Simple condition-based rules are auto-registered as closures.
-    Complex rules should be registered via @register_inference_rule in domain pack files.
-    """
-    rules_spec = domain_kb.get("inference_rules", [])
-    registered = 0
-    for spec in rules_spec:
-        rule_name = spec.get("name")
-        if not rule_name:
-            continue
-        if rule_name in _RULE_REGISTRY:
-            continue  # already registered — don't overwrite
-        _register_condition_rule(rule_name, spec)
-        registered += 1
-    logger.info("Loaded %d domain-specific inference rules from KB", registered)
-    return registered
-
-
-def _register_condition_rule(rule_name: str, spec: dict[str, Any]) -> None:
-    """Auto-generate and register a simple condition-based inference rule."""
-    target_field = spec["field"]
-    conditions = spec.get("conditions", [])
-    default_value = spec.get("value")
-    confidence = float(spec.get("confidence", 0.65))
-
-    def _rule(entity: dict, ctx: InferenceContext) -> InferenceResult | None:
-        for cond in conditions:
-            src_field = cond.get("source_field")
-            operator = cond.get("operator", "eq")
-            cond_value = cond.get("value")
-            entity_val = entity.get(src_field)
-            if entity_val is None:
-                return None
-            match operator:
-                case "eq":
-                    if entity_val != cond_value:
-                        return None
-                case "gt":
-                    try:
-                        if float(entity_val) <= float(cond_value):
-                            return None
-                    except (TypeError, ValueError):
-                        return None
-                case "lt":
-                    try:
-                        if float(entity_val) >= float(cond_value):
-                            return None
-                    except (TypeError, ValueError):
-                        return None
-                case "contains":
-                    if str(cond_value).lower() not in str(entity_val).lower():
-                        return None
-                case _:
-                    return None
-        output_value = spec.get("output_value", default_value)
-        if output_value is None:
-            return None
-        return InferenceResult(
-            field_name=target_field,
-            value=output_value,
-            confidence=confidence,
-            rule_name=rule_name,
-            provenance="domain_kb",
-            rationale="condition rule from KB",
-        )
-
-    _RULE_REGISTRY[rule_name] = _rule
-    logger.debug("Auto-registered condition rule: %s → %s", rule_name, target_field)
-
-
 def list_registered_rules() -> list[str]:
     return sorted(_RULE_REGISTRY.keys())

engine/startup_wiring.py

@@ -39,15 +39,6 @@ async def apply_all_gap_fixes(pg_dsn: str, neo4j_driver, domain_pack_loader) ->
     await configure_audit_pool(pg_pool)
     logger.info("startup: Gap-5 audit pool wired")
 
-    # ── Gap 3: Load domain KB rules into inference registry ──────────────────
-    from engine.inference_rule_registry import load_domain_rules
-
-    for domain_id in domain_pack_loader.list_domains():
-        spec = domain_pack_loader.load_domain(domain_id)
-        if spec and spec.kb:
-            load_domain_rules(spec.kb)
-    logger.info("startup: Gap-3 inference rules loaded")
-
     # ── Gap 2: Initialise GRAPH→ENRICH return channel ────────────────────────
     from engine.graph_return_channel import GraphToEnrichReturnChannel
 
@@ -68,8 +59,7 @@ async def apply_all_gap_fixes(pg_dsn: str, neo4j_driver, domain_pack_loader) ->
     except ImportError:
         logger.warning("startup: GDSScheduler not found — register Gap-6 hook manually")
 
-    # ── Gap 9: v1 bridge blocked by file replacement (no action needed here) ──
-    # engine/inference_bridge.py has been replaced with inference_bridge_v1_guard.py
-    # Any stray import will raise ImportError at import time, not at startup.
+    # Gap 9: the removed v1 inference bridge has no startup wiring.
+    # Do not add a successor bridge unless a real producer/consumer contract exists.
 
     logger.info("startup: all gap fixes applied successfully")

tests/gap_fixes/test_gap9_inference_authority.py

@@ -0,0 +1,50 @@
+"""Regression coverage for removed inference-bridge and raw-KB authority."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from engine import inference_rule_registry as registry
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_removed_inference_bridge_has_no_compatibility_module() -> None:
+    assert not (ROOT / "engine" / "inference_bridge.py").exists()
+
+
+def test_startup_recipe_does_not_reintroduce_undeclared_kb_loading() -> None:
+    source = (ROOT / "engine" / "startup_wiring.py").read_text(encoding="utf-8")
+    assert "spec.kb" not in source
+    assert "load_domain_rules" not in source
+
+
+def test_registry_exposes_only_supported_in_code_rule_surface() -> None:
+    assert not hasattr(registry, "load_domain_rules")
+    assert not hasattr(registry, "NaryFact")
+    assert not hasattr(registry, "to_rule_engine_format")
+
+
+def test_supported_registry_still_returns_canonical_result_type() -> None:
+    context = registry.InferenceContext(tenant_id="test", domain_id="plasticos", pass_number=1)
+    result = registry.execute_rule(
+        "infer_material_grade_from_mfi",
+        {"melt_flow_index": 5.0, "material_type": "HDPE"},
+        context,
+    )
+    assert isinstance(result, registry.InferenceResult)
+    assert result.field_name == "material_grade"
+    assert result.value == "HD_injection"
+
+
+def test_runtime_surface_contains_no_ghost_successor_reference() -> None:
+    ghost_module = "inference_bridge" + "_v2"
+    ghost_doc = "docs/migration/" + ghost_module + ".md"
+    offenders: list[str] = []
+
+    for path in sorted((ROOT / "engine").rglob("*.py")):
+        source = path.read_text(encoding="utf-8")
+        if ghost_module in source or ghost_doc in source:
+            offenders.append(str(path.relative_to(ROOT)))
+
+    assert offenders == []