feat: OVOS-CONTEXT-1 §7 uniform slot fill + INTENT-2 §4.3 slot blacklist

Fill every declared template slot of a matched intent from a live session
context entry (private <skill_id>:name over shared bare name) when the
utterance left it unresolved, independent of requires_context. A slot the
utterance binds to a blacklisted value (INTENT-2 §4.3, whole-word-sequence)
is treated as unresolved so context supplies it. Utterance-produced values
always win; requires/excludes flag gating (gate_satisfied) still governs
only the flags.

Build a per-intent slot-name index at registration by parsing {slot} markers
from the template samples, so both the spec-template and legacy paths are
covered. Pin ovos-spec-tools>=1.4.0a2 for context_slot_candidates.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: JarbasAl

padacioso/__init__.py

@@ -73,6 +73,11 @@ def fuzzy_match(x, against):
 class IntentContainer:
     def __init__(self, fuzz=False, n_workers=4):
         self.intent_samples, self.entity_samples = {}, {}
+        # OVOS-CONTEXT-1 §7 — per-intent set of declared template slot names,
+        # parsed from the ``{slot}`` markers of the samples at registration.
+        # Consumed by the pipeline to offer live context entries as slot
+        # candidates for slots the utterance itself left unresolved.
+        self.intent_slots = {}
         # self.intents, self.entities = {}, {}
         self.fuzz = fuzz
         self.workers = n_workers
@@ -105,6 +110,19 @@ def _get_fuzzed(sample: str) -> List[str]:
             fuzzed.append(" ".join(new_words))
         return fuzzed + [f"* {sample}", f"{sample} *"]
 
+    @staticmethod
+    def _slot_names(patterns: List[str]) -> set:
+        """Return the set of ``{slot}`` names declared across the patterns.
+
+        simplematch markers are ``{name}`` or ``{name:type}``; only the name is
+        kept. Names are lower-cased to line up with the lower-cased match keys.
+        """
+        names = set()
+        for p in patterns:
+            for m in re.finditer(r"\{\s*(\w+)", p):
+                names.add(m.group(1).lower())
+        return names
+
     @staticmethod
     def _literal_words(pattern: str) -> frozenset:
         """Return the set of non-entity, non-wildcard words in a pattern."""
@@ -130,6 +148,7 @@ def add_intent(self, name: str, lines: List[str]):
         # short-circuit before greedy entity patterns consume the query
         regexes.sort(key=lambda r: (0 if "{" not in r and "*" not in r else 1, -len(r)))
         self.intent_samples[name] = regexes
+        self.intent_slots[name] = self._slot_names(regexes)
         for r in regexes:
             cm = simplematch.Matcher(r, case_sensitive=True)
             um = simplematch.Matcher(r, case_sensitive=False)
@@ -153,6 +172,7 @@ def remove_intent(self, name: str):
         """
         if name in self.intent_samples:
             regexes = self.intent_samples.pop(name)
+            self.intent_slots.pop(name, None)
             for rx in regexes:
                 if rx in self._cased_matchers:
                     self._cased_matchers.pop(rx)

padacioso/opm.py

@@ -1,5 +1,6 @@
 """Intent service wrapping padacioso."""
 
+import re
 from functools import lru_cache
 from os.path import isfile
 from typing import Optional, Dict, List, Union
@@ -9,7 +10,8 @@
 from ovos_bus_client.session import SessionManager, Session
 from ovos_config.config import Configuration
 from ovos_plugin_manager.templates.pipeline import ConfidenceMatcherPipeline, IntentHandlerMatch
-from ovos_spec_tools import closest_lang, standardize_lang, SpecMessage, gate_satisfied
+from ovos_spec_tools import (closest_lang, standardize_lang, SpecMessage,
+                             gate_satisfied, context_slot_candidates)
 from ovos_utils import flatten_list
 from ovos_utils.fakebus import FakeBus
 from ovos_utils.log import LOG, log_deprecation
@@ -104,6 +106,11 @@ def __init__(self, bus: Optional[Union[MessageBusClient, FakeBus]] = None,
         # lang and of the enable/disable/detach match state: retained until the
         # intent is deregistered/detached so a re-armed intent keeps its gate.
         self._intent_context_gates = {}
+        # OVOS-INTENT-2 §4.3 — per-slot value blacklists, keyed by internal
+        # intent name -> {slot_name: [blacklisted values]}. A slot bound by the
+        # utterance to a blacklisted value is treated as unresolved so the
+        # OVOS-CONTEXT-1 §7 slot fill can supply it from session context.
+        self._intent_slot_blacklists = {}
         self.max_words = 50  # if an utterance contains more words than this, don't attempt to match
         LOG.debug('Loaded Padacioso intent parser.')
 
@@ -121,6 +128,24 @@ def _store_context_gate(self, name: str, data: Dict):
         else:
             self._intent_context_gates.pop(name, None)
 
+    def _store_slot_blacklist(self, name: str, data: Dict):
+        """OVOS-INTENT-2 §4.3 — retain per-slot value blacklists for an intent.
+
+        The registration payload may carry a ``slot_blacklist`` mapping (or a
+        dict-typed ``blacklist``) keyed by slot name -> list of values that must
+        never bind that slot. Absent/empty declarations clear any prior entry.
+        """
+        blacklist = data.get("slot_blacklist")
+        if blacklist is None and isinstance(data.get("blacklist"), dict):
+            # the intent-level ``blacklist`` (§6.1 suppression phrases) is a
+            # list; a dict here is the per-slot exclusion contract instead.
+            blacklist = data.get("blacklist")
+        if blacklist:
+            self._intent_slot_blacklists[name] = {
+                slot.lower(): list(values) for slot, values in blacklist.items()}
+        else:
+            self._intent_slot_blacklists.pop(name, None)
+
     @staticmethod
     def _internal_name(skill_id: str, intent_name: str) -> str:
         """Compose the engine-internal namespaced intent name.
@@ -201,6 +226,7 @@ def __detach_intent(self, intent_name):
         if intent_name in self.registered_intents:
             self.registered_intents.remove(intent_name)
             self._intent_context_gates.pop(intent_name, None)
+            self._intent_slot_blacklists.pop(intent_name, None)
             for lang in self.containers:
                 self.containers[lang].remove_intent(intent_name)
             # the container was mutated; drop stale cached matches
@@ -278,6 +304,7 @@ def register_intent(self, message):
         if lang in self.containers:
             self.registered_intents.append(message.data['name'])
             self._store_context_gate(message.data['name'], message.data)
+            self._store_slot_blacklist(message.data['name'], message.data)
             try:
                 self._register_object(message, 'intent',
                                       self.containers[lang].add_intent)
@@ -344,6 +371,7 @@ def handle_register_template(self, message: Message):
         self.registered_intents.append(name)
         self._template_samples[(lang, name)] = list(samples)
         self._store_context_gate(name, data)
+        self._store_slot_blacklist(name, data)
         try:
             self.containers[lang].add_intent(name, samples)
         except RuntimeError:
@@ -504,11 +532,11 @@ def calc_intent(self, utterances: List[str], lang: str = None,
                                           blacklisted_intents, blacklisted_skills)
                    for utt in utterances]
         intents = [i for i in intents if i is not None]
+        ctx = sess.intent_context or {}
         # OVOS-CONTEXT-1 §6/§6.1 — drop candidates whose requires_context /
         # excludes_context gate is not satisfied by the live session context.
         # gate_satisfied handles §2 liveness, §3.1 scope and §4 decay.
         if self._intent_context_gates:
-            ctx = sess.intent_context or {}
             gated = []
             for i in intents:
                 gate = self._intent_context_gates.get(i.name)
@@ -524,7 +552,53 @@ def calc_intent(self, utterances: List[str], lang: str = None,
             intents = gated
         # select best
         if intents:
-            return max(intents, key=lambda k: k.conf)
+            best = max(intents, key=lambda k: k.conf)
+            self._apply_slot_context(best, ctx, lang)
+            return best
+
+    def _apply_slot_context(self, intent: PadaciosoIntent, ctx: Dict, lang: str):
+        """OVOS-CONTEXT-1 §7 uniform slot fill + OVOS-INTENT-2 §4.3 blacklist.
+
+        For the matched intent, drop any slot the utterance bound to a
+        blacklisted value (§4.3, whole-word-sequence), then, for every declared
+        template slot still unresolved, offer a live non-null session context
+        entry as its value. Private ``<skill_id>:name`` entries take precedence
+        over shared bare ``name`` (handled by ``context_slot_candidates``). A
+        value the utterance itself produced always wins and is never overwritten.
+        This is independent of ``requires_context``; the gate above governs only
+        the flag-style requires/excludes declarations.
+        """
+        slot_names = self.containers[lang].intent_slots.get(intent.name)
+        if not slot_names:
+            return
+        # §4.3 — un-bind slots the utterance filled with a blacklisted value
+        for slot, values in self._intent_slot_blacklists.get(intent.name, {}).items():
+            bound = intent.matches.get(slot)
+            if bound is not None and self._value_blacklisted(str(bound), values):
+                intent.matches.pop(slot, None)
+        # §7 — fill unresolved declared slots from live context
+        unresolved = [s for s in slot_names if not intent.matches.get(s)]
+        if not unresolved:
+            return
+        owner_id = intent.name.split(":")[0]
+        for slot, value in context_slot_candidates(
+                ctx or {}, unresolved, owner_id=owner_id).items():
+            if not intent.matches.get(slot):  # utterance value always wins
+                intent.matches[slot] = value
+
+    @staticmethod
+    def _value_blacklisted(value: str, blacklist: List[str]) -> bool:
+        """Whole-word-sequence membership of a bound value in a blacklist."""
+        v = value.lower()
+        v_words = v.split()
+        for bad in blacklist:
+            bad = str(bad).lower()
+            if ' ' not in bad:
+                if bad in v_words:
+                    return True
+            elif re.search(r'\b' + re.escape(bad) + r'\b', v):
+                return True
+        return False
 
     def _get_closest_lang(self, lang: str) -> Optional[str]:
         if self.containers:

pyproject.toml

@@ -12,7 +12,7 @@ authors = [{ name = "jarbasai", email = "jarbasai@mailfence.com" }]
 requires-python = ">=3.8"
 dependencies = [
     "simplematch",
-    "ovos-spec-tools>=1.4.0a1",
+    "ovos-spec-tools>=1.4.0a2",
     "ovos-utils>=0.3.5,<1.0.0",
 ]
 

test/test_pipeline.py

@@ -315,5 +315,93 @@ def test_gate_dropped_on_deregister(self):
         self.assertNotIn("tv.skill:off", svc._intent_context_gates)
 
 
+class ContextSlotFillTest(unittest.TestCase):
+    """OVOS-CONTEXT-1 §7 uniform slot fill + OVOS-INTENT-2 §4.3 blacklist."""
+
+    def get_service(self):
+        from ovos_spec_tools import SpecMessage
+        self.SpecMessage = SpecMessage
+        return PadaciosoPipeline(FakeBus(), {"fuzz": False})
+
+    def _msg_with_context(self, intent_context):
+        from ovos_bus_client.session import Session
+        sess = Session("test-session")
+        sess.intent_context = intent_context
+        return Message("recognizer_loop:utterance", {},
+                       {"session": sess.serialize()})
+
+    def _register(self, svc, **data):
+        data.setdefault("lang", "en-US")
+        svc.handle_register_template(Message(
+            self.SpecMessage.INTENT_REGISTER_TEMPLATE.value, data))
+
+    def test_slot_filled_from_context_without_requires(self):
+        # (a) a declared slot fills from live context with NO requires_context.
+        # The utterance matches the slot-free variant, so {location} — declared
+        # by the sibling sample — is left unresolved and taken from context.
+        svc = self.get_service()
+        self._register(svc, skill_id="weather.skill", intent_name="forecast",
+                       samples=["whats the weather",
+                                "whats the weather in {location}"])
+        self.assertNotIn("weather.skill:forecast", svc._intent_context_gates)
+        msg = self._msg_with_context(
+            {"weather.skill:location": {"value": "Lisbon"}})
+        intent = svc.calc_intent("whats the weather", "en-US", msg)
+        self.assertEqual(intent.name, "weather.skill:forecast")
+        self.assertEqual(intent.matches.get("location"), "Lisbon")
+
+    def test_utterance_value_wins_over_context(self):
+        # (b) a value the utterance produces is never overwritten by context
+        svc = self.get_service()
+        self._register(svc, skill_id="weather.skill", intent_name="how_tall",
+                       samples=["how tall is {person}"])
+        msg = self._msg_with_context(
+            {"weather.skill:person": {"value": "Michael Jordan"}})
+        intent = svc.calc_intent("how tall is shaquille oneal", "en-US", msg)
+        self.assertEqual(intent.matches.get("person"), "shaquille oneal")
+
+    def test_blacklisted_value_becomes_context(self):
+        # (c) a blacklisted bound value ("he") -> unresolved -> context fills
+        svc = self.get_service()
+        self._register(svc, skill_id="weather.skill", intent_name="how_tall",
+                       samples=["how tall is {person}"],
+                       slot_blacklist={"person": ["he", "she", "it"]})
+        self.assertIn("weather.skill:how_tall", svc._intent_slot_blacklists)
+        msg = self._msg_with_context(
+            {"weather.skill:person": {"value": "Michael Jordan"}})
+        intent = svc.calc_intent("how tall is he", "en-US", msg)
+        self.assertEqual(intent.matches.get("person"), "Michael Jordan")
+
+    def test_flag_key_still_only_gates(self):
+        # (d) a requires_context flag key gates the match but is not a slot fill;
+        # it never lands in the match data
+        svc = self.get_service()
+        self._register(svc, skill_id="tv.skill", intent_name="turn_off",
+                       samples=["turn off the tv"],
+                       requires_context=["tv_on"])
+        msg = self._msg_with_context({"tv.skill:tv_on": {"value": True}})
+        intent = svc.calc_intent("turn off the tv", "en-US", msg)
+        self.assertEqual(intent.name, "tv.skill:turn_off")
+        self.assertNotIn("tv_on", intent.matches)
+
+    def test_shared_scope_and_private_precedence(self):
+        svc = self.get_service()
+        self._register(svc, skill_id="weather.skill", intent_name="forecast",
+                       samples=["whats the weather",
+                                "whats the weather in {location}"])
+        # shared bare key used when no private entry exists
+        msg = self._msg_with_context({"location": {"value": "shared city"}})
+        self.assertEqual(
+            svc.calc_intent("whats the weather", "en-US", msg
+                            ).matches.get("location"), "shared city")
+        # private entry wins over shared
+        msg = self._msg_with_context(
+            {"location": {"value": "shared city"},
+             "weather.skill:location": {"value": "priv city"}})
+        self.assertEqual(
+            svc.calc_intent("whats the weather", "en-US", msg
+                            ).matches.get("location"), "priv city")
+
+
 if __name__ == "__main__":
     unittest.main()