From 57db82c4f73cd28ea62ef535bef54f6bc6743553 Mon Sep 17 00:00:00 2001 From: JarbasAi Date: Sat, 27 Jun 2026 18:17:39 +0100 Subject: [PATCH 1/3] feat: OVOS-CONTEXT-1 orchestrator-resident intent context MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement the core-resident half of OVOS-CONTEXT-1: the flat, decaying session.intent_context key/value store, alongside the legacy frame-based IntentContextManager. New module ovos_core.intent_services.intent_context provides: - §2 entry shape + liveness predicate (value/flag/null, turns/wallclock) - §3.1 scope resolution (private : vs shared bare key) - §6/§6.1 gate_satisfied predicate (requires/excludes, post-decay) - §7 context_supplied_slots fill rule (utterance value wins) - IntentContextStore: §4 prune-then-decrement decay, §4.1 mid-dispatch exemption, §5.3 ovos.session.sync entry-by-entry merge (set+null-delete), §2 max-entry cap eviction Wire into IntentService (the orchestrator): - handle_session_sync merges ovos.session.sync intent_context payloads - handle_utterance adopts inbound snapshot, prunes pre-match, decrements post-match over the pre-match key set (so mid-dispatch syncs survive) - _emit_match_message applies §5.1 promotion + §7 slot fill and stamps the working map onto the emitted session (legacy Session drops the field) - intent_context exposed as a lazily-backed property for safe partial construction Engine-side §6/§6.1 gating *inside* matchers (adapt/padacioso) is out of scope and deferred; core exposes the shared gating vocabulary they consult. SESSION_SYNC is a literal here; its SpecMessage registration is a SESSION-2/spec-tools follow-up. Co-Authored-By: Claude Opus 4.8 --- docs/intent-service.md | 36 +- ovos_core/intent_services/intent_context.py | 378 ++++++++++++++++++ ovos_core/intent_services/service.py | 144 ++++++- test/unittests/test_intent_context.py | 349 ++++++++++++++++ .../unittests/test_intent_service_extended.py | 3 + 5 files changed, 904 insertions(+), 6 deletions(-) create mode 100644 ovos_core/intent_services/intent_context.py create mode 100644 test/unittests/test_intent_context.py diff --git a/docs/intent-service.md b/docs/intent-service.md index c745a20460f..857dc49853b 100644 --- a/docs/intent-service.md +++ b/docs/intent-service.md @@ -65,9 +65,38 @@ intent.service.intent.get {utterance: "...", lang: "..."} | Event | Effect | |---|---| -| `add_context` | Inject entity into session context | -| `remove_context` | Remove named context entity | -| `clear_context` | Clear all context entities | +| `add_context` | Inject entity into legacy frame-based session context | +| `remove_context` | Remove named context entity (legacy frames) | +| `clear_context` | Clear all context entities (legacy frames) | +| `ovos.session.sync` | OVOS-CONTEXT-1 §5.3 — merge `session.intent_context` entry-by-entry | + +### OVOS-CONTEXT-1 intent context + +The orchestrator implements the **OVOS-CONTEXT-1** flat, decaying +`session.intent_context` key/value store alongside the legacy +frame-based `IntentContextManager`. The core-resident subsystem +(`ovos_core.intent_services.intent_context`) owns: + +- the entry shape and *liveness* predicate (§2); +- the prune-then-decrement decay lifecycle, once per utterance dispatch + (§4 / §4.1); +- the `ovos.session.sync` entry-by-entry merge — present entry objects + set/replace, `null` entries delete, absent keys unchanged (§5.3); +- the §3.1 scope-resolution helper, the §6 / §6.1 gating predicates, and + the §7 context-supplied slot fill, as pure functions any in-process + engine can apply. + +The orchestrator holds the authoritative map keyed by `session_id`, +prunes/decrements it each turn, and stamps it onto every serialized +session it emits, since the legacy `Session` object does not yet carry +`intent_context` as a first-class field. + +> **Engine-side gating is out of scope here.** The §6 / §6.1 +> `requires_context` / `excludes_context` enforcement *inside a matcher* +> (e.g. Adapt dropping a candidate whose required context is unsatisfied) +> lives in each pipeline plugin, not in core. Core provides the shared +> `gate_satisfied` / `context_supplied_slots` vocabulary those plugins +> consult. ## Open Data / Metrics Upload @@ -81,6 +110,7 @@ If `open_data.intent_urls` is configured, intent match results (utterance, inten | `add_context` | `handle_add_context` | | `remove_context` | `handle_remove_context` | | `clear_context` | `handle_clear_context` | +| `ovos.session.sync` | `handle_session_sync` | | `intent.service.intent.get` | `handle_get_intent` | | `intent.service.skills.deactivate` | `_handle_deactivate` | | `intent.service.pipelines.reload` | `handle_reload_pipelines` | diff --git a/ovos_core/intent_services/intent_context.py b/ovos_core/intent_services/intent_context.py new file mode 100644 index 00000000000..dc03f722d83 --- /dev/null +++ b/ovos_core/intent_services/intent_context.py @@ -0,0 +1,378 @@ +# Copyright 2024 OpenVoiceOS +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Core-resident implementation of OVOS-CONTEXT-1 — intent context. + +OVOS-CONTEXT-1 replaces the legacy frame-based ``IntentContextManager`` +(``ovos_bus_client.session.IntentContextManager``) with a **flat, +decaying key/value map** stored at ``session.intent_context`` (§2). + +This module is the orchestrator-resident half of the spec. It owns: + +- the entry shape and the *liveness* predicate (§2); +- the prune-then-decrement decay lifecycle (§4); +- the ``ovos.session.sync`` entry-by-entry merge (§5.3); +- the §3.1 scope-resolution helper that maps a gating declaration to a + stored key, the §6 / §6.1 gating predicates, and the §7 + context-supplied slot fill — provided here as pure functions so any + in-process engine (and core's own match post-processing) can apply + them identically. + +The *engine-side* enforcement of §6 / §6.1 gating inside a matcher (e.g. +the Adapt matcher dropping a candidate whose ``requires_context`` is +unsatisfied) is **out of scope for this module** — it belongs to each +pipeline plugin and is tracked as a follow-up. What lives here is the +shared, engine-agnostic vocabulary those plugins (and the orchestrator) +consult. + +Storage note: ``session.intent_context`` is a plain JSON object carried +inside the OVOS-SESSION-1 carrier (``Message.context.session``). The +legacy ``ovos_bus_client.session.Session`` object does not yet expose it +as a first-class attribute and drops it on round-trip, so the +orchestrator maintains the working map keyed by ``session_id`` and +stamps it back onto the serialized session it emits. See +``IntentContextStore`` and the wiring in ``service.py``. +""" +import time +from typing import Any, Dict, List, Optional, Union + +from ovos_utils.log import LOG + +#: OVOS-CONTEXT-1 §2 — the JSON path, inside the session carrier, that +#: holds the flat intent-context map. Stamped onto / read from the +#: serialized session dict (``message.context["session"][_FIELD]``). +INTENT_CONTEXT_FIELD = "intent_context" + +#: OVOS-CONTEXT-1 §2 / OVOS-MSG-1 §2.1.1 — the single load-bearing +#: separator between a private entry's owner and its sub-key. A prefixed +#: (private) key contains exactly one ``:``; a bare (shared) key none. +SCOPE_SEPARATOR = ":" + +#: OVOS-CONTEXT-1 §2 — the recommended maximum live entry count an +#: orchestrator SHOULD enforce, evicting the entry closest to natural +#: expiry when exceeded. +DEFAULT_MAX_ENTRIES = 1024 + + +def is_live(entry: Dict[str, Any], now: Optional[float] = None) -> bool: + """OVOS-CONTEXT-1 §2 liveness predicate. + + An entry is **live** iff both of: + + - ``turns_remaining`` is unset, ``null``, or strictly greater than 0; + - ``expires_at`` is unset, ``null``, or strictly greater than the + current Unix time. + + @param entry: a context entry object (``value`` plus optional + ``expires_at`` / ``turns_remaining``). + @param now: current Unix time; defaults to ``time.time()``. + @return: True if the entry is live. + """ + if not isinstance(entry, dict): + return False + now = time.time() if now is None else now + + turns = entry.get("turns_remaining") + if turns is not None and not turns > 0: + return False + + expires = entry.get("expires_at") + if expires is not None and not expires > now: + return False + + return True + + +def resolve_key(key: str, scope: str, owner_id: Optional[str]) -> Optional[str]: + """OVOS-CONTEXT-1 §3.1 — map a gating declaration to a stored key. + + - ``scope == "private"`` resolves to ``:``; shared + entries with the same key do **not** satisfy a private gate. + - ``scope == "shared"`` resolves to the bare ````; private + entries with the same name do **not** satisfy a shared gate. + + @param key: the caller-chosen sub-key (unprefixed). + @param scope: ``"private"`` or ``"shared"``. + @param owner_id: the declaring intent's ``skill_id`` / ``pipeline_id``; + required for private scope. + @return: the stored key, or None if a private lookup has no owner. + """ + if scope == "shared": + return key + # private (the safe default) + if not owner_id: + return None + return f"{owner_id}{SCOPE_SEPARATOR}{key}" + + +def normalize_declaration(entry: Union[str, Dict[str, Any]]) -> Optional[Dict[str, str]]: + """OVOS-CONTEXT-1 §6 — normalize one ``requires_context`` / + ``excludes_context`` entry to ``{key, scope}``. + + A bare string is interpreted as ``{key: , scope: "private"}`` + — the safe default (§6). A mapping may set an explicit ``scope``. + + @param entry: a bare key string or a ``{key, scope}`` mapping. + @return: a normalized ``{key, scope}`` dict, or None if malformed. + """ + if isinstance(entry, str): + return {"key": entry, "scope": "private"} + if isinstance(entry, dict) and entry.get("key"): + scope = entry.get("scope", "private") + if scope not in ("private", "shared"): + LOG.warning(f"invalid context scope '{scope}', defaulting to private") + scope = "private" + return {"key": entry["key"], "scope": scope} + LOG.warning(f"malformed context declaration: {entry!r}") + return None + + +def gate_satisfied(intent_context: Dict[str, Any], + requires: Optional[List[Union[str, Dict]]], + excludes: Optional[List[Union[str, Dict]]], + owner_id: Optional[str], + now: Optional[float] = None) -> bool: + """OVOS-CONTEXT-1 §6 / §6.1 — evaluate the positive and negative + gating contracts against a (post-decay, §4) context snapshot. + + A match is permitted iff **every** ``requires_context`` key resolves + to a live entry **and** **no** ``excludes_context`` key resolves to a + live entry, each resolved per §3.1. + + @param intent_context: the flat ``session.intent_context`` map. + @param requires: ``requires_context`` declarations, or None/empty. + @param excludes: ``excludes_context`` declarations, or None/empty. + @param owner_id: the declaring intent's ``skill_id`` / ``pipeline_id``. + @param now: current Unix time; defaults to ``time.time()``. + @return: True if the gate permits the match. + """ + intent_context = intent_context or {} + now = time.time() if now is None else now + + for decl in (requires or []): + norm = normalize_declaration(decl) + if norm is None: + return False # malformed declaration can never be satisfied + stored = resolve_key(norm["key"], norm["scope"], owner_id) + entry = intent_context.get(stored) if stored else None + if entry is None or not is_live(entry, now): + return False + + for decl in (excludes or []): + norm = normalize_declaration(decl) + if norm is None: + continue + stored = resolve_key(norm["key"], norm["scope"], owner_id) + entry = intent_context.get(stored) if stored else None + if entry is not None and is_live(entry, now): + return False + + return True + + +def context_supplied_slots(intent_context: Dict[str, Any], + requires: Optional[List[Union[str, Dict]]], + slot_names: List[str], + owner_id: Optional[str], + filled_slots: Optional[Dict[str, Any]] = None, + now: Optional[float] = None) -> Dict[str, Any]: + """OVOS-CONTEXT-1 §7 — the context-supplied slot rule. + + When a ``requires_context`` key ``k`` **also names a slot** of the + intent definition, and the §3.1-selected entry's ``value`` is + non-null, and the utterance did **not** itself fill slot ``k``, + populate ``Match.slots[k]`` from that value (keyed by ``k``, + unprefixed). Utterance-produced values always win — context is a + fallback, not an override. + + @param intent_context: the flat ``session.intent_context`` map. + @param requires: the intent's ``requires_context`` declarations. + @param slot_names: the slot / vocabulary names of the intent + definition. + @param owner_id: the declaring intent's ``skill_id`` / ``pipeline_id``. + @param filled_slots: slots the utterance itself produced (these win). + @param now: current Unix time; defaults to ``time.time()``. + @return: a mapping of slot-name -> context-supplied value (only the + slots this rule fills; empty if none apply). + """ + intent_context = intent_context or {} + filled_slots = filled_slots or {} + slot_names = set(slot_names or []) + now = time.time() if now is None else now + supplied: Dict[str, Any] = {} + + for decl in (requires or []): + norm = normalize_declaration(decl) + if norm is None: + continue + key = norm["key"] + if key not in slot_names: + continue # gated only, §7 does not apply + if filled_slots.get(key) not in (None, ""): + continue # utterance-produced value wins + stored = resolve_key(key, norm["scope"], owner_id) + entry = intent_context.get(stored) if stored else None + if entry is None or not is_live(entry, now): + continue + value = entry.get("value") + if value is None: + continue # flag-context has no value to supply + supplied[key] = value + + return supplied + + +class IntentContextStore: + """Orchestrator-resident store of ``session.intent_context`` maps. + + Holds the authoritative flat context map for each session, applies + the §4 decay lifecycle (prune before a match round, decrement after), + and the §5.3 ``ovos.session.sync`` entry-by-entry merge. + + The legacy ``ovos_bus_client.session.Session`` object does not carry + ``intent_context`` and drops it on (de)serialization, so the + orchestrator is the source of truth: it reads the inbound snapshot + off the wire, reconciles it with what it already holds, and stamps + the working map back onto the serialized session it emits. + """ + + def __init__(self, max_entries: int = DEFAULT_MAX_ENTRIES): + # session_id -> {key: entry} + self._maps: Dict[str, Dict[str, Any]] = {} + self.max_entries = max_entries + + def get(self, session_id: str) -> Dict[str, Any]: + """Return the (mutable) working map for a session, creating it + empty if absent.""" + return self._maps.setdefault(session_id, {}) + + def set(self, session_id: str, intent_context: Dict[str, Any]) -> None: + """Replace the working map for a session wholesale. + + Used when adopting an inbound snapshot for an unseen session + (the carrier is replayable, OVOS-CONTEXT-1 §9). + """ + self._maps[session_id] = dict(intent_context or {}) + + def adopt_inbound(self, session_id: str, + inbound: Optional[Dict[str, Any]]) -> Dict[str, Any]: + """Reconcile a wire-carried ``intent_context`` snapshot for a + session the orchestrator has not seen before. + + On ordinary Messages ``session.intent_context`` is **read-only** + (§8): the orchestrator keeps its own working copy. But sessions + are replayable carriers (§9) — when an utterance arrives for a + session id we hold no map for, the inbound snapshot is the only + state available, so we adopt it. For a session we already track, + the inbound snapshot is ignored (the §5 pathways are the only + writers). + + @return: the working map for the session. + """ + if session_id not in self._maps: + self._maps[session_id] = dict(inbound or {}) + return self._maps[session_id] + + def prune(self, session_id: str, now: Optional[float] = None) -> Dict[str, Any]: + """OVOS-CONTEXT-1 §4 (pre-match) — remove every non-live entry. + + This is the gating snapshot every matcher sees during the + upcoming match round. + + @return: the pruned working map. + """ + now = time.time() if now is None else now + ctx = self.get(session_id) + dead = [k for k, e in ctx.items() if not is_live(e, now)] + for k in dead: + ctx.pop(k, None) + return ctx + + def decrement(self, session_id: str, + only_keys: Optional[set] = None) -> Dict[str, Any]: + """OVOS-CONTEXT-1 §4 (post-match) — decrement ``turns_remaining`` + on every remaining entry that sets it, whether or not any intent + matched. + + Per §4.1, an entry written by an ``ovos.session.sync`` emitted + **mid-dispatch** must not be decremented by the dispatch it was + written in. The orchestrator captures the key set present at the + pre-match prune and passes it as ``only_keys`` so freshly-synced + keys are skipped, landing alive for exactly the next match round. + + @param only_keys: if given, decrement only entries whose key is in + this set (the snapshot present before the match round). + @return: the working map. + """ + ctx = self.get(session_id) + for key, entry in ctx.items(): + if only_keys is not None and key not in only_keys: + continue # §4.1 — mid-dispatch sync entry, not decremented + turns = entry.get("turns_remaining") + if turns is not None: + entry["turns_remaining"] = turns - 1 + return ctx + + def merge_sync(self, session_id: str, + payload: Optional[Dict[str, Any]]) -> Dict[str, Any]: + """OVOS-CONTEXT-1 §5.3 — apply an ``ovos.session.sync`` + ``intent_context`` payload **entry-by-entry**: + + - a key mapping to an entry object **sets or replaces** that key; + - a key mapping to JSON ``null`` **removes** that key; + - keys absent from the payload are left unchanged. + + Concurrent handlers writing disjoint keys therefore do not + overwrite each other. + + @return: the merged working map. + """ + ctx = self.get(session_id) + if not payload: + return ctx + for key, entry in payload.items(): + if entry is None: + ctx.pop(key, None) + elif isinstance(entry, dict): + ctx[key] = entry + else: + LOG.warning(f"ignoring malformed intent_context entry " + f"for key '{key}': {entry!r}") + self._enforce_cap(session_id) + return ctx + + def _enforce_cap(self, session_id: str, now: Optional[float] = None) -> None: + """OVOS-CONTEXT-1 §2 — bound the live entry count, evicting the + entry closest to natural expiry when exceeded (smallest + ``turns_remaining``, then earliest ``expires_at``, then + arbitrary).""" + ctx = self.get(session_id) + if len(ctx) <= self.max_entries: + return + now = time.time() if now is None else now + + def _expiry_rank(item): + _, entry = item + turns = entry.get("turns_remaining") + expires = entry.get("expires_at") + # entries with neither sort last (least eligible for eviction) + return (turns if turns is not None else float("inf"), + expires if expires is not None else float("inf")) + + while len(ctx) > self.max_entries: + victim = min(ctx.items(), key=_expiry_rank)[0] + ctx.pop(victim, None) + + def discard(self, session_id: str) -> None: + """Drop the working map for a session (e.g. on reset).""" + self._maps.pop(session_id, None) diff --git a/ovos_core/intent_services/service.py b/ovos_core/intent_services/service.py index 30aab6f1c08..95224d69be1 100644 --- a/ovos_core/intent_services/service.py +++ b/ovos_core/intent_services/service.py @@ -35,6 +35,17 @@ from ovos_plugin_manager.pipeline import OVOSPipelineFactory from ovos_plugin_manager.templates.pipeline import IntentHandlerMatch, ConfidenceMatcherPipeline +from ovos_core.intent_services.intent_context import ( + IntentContextStore, + INTENT_CONTEXT_FIELD, +) + +#: OVOS-SESSION-2 §2.7 / OVOS-CONTEXT-1 §5.3 — the bus topic skills emit +#: to add, update, or remove ``session.intent_context`` entries. Defined +#: as a literal here (its registration as a ``SpecMessage`` member is an +#: OVOS-SESSION-2 / ovos-spec-tools follow-up, not a CONTEXT-1 concern). +SESSION_SYNC = "ovos.session.sync" + # Module-level constants for pipeline matcher migration and optimization _PIPELINE_MIGRATION_MAP = { @@ -127,10 +138,20 @@ def __init__(self, bus, config=None, preload_pipelines=True, self.bus.on(SpecMessage.UTTERANCE, self.handle_utterance) + # OVOS-CONTEXT-1 — orchestrator-resident store of the flat + # decaying ``session.intent_context`` map (§2). The orchestrator + # owns this state: it prunes/decrements it each turn (§4), merges + # ``ovos.session.sync`` payloads into it (§5.3), and stamps it + # onto the serialized session it emits. Lazily-backed via the + # ``intent_context`` property so partial constructions stay safe. + self._intent_context = IntentContextStore() + # Context related handlers self.bus.on('add_context', self.handle_add_context) self.bus.on('remove_context', self.handle_remove_context) self.bus.on('clear_context', self.handle_clear_context) + # OVOS-CONTEXT-1 §5.3 — entry-by-entry intent_context merge + self.bus.on(SESSION_SYNC, self.handle_session_sync) # Intents API self.bus.on('intent.service.intent.get', self.handle_get_intent) @@ -144,6 +165,20 @@ def __init__(self, bus, config=None, preload_pipelines=True, if preload_pipelines: self.bus.emit(Message('intent.service.pipelines.reload')) + @property + def intent_context(self) -> IntentContextStore: + """OVOS-CONTEXT-1 — the orchestrator's intent_context store, + lazily created so partial constructions (e.g. ``__new__`` in + tests, or subclasses skipping ``__init__``) remain safe.""" + store = getattr(self, "_intent_context", None) + if store is None: + store = self._intent_context = IntentContextStore() + return store + + @intent_context.setter + def intent_context(self, value: IntentContextStore) -> None: + self._intent_context = value + def handle_reload_pipelines(self, message: Message): pipeline_plugins = OVOSPipelineFactory.get_installed_pipeline_ids() LOG.debug(f"Installed pipeline plugins: {pipeline_plugins}") @@ -341,8 +376,23 @@ def _emit_match_message(self, match: IntentHandlerMatch, message: Message, lang: # emit event for skills callback -> self.handle_activate self.bus.emit(reply.forward(f"{match.skill_id}.activate")) - # update Session if modified by pipeline - reply.context["session"] = sess.serialize() + # OVOS-CONTEXT-1 §5.1 — a pipeline plugin MAY promote captures + # to context by returning an updated intent_context map on its + # Match. We apply it before the dispatch is emitted so it is + # live as a gate for the very next utterance. + promoted = getattr(match, "intent_context", None) + if promoted: + self.intent_context.merge_sync(sess.session_id, promoted) + + # OVOS-CONTEXT-1 §7 — context-supplied slot fill. When the + # matched intent declares a requires_context key that also + # names a slot it left unfilled, populate the slot from the + # live entry's value (utterance-produced values always win). + self._apply_context_slots(match, sess, reply) + + # update Session if modified by pipeline; stamp the working + # intent_context so the post-decay snapshot rides forward + reply.context["session"] = self._stamp_intent_context(sess) # finally emit reply message self.bus.emit(reply) @@ -456,7 +506,18 @@ def handle_utterance(self, message: Message): # get session sess = self._validate_session(message, lang) - message.context["session"] = sess.serialize() + + # OVOS-CONTEXT-1 §4 (pre-match) — adopt any wire-carried snapshot + # for a session we don't yet track (replayable carriers, §9), + # then prune every dead entry so every matcher in this round sees + # the same post-decay gating snapshot. + inbound_ctx = (message.context.get("session") or {}).get(INTENT_CONTEXT_FIELD) + self.intent_context.adopt_inbound(sess.session_id, inbound_ctx) + pre_match_ctx = self.intent_context.prune(sess.session_id) + # snapshot of keys present before the match round; entries synced + # in mid-dispatch (§4.1) are excluded from the post-match decrement + pre_match_keys = set(pre_match_ctx.keys()) + message.context["session"] = self._stamp_intent_context(sess) # match match = None @@ -497,6 +558,14 @@ def handle_utterance(self, message: Message): LOG.debug(f"intent matching took: {stopwatch.time}") + # OVOS-CONTEXT-1 §4 (post-match) — decrement turns_remaining on + # every surviving entry, whether or not any intent matched. This + # runs over the entries present at match time; entries written by + # an ``ovos.session.sync`` emitted mid-dispatch (§4.1) are merged + # into the store by handle_session_sync and are *not* decremented + # here, landing alive for exactly the next match round. + self.intent_context.decrement(sess.session_id, only_keys=pre_match_keys) + # sync any changes made to the default session, eg by ConverseService if sess.session_id == "default": SessionManager.sync(message) @@ -516,6 +585,74 @@ def send_complete_intent_failure(self, message): self.bus.emit(message.reply('complete_intent_failure', message.data)) self.bus.emit(message.reply("ovos.utterance.handled")) + def _apply_context_slots(self, match, sess, reply) -> None: + """OVOS-CONTEXT-1 §7 — apply the context-supplied slot rule to a + match before its dispatch is emitted. + + The rule needs the matched intent's ``requires_context`` list and + its slot / vocabulary names. An engine that implements §7 fills + these slots itself; this orchestrator-resident pass is the + fallback for matches that surface the declaration on the Match + (``requires_context`` + ``slot_names`` attributes) but leave the + fill to core. It is a no-op for matches that expose neither, so + it never disturbs engines that already conform. + + @param match: the IntentHandlerMatch being dispatched. + @param sess: the session whose intent_context is consulted. + @param reply: the dispatch Message whose ``data`` slots are filled. + """ + requires = getattr(match, "requires_context", None) + slot_names = getattr(match, "slot_names", None) + if not requires or not slot_names: + return + from ovos_core.intent_services.intent_context import context_supplied_slots + ctx = self.intent_context.get(sess.session_id) + supplied = context_supplied_slots( + intent_context=ctx, + requires=requires, + slot_names=slot_names, + owner_id=match.skill_id, + filled_slots=reply.data, + ) + for key, value in supplied.items(): + reply.data[key] = value + if supplied: + LOG.debug(f"context-supplied slots (§7): {supplied}") + + def _stamp_intent_context(self, sess) -> dict: + """Serialize a session and stamp the orchestrator's working + ``intent_context`` map onto it (OVOS-CONTEXT-1 §2 / §8). + + The legacy ``Session`` object does not carry ``intent_context``, + so the orchestrator — the field's authority — writes its working + copy onto every serialized session it emits, ensuring the + post-decay snapshot rides forward on the wire to each matcher. + + @param sess: the bus-client Session being emitted. + @return: the serialized session dict including ``intent_context``. + """ + data = sess.serialize() + ctx = self.intent_context.get(sess.session_id) + if ctx: + data[INTENT_CONTEXT_FIELD] = ctx + return data + + def handle_session_sync(self, message: Message): + """OVOS-CONTEXT-1 §5.3 — apply an ``ovos.session.sync`` payload's + ``intent_context`` entry-by-entry into the working map. + + Present entry objects set or replace the keyed entry; ``null`` + entries delete the key; absent keys are unchanged. This is the + only mutation pathway available to skill handlers (§5.3 / §8). + """ + sess = SessionManager.get(message) + payload = (message.data.get("session") or {}).get(INTENT_CONTEXT_FIELD) + if payload is None: + return + self.intent_context.merge_sync(sess.session_id, payload) + LOG.debug(f"merged intent_context sync for session " + f"'{sess.session_id}': {list(payload.keys())}") + @staticmethod def handle_add_context(message: Message): """Add context @@ -626,6 +763,7 @@ def shutdown(self): self.bus.remove('add_context', self.handle_add_context) self.bus.remove('remove_context', self.handle_remove_context) self.bus.remove('clear_context', self.handle_clear_context) + self.bus.remove(SESSION_SYNC, self.handle_session_sync) self.bus.remove('intent.service.intent.get', self.handle_get_intent) self.status.set_stopping() diff --git a/test/unittests/test_intent_context.py b/test/unittests/test_intent_context.py new file mode 100644 index 00000000000..e16b8d7f0dd --- /dev/null +++ b/test/unittests/test_intent_context.py @@ -0,0 +1,349 @@ +# Copyright 2024 OpenVoiceOS +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""OVOS-CONTEXT-1 conformance tests for the core-resident subsystem. + +Covers the orchestrator MUST clauses this implementation flips: + +- §2 the flat entry shape + liveness predicate; +- §4 prune-then-decrement decay across turns; +- §5.3 ``ovos.session.sync`` entry-by-entry merge (set + null-delete); +- §3.1 scope resolution, §6 / §6.1 gating predicates; +- §7 context-supplied slot fill. + +Plus a live FakeBus integration check exercising the orchestrator wiring +in ``IntentService``. +""" +import time +import unittest +from collections import defaultdict +from unittest.mock import MagicMock, patch + +from ovos_bus_client.message import Message +from ovos_bus_client.session import Session +from ovos_utils.fakebus import FakeBus + +from ovos_core.intent_services.intent_context import ( + IntentContextStore, + is_live, + resolve_key, + normalize_declaration, + gate_satisfied, + context_supplied_slots, + INTENT_CONTEXT_FIELD, +) +from ovos_core.intent_services.service import IntentService, SESSION_SYNC + + +def _make_service(config=None) -> IntentService: + """Construct IntentService without loading real pipelines/plugins.""" + bus = FakeBus() + svc = IntentService.__new__(IntentService) + svc.bus = bus + svc.config = config or {} + svc.pipeline_plugins = {} + svc._deactivations = defaultdict(list) + svc.intent_context = IntentContextStore() + + ut = MagicMock() + ut.transform.side_effect = lambda utt, ctx: (utt, ctx) + svc.utterance_plugins = ut + mt = MagicMock() + mt.transform.side_effect = lambda ctx: ctx + svc.metadata_plugins = mt + it = MagicMock() + it.transform.side_effect = lambda intent: intent + svc.intent_plugins = it + svc.status = MagicMock() + return svc + + +# --------------------------------------------------------------------------- +# §2 — entry shape & liveness predicate +# --------------------------------------------------------------------------- + +class TestLiveness(unittest.TestCase): + def test_entry_with_neither_timer_is_live(self): + self.assertTrue(is_live({"value": "Bob"})) + + def test_turns_zero_is_dead(self): + # §4: turns_remaining 0 is dead on arrival + self.assertFalse(is_live({"value": None, "turns_remaining": 0})) + + def test_turns_positive_is_live(self): + self.assertTrue(is_live({"value": None, "turns_remaining": 1})) + + def test_turns_negative_is_dead(self): + self.assertFalse(is_live({"value": None, "turns_remaining": -1})) + + def test_null_turns_is_live(self): + self.assertTrue(is_live({"value": "x", "turns_remaining": None})) + + def test_expired_wallclock_is_dead(self): + self.assertFalse(is_live({"value": "x", "expires_at": time.time() - 1})) + + def test_future_wallclock_is_live(self): + self.assertTrue(is_live({"value": "x", "expires_at": time.time() + 60})) + + def test_both_must_hold(self): + # live turns but expired wallclock -> dead + self.assertFalse(is_live({"value": "x", "turns_remaining": 5, + "expires_at": time.time() - 1})) + + +# --------------------------------------------------------------------------- +# §3 / §3.1 — scope resolution +# --------------------------------------------------------------------------- + +class TestScopeResolution(unittest.TestCase): + def test_private_resolves_to_prefixed_key(self): + self.assertEqual(resolve_key("confirming_milk", "private", "tea.skill"), + "tea.skill:confirming_milk") + + def test_shared_resolves_to_bare_key(self): + self.assertEqual(resolve_key("person", "shared", "bio.skill"), "person") + + def test_private_without_owner_is_none(self): + self.assertIsNone(resolve_key("k", "private", None)) + + def test_bare_string_defaults_to_private(self): + self.assertEqual(normalize_declaration("person"), + {"key": "person", "scope": "private"}) + + def test_long_form_shared(self): + self.assertEqual( + normalize_declaration({"key": "active_room", "scope": "shared"}), + {"key": "active_room", "scope": "shared"}) + + +# --------------------------------------------------------------------------- +# §6 / §6.1 — gating predicates +# --------------------------------------------------------------------------- + +class TestGating(unittest.TestCase): + def test_private_gate_satisfied(self): + ctx = {"tea.skill:confirming_milk": {"value": None, "turns_remaining": 1}} + self.assertTrue(gate_satisfied(ctx, ["confirming_milk"], None, "tea.skill")) + + def test_private_gate_not_satisfied_by_shared(self): + # §3.1: a shared entry of the same name does not satisfy a private gate + ctx = {"confirming_milk": {"value": None, "turns_remaining": 1}} + self.assertFalse(gate_satisfied(ctx, ["confirming_milk"], None, "tea.skill")) + + def test_shared_gate_satisfied(self): + ctx = {"person": {"value": "Bob", "turns_remaining": 3}} + self.assertTrue(gate_satisfied( + ctx, [{"key": "person", "scope": "shared"}], None, "bio.skill")) + + def test_shared_gate_not_satisfied_by_other_skills_private(self): + # §3.2 step 3: people.skill's private entry invisible to bio.skill + ctx = {"people.skill:person": {"value": "Bob", "turns_remaining": 3}} + self.assertFalse(gate_satisfied( + ctx, [{"key": "person", "scope": "shared"}], None, "bio.skill")) + + def test_dead_entry_does_not_satisfy(self): + ctx = {"tea.skill:flag": {"value": None, "turns_remaining": 0}} + self.assertFalse(gate_satisfied(ctx, ["flag"], None, "tea.skill")) + + def test_excludes_blocks_when_live(self): + ctx = {"greet.skill:said_hello": {"value": None}} + self.assertFalse(gate_satisfied(ctx, None, ["said_hello"], "greet.skill")) + + def test_excludes_permits_when_absent(self): + self.assertTrue(gate_satisfied({}, None, ["said_hello"], "greet.skill")) + + def test_both_lists_apply(self): + ctx = {"s.skill:need": {"value": None}} + self.assertTrue(gate_satisfied(ctx, ["need"], ["forbid"], "s.skill")) + ctx["s.skill:forbid"] = {"value": None} + self.assertFalse(gate_satisfied(ctx, ["need"], ["forbid"], "s.skill")) + + +# --------------------------------------------------------------------------- +# §7 — context-supplied slot fill +# --------------------------------------------------------------------------- + +class TestSlotFill(unittest.TestCase): + def test_shared_value_fills_unfilled_slot(self): + ctx = {"person": {"value": "Bob", "turns_remaining": 3}} + supplied = context_supplied_slots( + ctx, [{"key": "person", "scope": "shared"}], + slot_names=["person"], owner_id="bio.skill", filled_slots={}) + self.assertEqual(supplied, {"person": "Bob"}) + + def test_utterance_value_wins(self): + ctx = {"person": {"value": "Bob"}} + supplied = context_supplied_slots( + ctx, [{"key": "person", "scope": "shared"}], + slot_names=["person"], owner_id="bio.skill", + filled_slots={"person": "Alice"}) + self.assertEqual(supplied, {}) + + def test_flag_context_supplies_nothing(self): + ctx = {"bio.skill:person": {"value": None}} + supplied = context_supplied_slots( + ctx, ["person"], slot_names=["person"], + owner_id="bio.skill", filled_slots={}) + self.assertEqual(supplied, {}) + + def test_gated_only_key_not_a_slot(self): + ctx = {"bio.skill:mode": {"value": "x"}} + supplied = context_supplied_slots( + ctx, ["mode"], slot_names=["person"], + owner_id="bio.skill", filled_slots={}) + self.assertEqual(supplied, {}) + + +# --------------------------------------------------------------------------- +# §4 — decay lifecycle on the store +# --------------------------------------------------------------------------- + +class TestDecay(unittest.TestCase): + def test_prune_removes_dead(self): + store = IntentContextStore() + store.set("s", {"live": {"value": "a", "turns_remaining": 1}, + "dead": {"value": "b", "turns_remaining": 0}}) + store.prune("s") + self.assertIn("live", store.get("s")) + self.assertNotIn("dead", store.get("s")) + + def test_decrement_counts_down(self): + store = IntentContextStore() + store.set("s", {"k": {"value": None, "turns_remaining": 2}}) + store.decrement("s") + self.assertEqual(store.get("s")["k"]["turns_remaining"], 1) + + def test_turns_one_lives_exactly_next_round(self): + # §4: turns_remaining 1 is live for the next match round, gone after + store = IntentContextStore() + store.set("s", {"k": {"value": None, "turns_remaining": 1}}) + # round 1: prune keeps it (live), then decrement -> 0 + store.prune("s") + self.assertIn("k", store.get("s")) + store.decrement("s") + # round 2: prune removes it (turns 0 is dead) + store.prune("s") + self.assertNotIn("k", store.get("s")) + + def test_decrement_decrements_unmatched_turn(self): + # §4: decrement runs whether or not an intent matched + store = IntentContextStore() + store.set("s", {"k": {"value": None, "turns_remaining": 1}}) + store.decrement("s") + self.assertEqual(store.get("s")["k"]["turns_remaining"], 0) + + def test_decrement_only_keys_skips_midispatch(self): + # §4.1: an entry synced mid-dispatch is not decremented this turn + store = IntentContextStore() + store.set("s", {"old": {"value": None, "turns_remaining": 1}}) + pre = set(store.get("s").keys()) + # mid-dispatch sync writes a fresh entry + store.merge_sync("s", {"new": {"value": None, "turns_remaining": 1}}) + store.decrement("s", only_keys=pre) + self.assertEqual(store.get("s")["old"]["turns_remaining"], 0) + self.assertEqual(store.get("s")["new"]["turns_remaining"], 1) + + +# --------------------------------------------------------------------------- +# §5.3 — ovos.session.sync entry-by-entry merge +# --------------------------------------------------------------------------- + +class TestSyncMerge(unittest.TestCase): + def test_set_and_replace(self): + store = IntentContextStore() + store.merge_sync("s", {"k": {"value": "a"}}) + self.assertEqual(store.get("s")["k"]["value"], "a") + store.merge_sync("s", {"k": {"value": "b"}}) + self.assertEqual(store.get("s")["k"]["value"], "b") + + def test_null_deletes(self): + store = IntentContextStore() + store.merge_sync("s", {"k": {"value": "a"}, "j": {"value": "b"}}) + store.merge_sync("s", {"k": None}) + self.assertNotIn("k", store.get("s")) + self.assertIn("j", store.get("s")) # disjoint key untouched + + def test_disjoint_keys_do_not_overwrite(self): + store = IntentContextStore() + store.merge_sync("s", {"a.skill:x": {"value": "1"}}) + store.merge_sync("s", {"b.skill:y": {"value": "2"}}) + self.assertEqual(set(store.get("s").keys()), + {"a.skill:x", "b.skill:y"}) + + def test_cap_eviction(self): + store = IntentContextStore(max_entries=2) + store.merge_sync("s", {"near": {"value": "x", "turns_remaining": 1}, + "far": {"value": "y", "turns_remaining": 99}, + "perm": {"value": "z"}}) + ctx = store.get("s") + self.assertEqual(len(ctx), 2) + # the entry closest to expiry (smallest turns_remaining) is evicted + self.assertNotIn("near", ctx) + + +# --------------------------------------------------------------------------- +# live FakeBus integration through IntentService +# --------------------------------------------------------------------------- + +class TestSessionSyncHandler(unittest.TestCase): + def test_sync_handler_merges_set_and_delete(self): + svc = _make_service() + sess = Session("live-sess") + # seed an existing entry directly + svc.intent_context.set(sess.session_id, {"keep": {"value": "k"}}) + + payload = {INTENT_CONTEXT_FIELD: { + "tea.skill:confirming_milk": {"value": None, "turns_remaining": 1}, + "keep": None, # delete + }} + msg = Message(SESSION_SYNC, data={"session": payload}, + context={"session": sess.serialize()}) + with patch("ovos_core.intent_services.service.SessionManager.get", + return_value=sess): + svc.handle_session_sync(msg) + + ctx = svc.intent_context.get(sess.session_id) + self.assertIn("tea.skill:confirming_milk", ctx) + self.assertNotIn("keep", ctx) + + def test_decay_over_turns_via_handle_utterance(self): + svc = _make_service() + sess = Session("turn-sess") + svc.intent_context.set(sess.session_id, + {"tea.skill:flag": {"value": None, + "turns_remaining": 1}}) + + def _drive(): + msg = Message("recognizer_loop:utterance", + data={"utterances": ["hello"]}, + context={"session": sess.serialize()}) + with patch.object(svc, "get_pipeline", return_value=[]), \ + patch("ovos_core.intent_services.service.SessionManager.get", + return_value=sess), \ + patch.object(svc, "_validate_session", return_value=sess), \ + patch("ovos_core.intent_services.service.SessionManager.sync"): + svc.handle_utterance(msg) + + # turn 1: flag is live during the round, decremented to 0 after + _drive() + self.assertEqual( + svc.intent_context.get(sess.session_id)["tea.skill:flag"]["turns_remaining"], + 0) + # turn 2: pre-match prune removes the now-dead flag + _drive() + self.assertNotIn("tea.skill:flag", + svc.intent_context.get(sess.session_id)) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/unittests/test_intent_service_extended.py b/test/unittests/test_intent_service_extended.py index 4f424ad7d65..226cb7dc794 100644 --- a/test/unittests/test_intent_service_extended.py +++ b/test/unittests/test_intent_service_extended.py @@ -36,6 +36,9 @@ def _make_service(config=None) -> IntentService: svc.config = config or {} svc.pipeline_plugins = {} svc._deactivations = defaultdict(list) + # OVOS-CONTEXT-1 — orchestrator-resident intent_context store + from ovos_core.intent_services.intent_context import IntentContextStore + svc.intent_context = IntentContextStore() # Minimal stub objects for transformer services ut = MagicMock() From fcd1d73299f3f0bc67608190edd59164b88041d8 Mon Sep 17 00:00:00 2001 From: JarbasAi Date: Sat, 27 Jun 2026 19:33:51 +0100 Subject: [PATCH 2/3] =?UTF-8?q?refactor!:=20delegate=20ovos.session.sync?= =?UTF-8?q?=20merge=20to=20SessionManager=20(OVOS-CONTEXT-1=20=C2=A75.3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The session.intent_context map and the §5.3 ovos.session.sync entry-by-entry merge (set + null-delete) are now owned by the SessionManager singleton (bus-client #239): it carries intent_context as a first-class round-tripping Session field and applies the merge in SessionManager.handle_session_sync / merge_intent_context. The orchestrator no longer subscribes to ovos.session.sync and holds no parallel session_id-keyed store: - service.py: drop the SESSION_SYNC subscription, the handle_session_sync handler, the bus.remove on shutdown, the _intent_context store, the intent_context property/setter, and _stamp_intent_context. The §4 decay now operates on the session's own intent_context map (prune-then- decrement around the match round), written back via SessionManager.update so the singleton stays authoritative; §4.1 mid-dispatch sync keys are skipped from the decrement. §5.1 promotion merges via SessionManager.merge_intent_context; §7 slot fill reads sess.intent_context. - intent_context.py: IntentContextStore (the {session_id: map} store + its merge_sync) is gone. The decay/liveness/scope/gate/fill logic is kept as stateless module helpers (prune, decrement, enforce_cap + the existing pure predicates) that operate on a passed-in intent_context dict. - pyproject: floor-pin ovos_bus_client>=2.5.0a1 (the alpha carrying #239). - tests: §5.3 merge tests re-homed to bus-client #239; core keeps/expands the decay/liveness/scope/gate/fill tests plus a live check driving the REAL SessionManager merge and asserting core sees merged+decayed context. Co-Authored-By: Claude Opus 4.8 --- ovos_core/intent_services/intent_context.py | 268 +++++++----------- ovos_core/intent_services/service.py | 132 +++------ pyproject.toml | 5 +- test/unittests/test_intent_context.py | 253 ++++++++++------- .../unittests/test_intent_service_extended.py | 6 +- 5 files changed, 309 insertions(+), 355 deletions(-) diff --git a/ovos_core/intent_services/intent_context.py b/ovos_core/intent_services/intent_context.py index dc03f722d83..f58069df23a 100644 --- a/ovos_core/intent_services/intent_context.py +++ b/ovos_core/intent_services/intent_context.py @@ -11,22 +11,32 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -"""Core-resident implementation of OVOS-CONTEXT-1 — intent context. +"""Core-resident helpers for OVOS-CONTEXT-1 — intent context. OVOS-CONTEXT-1 replaces the legacy frame-based ``IntentContextManager`` (``ovos_bus_client.session.IntentContextManager``) with a **flat, decaying key/value map** stored at ``session.intent_context`` (§2). -This module is the orchestrator-resident half of the spec. It owns: - -- the entry shape and the *liveness* predicate (§2); -- the prune-then-decrement decay lifecycle (§4); -- the ``ovos.session.sync`` entry-by-entry merge (§5.3); +The authoritative owner of that map is the ``SessionManager`` singleton +(``ovos_bus_client.session.SessionManager``): it carries +``intent_context`` as a first-class, round-tripping field on every +``Session`` and applies the §5.3 ``ovos.session.sync`` entry-by-entry +merge (set + null-delete) itself — see ``SessionManager.handle_session_sync`` +/ ``SessionManager.merge_intent_context`` (bus-client #239). The +orchestrator does **not** subscribe to ``ovos.session.sync`` and does +**not** hold a parallel store. + +This module is therefore a set of **stateless helpers** the orchestrator +(and any in-process engine) applies to a session's ``intent_context`` +map. It owns no state; every function takes the map (or an entry) as an +argument and returns a value or mutates the passed-in dict in place: + +- the entry-shape *liveness* predicate (§2); +- the prune-then-decrement decay lifecycle (§4 / §4.1); +- the §2 live-entry cap eviction; - the §3.1 scope-resolution helper that maps a gating declaration to a stored key, the §6 / §6.1 gating predicates, and the §7 - context-supplied slot fill — provided here as pure functions so any - in-process engine (and core's own match post-processing) can apply - them identically. + context-supplied slot fill. The *engine-side* enforcement of §6 / §6.1 gating inside a matcher (e.g. the Adapt matcher dropping a candidate whose ``requires_context`` is @@ -34,23 +44,15 @@ pipeline plugin and is tracked as a follow-up. What lives here is the shared, engine-agnostic vocabulary those plugins (and the orchestrator) consult. - -Storage note: ``session.intent_context`` is a plain JSON object carried -inside the OVOS-SESSION-1 carrier (``Message.context.session``). The -legacy ``ovos_bus_client.session.Session`` object does not yet expose it -as a first-class attribute and drops it on round-trip, so the -orchestrator maintains the working map keyed by ``session_id`` and -stamps it back onto the serialized session it emits. See -``IntentContextStore`` and the wiring in ``service.py``. """ import time -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, List, Optional, Set, Union from ovos_utils.log import LOG -#: OVOS-CONTEXT-1 §2 — the JSON path, inside the session carrier, that -#: holds the flat intent-context map. Stamped onto / read from the -#: serialized session dict (``message.context["session"][_FIELD]``). +#: OVOS-CONTEXT-1 §2 — the JSON field, inside the session carrier, that +#: holds the flat intent-context map. First-class on ``Session`` in +#: bus-client (round-trips through serialize/deserialize). INTENT_CONTEXT_FIELD = "intent_context" #: OVOS-CONTEXT-1 §2 / OVOS-MSG-1 §2.1.1 — the single load-bearing @@ -232,147 +234,89 @@ def context_supplied_slots(intent_context: Dict[str, Any], return supplied -class IntentContextStore: - """Orchestrator-resident store of ``session.intent_context`` maps. +# --------------------------------------------------------------------------- +# §4 — decay lifecycle (stateless; operates on a passed-in intent_context map) +# --------------------------------------------------------------------------- - Holds the authoritative flat context map for each session, applies - the §4 decay lifecycle (prune before a match round, decrement after), - and the §5.3 ``ovos.session.sync`` entry-by-entry merge. +def prune(intent_context: Dict[str, Any], + now: Optional[float] = None) -> Dict[str, Any]: + """OVOS-CONTEXT-1 §4 (pre-match) — remove every non-live entry from + the given map, in place. - The legacy ``ovos_bus_client.session.Session`` object does not carry - ``intent_context`` and drops it on (de)serialization, so the - orchestrator is the source of truth: it reads the inbound snapshot - off the wire, reconciles it with what it already holds, and stamps - the working map back onto the serialized session it emits. - """ + This is the gating snapshot every matcher sees during the upcoming + match round. - def __init__(self, max_entries: int = DEFAULT_MAX_ENTRIES): - # session_id -> {key: entry} - self._maps: Dict[str, Dict[str, Any]] = {} - self.max_entries = max_entries - - def get(self, session_id: str) -> Dict[str, Any]: - """Return the (mutable) working map for a session, creating it - empty if absent.""" - return self._maps.setdefault(session_id, {}) - - def set(self, session_id: str, intent_context: Dict[str, Any]) -> None: - """Replace the working map for a session wholesale. - - Used when adopting an inbound snapshot for an unseen session - (the carrier is replayable, OVOS-CONTEXT-1 §9). - """ - self._maps[session_id] = dict(intent_context or {}) - - def adopt_inbound(self, session_id: str, - inbound: Optional[Dict[str, Any]]) -> Dict[str, Any]: - """Reconcile a wire-carried ``intent_context`` snapshot for a - session the orchestrator has not seen before. - - On ordinary Messages ``session.intent_context`` is **read-only** - (§8): the orchestrator keeps its own working copy. But sessions - are replayable carriers (§9) — when an utterance arrives for a - session id we hold no map for, the inbound snapshot is the only - state available, so we adopt it. For a session we already track, - the inbound snapshot is ignored (the §5 pathways are the only - writers). - - @return: the working map for the session. - """ - if session_id not in self._maps: - self._maps[session_id] = dict(inbound or {}) - return self._maps[session_id] - - def prune(self, session_id: str, now: Optional[float] = None) -> Dict[str, Any]: - """OVOS-CONTEXT-1 §4 (pre-match) — remove every non-live entry. - - This is the gating snapshot every matcher sees during the - upcoming match round. - - @return: the pruned working map. - """ - now = time.time() if now is None else now - ctx = self.get(session_id) - dead = [k for k, e in ctx.items() if not is_live(e, now)] - for k in dead: - ctx.pop(k, None) - return ctx - - def decrement(self, session_id: str, - only_keys: Optional[set] = None) -> Dict[str, Any]: - """OVOS-CONTEXT-1 §4 (post-match) — decrement ``turns_remaining`` - on every remaining entry that sets it, whether or not any intent - matched. - - Per §4.1, an entry written by an ``ovos.session.sync`` emitted - **mid-dispatch** must not be decremented by the dispatch it was - written in. The orchestrator captures the key set present at the - pre-match prune and passes it as ``only_keys`` so freshly-synced - keys are skipped, landing alive for exactly the next match round. - - @param only_keys: if given, decrement only entries whose key is in - this set (the snapshot present before the match round). - @return: the working map. - """ - ctx = self.get(session_id) - for key, entry in ctx.items(): - if only_keys is not None and key not in only_keys: - continue # §4.1 — mid-dispatch sync entry, not decremented - turns = entry.get("turns_remaining") - if turns is not None: - entry["turns_remaining"] = turns - 1 - return ctx - - def merge_sync(self, session_id: str, - payload: Optional[Dict[str, Any]]) -> Dict[str, Any]: - """OVOS-CONTEXT-1 §5.3 — apply an ``ovos.session.sync`` - ``intent_context`` payload **entry-by-entry**: - - - a key mapping to an entry object **sets or replaces** that key; - - a key mapping to JSON ``null`` **removes** that key; - - keys absent from the payload are left unchanged. - - Concurrent handlers writing disjoint keys therefore do not - overwrite each other. - - @return: the merged working map. - """ - ctx = self.get(session_id) - if not payload: - return ctx - for key, entry in payload.items(): - if entry is None: - ctx.pop(key, None) - elif isinstance(entry, dict): - ctx[key] = entry - else: - LOG.warning(f"ignoring malformed intent_context entry " - f"for key '{key}': {entry!r}") - self._enforce_cap(session_id) - return ctx - - def _enforce_cap(self, session_id: str, now: Optional[float] = None) -> None: - """OVOS-CONTEXT-1 §2 — bound the live entry count, evicting the - entry closest to natural expiry when exceeded (smallest - ``turns_remaining``, then earliest ``expires_at``, then - arbitrary).""" - ctx = self.get(session_id) - if len(ctx) <= self.max_entries: - return - now = time.time() if now is None else now - - def _expiry_rank(item): - _, entry = item - turns = entry.get("turns_remaining") - expires = entry.get("expires_at") - # entries with neither sort last (least eligible for eviction) - return (turns if turns is not None else float("inf"), - expires if expires is not None else float("inf")) - - while len(ctx) > self.max_entries: - victim = min(ctx.items(), key=_expiry_rank)[0] - ctx.pop(victim, None) - - def discard(self, session_id: str) -> None: - """Drop the working map for a session (e.g. on reset).""" - self._maps.pop(session_id, None) + @param intent_context: the session's flat ``intent_context`` map + (mutated in place). + @param now: current Unix time; defaults to ``time.time()``. + @return: the same (pruned) map. + """ + if not intent_context: + return intent_context + now = time.time() if now is None else now + dead = [k for k, e in intent_context.items() if not is_live(e, now)] + for k in dead: + intent_context.pop(k, None) + return intent_context + + +def decrement(intent_context: Dict[str, Any], + only_keys: Optional[Set[str]] = None) -> Dict[str, Any]: + """OVOS-CONTEXT-1 §4 (post-match) — decrement ``turns_remaining`` on + every remaining entry that sets it, whether or not any intent + matched. + + Per §4.1, an entry written by an ``ovos.session.sync`` emitted + **mid-dispatch** must not be decremented by the dispatch it was + written in. The orchestrator captures the key set present at the + pre-match prune and passes it as ``only_keys`` so freshly-synced keys + are skipped, landing alive for exactly the next match round. + + @param intent_context: the session's flat ``intent_context`` map + (mutated in place). + @param only_keys: if given, decrement only entries whose key is in + this set (the snapshot present before the match round). + @return: the same map. + """ + if not intent_context: + return intent_context + for key, entry in intent_context.items(): + if only_keys is not None and key not in only_keys: + continue # §4.1 — mid-dispatch sync entry, not decremented + if not isinstance(entry, dict): + continue + turns = entry.get("turns_remaining") + if turns is not None: + entry["turns_remaining"] = turns - 1 + return intent_context + + +def enforce_cap(intent_context: Dict[str, Any], + max_entries: int = DEFAULT_MAX_ENTRIES, + now: Optional[float] = None) -> Dict[str, Any]: + """OVOS-CONTEXT-1 §2 — bound the live entry count of the given map, + evicting the entry closest to natural expiry when exceeded (smallest + ``turns_remaining``, then earliest ``expires_at``, then arbitrary). + + @param intent_context: the session's flat ``intent_context`` map + (mutated in place). + @param max_entries: the recommended live-entry ceiling. + @param now: current Unix time; defaults to ``time.time()`` (unused for + ranking but accepted for call-site symmetry). + @return: the same map. + """ + if not intent_context or len(intent_context) <= max_entries: + return intent_context + + def _expiry_rank(item): + _, entry = item + turns = entry.get("turns_remaining") if isinstance(entry, dict) else None + expires = entry.get("expires_at") if isinstance(entry, dict) else None + # entries with neither sort last (least eligible for eviction) + return (turns if turns is not None else float("inf"), + expires if expires is not None else float("inf")) + + while len(intent_context) > max_entries: + victim = min(intent_context.items(), key=_expiry_rank)[0] + intent_context.pop(victim, None) + return intent_context diff --git a/ovos_core/intent_services/service.py b/ovos_core/intent_services/service.py index 95224d69be1..846620a7f3a 100644 --- a/ovos_core/intent_services/service.py +++ b/ovos_core/intent_services/service.py @@ -36,16 +36,12 @@ from ovos_plugin_manager.templates.pipeline import IntentHandlerMatch, ConfidenceMatcherPipeline from ovos_core.intent_services.intent_context import ( - IntentContextStore, - INTENT_CONTEXT_FIELD, + context_supplied_slots, + prune as prune_intent_context, + decrement as decrement_intent_context, + enforce_cap as enforce_intent_context_cap, ) -#: OVOS-SESSION-2 §2.7 / OVOS-CONTEXT-1 §5.3 — the bus topic skills emit -#: to add, update, or remove ``session.intent_context`` entries. Defined -#: as a literal here (its registration as a ``SpecMessage`` member is an -#: OVOS-SESSION-2 / ovos-spec-tools follow-up, not a CONTEXT-1 concern). -SESSION_SYNC = "ovos.session.sync" - # Module-level constants for pipeline matcher migration and optimization _PIPELINE_MIGRATION_MAP = { @@ -138,20 +134,19 @@ def __init__(self, bus, config=None, preload_pipelines=True, self.bus.on(SpecMessage.UTTERANCE, self.handle_utterance) - # OVOS-CONTEXT-1 — orchestrator-resident store of the flat - # decaying ``session.intent_context`` map (§2). The orchestrator - # owns this state: it prunes/decrements it each turn (§4), merges - # ``ovos.session.sync`` payloads into it (§5.3), and stamps it - # onto the serialized session it emits. Lazily-backed via the - # ``intent_context`` property so partial constructions stay safe. - self._intent_context = IntentContextStore() + # OVOS-CONTEXT-1 — the flat, decaying ``session.intent_context`` + # map (§2) is owned by the ``SessionManager`` singleton: it carries + # the field first-class on every Session and applies the §5.3 + # ``ovos.session.sync`` entry-by-entry merge itself (bus-client + # #239, ``SessionManager.handle_session_sync``). The orchestrator + # does NOT subscribe to ``ovos.session.sync`` and holds no parallel + # store — it only applies the §4 decay lifecycle around each match + # round on the session's own map (see ``handle_utterance``). # Context related handlers self.bus.on('add_context', self.handle_add_context) self.bus.on('remove_context', self.handle_remove_context) self.bus.on('clear_context', self.handle_clear_context) - # OVOS-CONTEXT-1 §5.3 — entry-by-entry intent_context merge - self.bus.on(SESSION_SYNC, self.handle_session_sync) # Intents API self.bus.on('intent.service.intent.get', self.handle_get_intent) @@ -165,20 +160,6 @@ def __init__(self, bus, config=None, preload_pipelines=True, if preload_pipelines: self.bus.emit(Message('intent.service.pipelines.reload')) - @property - def intent_context(self) -> IntentContextStore: - """OVOS-CONTEXT-1 — the orchestrator's intent_context store, - lazily created so partial constructions (e.g. ``__new__`` in - tests, or subclasses skipping ``__init__``) remain safe.""" - store = getattr(self, "_intent_context", None) - if store is None: - store = self._intent_context = IntentContextStore() - return store - - @intent_context.setter - def intent_context(self, value: IntentContextStore) -> None: - self._intent_context = value - def handle_reload_pipelines(self, message: Message): pipeline_plugins = OVOSPipelineFactory.get_installed_pipeline_ids() LOG.debug(f"Installed pipeline plugins: {pipeline_plugins}") @@ -378,11 +359,17 @@ def _emit_match_message(self, match: IntentHandlerMatch, message: Message, lang: # OVOS-CONTEXT-1 §5.1 — a pipeline plugin MAY promote captures # to context by returning an updated intent_context map on its - # Match. We apply it before the dispatch is emitted so it is - # live as a gate for the very next utterance. + # Match. We merge it (entry-by-entry, §5.3 semantics) onto the + # session's own map via the SessionManager singleton, before + # the dispatch is emitted, so it is live as a gate for the very + # next utterance. promoted = getattr(match, "intent_context", None) if promoted: - self.intent_context.merge_sync(sess.session_id, promoted) + merged = SessionManager.merge_intent_context( + dict(sess.intent_context or {}), promoted) + enforce_intent_context_cap(merged) + sess.intent_context = merged or None + SessionManager.update(sess) # OVOS-CONTEXT-1 §7 — context-supplied slot fill. When the # matched intent declares a requires_context key that also @@ -390,9 +377,9 @@ def _emit_match_message(self, match: IntentHandlerMatch, message: Message, lang: # live entry's value (utterance-produced values always win). self._apply_context_slots(match, sess, reply) - # update Session if modified by pipeline; stamp the working - # intent_context so the post-decay snapshot rides forward - reply.context["session"] = self._stamp_intent_context(sess) + # update Session if modified by pipeline; the intent_context + # map round-trips on the Session itself (bus-client #239) + reply.context["session"] = sess.serialize() # finally emit reply message self.bus.emit(reply) @@ -507,17 +494,19 @@ def handle_utterance(self, message: Message): # get session sess = self._validate_session(message, lang) - # OVOS-CONTEXT-1 §4 (pre-match) — adopt any wire-carried snapshot - # for a session we don't yet track (replayable carriers, §9), - # then prune every dead entry so every matcher in this round sees - # the same post-decay gating snapshot. - inbound_ctx = (message.context.get("session") or {}).get(INTENT_CONTEXT_FIELD) - self.intent_context.adopt_inbound(sess.session_id, inbound_ctx) - pre_match_ctx = self.intent_context.prune(sess.session_id) + # OVOS-CONTEXT-1 §4 (pre-match) — the session carries its own flat + # ``intent_context`` map (owned by SessionManager, bus-client #239). + # Prune every dead entry so every matcher in this round sees the + # same post-decay gating snapshot, then write the pruned map back + # via the singleton so it stays authoritative. + intent_ctx = dict(sess.intent_context or {}) + prune_intent_context(intent_ctx) # snapshot of keys present before the match round; entries synced # in mid-dispatch (§4.1) are excluded from the post-match decrement - pre_match_keys = set(pre_match_ctx.keys()) - message.context["session"] = self._stamp_intent_context(sess) + pre_match_keys = set(intent_ctx.keys()) + sess.intent_context = intent_ctx or None + SessionManager.update(sess) + message.context["session"] = sess.serialize() # match match = None @@ -562,9 +551,15 @@ def handle_utterance(self, message: Message): # every surviving entry, whether or not any intent matched. This # runs over the entries present at match time; entries written by # an ``ovos.session.sync`` emitted mid-dispatch (§4.1) are merged - # into the store by handle_session_sync and are *not* decremented - # here, landing alive for exactly the next match round. - self.intent_context.decrement(sess.session_id, only_keys=pre_match_keys) + # onto the managed session by ``SessionManager.handle_session_sync`` + # and are *not* decremented here (``only_keys`` skips them), landing + # alive for exactly the next match round. We re-read the authoritative + # session so any such mid-dispatch merge is reflected. + sess = SessionManager.sessions.get(sess.session_id, sess) + post_ctx = dict(sess.intent_context or {}) + decrement_intent_context(post_ctx, only_keys=pre_match_keys) + sess.intent_context = post_ctx or None + SessionManager.update(sess) # sync any changes made to the default session, eg by ConverseService if sess.session_id == "default": @@ -605,10 +600,8 @@ def _apply_context_slots(self, match, sess, reply) -> None: slot_names = getattr(match, "slot_names", None) if not requires or not slot_names: return - from ovos_core.intent_services.intent_context import context_supplied_slots - ctx = self.intent_context.get(sess.session_id) supplied = context_supplied_slots( - intent_context=ctx, + intent_context=sess.intent_context or {}, requires=requires, slot_names=slot_names, owner_id=match.skill_id, @@ -619,40 +612,6 @@ def _apply_context_slots(self, match, sess, reply) -> None: if supplied: LOG.debug(f"context-supplied slots (§7): {supplied}") - def _stamp_intent_context(self, sess) -> dict: - """Serialize a session and stamp the orchestrator's working - ``intent_context`` map onto it (OVOS-CONTEXT-1 §2 / §8). - - The legacy ``Session`` object does not carry ``intent_context``, - so the orchestrator — the field's authority — writes its working - copy onto every serialized session it emits, ensuring the - post-decay snapshot rides forward on the wire to each matcher. - - @param sess: the bus-client Session being emitted. - @return: the serialized session dict including ``intent_context``. - """ - data = sess.serialize() - ctx = self.intent_context.get(sess.session_id) - if ctx: - data[INTENT_CONTEXT_FIELD] = ctx - return data - - def handle_session_sync(self, message: Message): - """OVOS-CONTEXT-1 §5.3 — apply an ``ovos.session.sync`` payload's - ``intent_context`` entry-by-entry into the working map. - - Present entry objects set or replace the keyed entry; ``null`` - entries delete the key; absent keys are unchanged. This is the - only mutation pathway available to skill handlers (§5.3 / §8). - """ - sess = SessionManager.get(message) - payload = (message.data.get("session") or {}).get(INTENT_CONTEXT_FIELD) - if payload is None: - return - self.intent_context.merge_sync(sess.session_id, payload) - LOG.debug(f"merged intent_context sync for session " - f"'{sess.session_id}': {list(payload.keys())}") - @staticmethod def handle_add_context(message: Message): """Add context @@ -763,7 +722,6 @@ def shutdown(self): self.bus.remove('add_context', self.handle_add_context) self.bus.remove('remove_context', self.handle_remove_context) self.bus.remove('clear_context', self.handle_clear_context) - self.bus.remove(SESSION_SYNC, self.handle_session_sync) self.bus.remove('intent.service.intent.get', self.handle_get_intent) self.status.set_stopping() diff --git a/pyproject.toml b/pyproject.toml index 711fde85974..21b6c9a04c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,10 @@ dependencies = [ "watchdog>=2.1, <3.0", "combo-lock>=0.2.2, <0.4", "ovos-utils>=0.11.1a1,<1.0.0", - "ovos_bus_client>=2.2.0a1,<3.0.0", + # OVOS-CONTEXT-1 §5.3: the SessionManager singleton owns the + # session.intent_context first-class field and the ovos.session.sync + # entry-by-entry merge (set + null-delete) — requires bus-client #239. + "ovos_bus_client>=2.5.0a1,<3.0.0", "ovos-plugin-manager>=2.5.0a1,<3.0.0", "ovos-config>=0.0.13,<3.0.0", "ovos-workshop>=8.3.0a1,<10.0.0", diff --git a/test/unittests/test_intent_context.py b/test/unittests/test_intent_context.py index e16b8d7f0dd..646466cc199 100644 --- a/test/unittests/test_intent_context.py +++ b/test/unittests/test_intent_context.py @@ -11,38 +11,43 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -"""OVOS-CONTEXT-1 conformance tests for the core-resident subsystem. +"""OVOS-CONTEXT-1 conformance tests for the core-resident helpers. -Covers the orchestrator MUST clauses this implementation flips: +The §5.3 ``ovos.session.sync`` entry-by-entry merge is owned by the +``SessionManager`` singleton (bus-client #239) and is covered there; core +does **not** re-implement it. This module covers the stateless helpers the +orchestrator applies to a session's ``intent_context`` map: -- §2 the flat entry shape + liveness predicate; -- §4 prune-then-decrement decay across turns; -- §5.3 ``ovos.session.sync`` entry-by-entry merge (set + null-delete); +- §2 the flat entry shape + liveness predicate + cap eviction; +- §4 / §4.1 prune-then-decrement decay across turns; - §3.1 scope resolution, §6 / §6.1 gating predicates; - §7 context-supplied slot fill. -Plus a live FakeBus integration check exercising the orchestrator wiring -in ``IntentService``. +Plus a live FakeBus integration check that drives ``ovos.session.sync`` +through the **real** ``SessionManager`` and asserts the orchestrator sees +the merged-then-decayed context on the session. """ import time import unittest from collections import defaultdict -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock from ovos_bus_client.message import Message -from ovos_bus_client.session import Session +from ovos_bus_client.session import Session, SessionManager from ovos_utils.fakebus import FakeBus from ovos_core.intent_services.intent_context import ( - IntentContextStore, is_live, resolve_key, normalize_declaration, gate_satisfied, context_supplied_slots, + prune, + decrement, + enforce_cap, INTENT_CONTEXT_FIELD, ) -from ovos_core.intent_services.service import IntentService, SESSION_SYNC +from ovos_core.intent_services.service import IntentService def _make_service(config=None) -> IntentService: @@ -53,7 +58,6 @@ def _make_service(config=None) -> IntentService: svc.config = config or {} svc.pipeline_plugins = {} svc._deactivations = defaultdict(list) - svc.intent_context = IntentContextStore() ut = MagicMock() ut.transform.side_effect = lambda utt, ctx: (utt, ctx) @@ -205,144 +209,189 @@ def test_gated_only_key_not_a_slot(self): # --------------------------------------------------------------------------- -# §4 — decay lifecycle on the store +# §4 — decay lifecycle (stateless helpers over a passed-in map) # --------------------------------------------------------------------------- class TestDecay(unittest.TestCase): def test_prune_removes_dead(self): - store = IntentContextStore() - store.set("s", {"live": {"value": "a", "turns_remaining": 1}, - "dead": {"value": "b", "turns_remaining": 0}}) - store.prune("s") - self.assertIn("live", store.get("s")) - self.assertNotIn("dead", store.get("s")) + ctx = {"live": {"value": "a", "turns_remaining": 1}, + "dead": {"value": "b", "turns_remaining": 0}} + prune(ctx) + self.assertIn("live", ctx) + self.assertNotIn("dead", ctx) + + def test_prune_is_in_place_and_returns_map(self): + ctx = {"dead": {"value": "b", "turns_remaining": 0}} + out = prune(ctx) + self.assertIs(out, ctx) + self.assertEqual(ctx, {}) def test_decrement_counts_down(self): - store = IntentContextStore() - store.set("s", {"k": {"value": None, "turns_remaining": 2}}) - store.decrement("s") - self.assertEqual(store.get("s")["k"]["turns_remaining"], 1) + ctx = {"k": {"value": None, "turns_remaining": 2}} + decrement(ctx) + self.assertEqual(ctx["k"]["turns_remaining"], 1) def test_turns_one_lives_exactly_next_round(self): # §4: turns_remaining 1 is live for the next match round, gone after - store = IntentContextStore() - store.set("s", {"k": {"value": None, "turns_remaining": 1}}) + ctx = {"k": {"value": None, "turns_remaining": 1}} # round 1: prune keeps it (live), then decrement -> 0 - store.prune("s") - self.assertIn("k", store.get("s")) - store.decrement("s") + prune(ctx) + self.assertIn("k", ctx) + decrement(ctx) # round 2: prune removes it (turns 0 is dead) - store.prune("s") - self.assertNotIn("k", store.get("s")) + prune(ctx) + self.assertNotIn("k", ctx) def test_decrement_decrements_unmatched_turn(self): # §4: decrement runs whether or not an intent matched - store = IntentContextStore() - store.set("s", {"k": {"value": None, "turns_remaining": 1}}) - store.decrement("s") - self.assertEqual(store.get("s")["k"]["turns_remaining"], 0) + ctx = {"k": {"value": None, "turns_remaining": 1}} + decrement(ctx) + self.assertEqual(ctx["k"]["turns_remaining"], 0) def test_decrement_only_keys_skips_midispatch(self): # §4.1: an entry synced mid-dispatch is not decremented this turn - store = IntentContextStore() - store.set("s", {"old": {"value": None, "turns_remaining": 1}}) - pre = set(store.get("s").keys()) + ctx = {"old": {"value": None, "turns_remaining": 1}} + pre = set(ctx.keys()) # mid-dispatch sync writes a fresh entry - store.merge_sync("s", {"new": {"value": None, "turns_remaining": 1}}) - store.decrement("s", only_keys=pre) - self.assertEqual(store.get("s")["old"]["turns_remaining"], 0) - self.assertEqual(store.get("s")["new"]["turns_remaining"], 1) + ctx["new"] = {"value": None, "turns_remaining": 1} + decrement(ctx, only_keys=pre) + self.assertEqual(ctx["old"]["turns_remaining"], 0) + self.assertEqual(ctx["new"]["turns_remaining"], 1) + + def test_decrement_leaves_untimed_entries(self): + ctx = {"perm": {"value": "x"}} + decrement(ctx) + self.assertNotIn("turns_remaining", ctx["perm"]) # --------------------------------------------------------------------------- -# §5.3 — ovos.session.sync entry-by-entry merge +# §2 — live-entry cap eviction # --------------------------------------------------------------------------- -class TestSyncMerge(unittest.TestCase): - def test_set_and_replace(self): - store = IntentContextStore() - store.merge_sync("s", {"k": {"value": "a"}}) - self.assertEqual(store.get("s")["k"]["value"], "a") - store.merge_sync("s", {"k": {"value": "b"}}) - self.assertEqual(store.get("s")["k"]["value"], "b") - - def test_null_deletes(self): - store = IntentContextStore() - store.merge_sync("s", {"k": {"value": "a"}, "j": {"value": "b"}}) - store.merge_sync("s", {"k": None}) - self.assertNotIn("k", store.get("s")) - self.assertIn("j", store.get("s")) # disjoint key untouched - - def test_disjoint_keys_do_not_overwrite(self): - store = IntentContextStore() - store.merge_sync("s", {"a.skill:x": {"value": "1"}}) - store.merge_sync("s", {"b.skill:y": {"value": "2"}}) - self.assertEqual(set(store.get("s").keys()), - {"a.skill:x", "b.skill:y"}) - - def test_cap_eviction(self): - store = IntentContextStore(max_entries=2) - store.merge_sync("s", {"near": {"value": "x", "turns_remaining": 1}, - "far": {"value": "y", "turns_remaining": 99}, - "perm": {"value": "z"}}) - ctx = store.get("s") +class TestCapEviction(unittest.TestCase): + def test_cap_evicts_entry_closest_to_expiry(self): + ctx = {"near": {"value": "x", "turns_remaining": 1}, + "far": {"value": "y", "turns_remaining": 99}, + "perm": {"value": "z"}} + enforce_cap(ctx, max_entries=2) self.assertEqual(len(ctx), 2) # the entry closest to expiry (smallest turns_remaining) is evicted self.assertNotIn("near", ctx) + def test_cap_noop_under_limit(self): + ctx = {"a": {"value": "1"}, "b": {"value": "2"}} + enforce_cap(ctx, max_entries=10) + self.assertEqual(set(ctx.keys()), {"a", "b"}) + # --------------------------------------------------------------------------- -# live FakeBus integration through IntentService +# live FakeBus integration through the REAL SessionManager # --------------------------------------------------------------------------- -class TestSessionSyncHandler(unittest.TestCase): - def test_sync_handler_merges_set_and_delete(self): - svc = _make_service() +class TestLiveSessionManagerSync(unittest.TestCase): + """Drive ``ovos.session.sync`` through the real SessionManager (which + owns the §5.3 merge, bus-client #239) and assert the orchestrator sees + the merged-then-decayed context on the session.""" + + def setUp(self): + # isolate the singleton between tests + SessionManager.sessions = {"default": Session("default")} + SessionManager.default_session = SessionManager.sessions["default"] + SessionManager.bus = None + + def tearDown(self): + SessionManager.sessions = {"default": Session("default")} + SessionManager.default_session = SessionManager.sessions["default"] + SessionManager.bus = None + + def test_real_sessionmanager_merges_sync(self): + # the merge itself lives in SessionManager (bus-client #239); here + # we drive the *real* handler to confirm core consumes a managed + # session whose intent_context the singleton has merged. The full + # set + null-delete matrix is covered by bus-client's own suite — + # we assert the additive set + delete is visible on the session. sess = Session("live-sess") - # seed an existing entry directly - svc.intent_context.set(sess.session_id, {"keep": {"value": "k"}}) + sess.intent_context = {"keep": {"value": "k"}} + SessionManager.update(sess) - payload = {INTENT_CONTEXT_FIELD: { + # a skill emits ovos.session.sync with an updated session snapshot + snap = sess.serialize() + snap[INTENT_CONTEXT_FIELD] = { "tea.skill:confirming_milk": {"value": None, "turns_remaining": 1}, "keep": None, # delete - }} - msg = Message(SESSION_SYNC, data={"session": payload}, - context={"session": sess.serialize()}) - with patch("ovos_core.intent_services.service.SessionManager.get", - return_value=sess): - svc.handle_session_sync(msg) - - ctx = svc.intent_context.get(sess.session_id) - self.assertIn("tea.skill:confirming_milk", ctx) - self.assertNotIn("keep", ctx) - - def test_decay_over_turns_via_handle_utterance(self): + } + SessionManager.handle_session_sync( + Message("ovos.session.sync", context={"session": snap})) + + merged = SessionManager.sessions["live-sess"].intent_context + self.assertIn("tea.skill:confirming_milk", merged) + self.assertNotIn("keep", merged) + + def test_orchestrator_decays_managed_session(self): + bus = FakeBus() + SessionManager.connect_to_bus(bus) svc = _make_service() + svc.bus = bus + sess = Session("turn-sess") - svc.intent_context.set(sess.session_id, - {"tea.skill:flag": {"value": None, - "turns_remaining": 1}}) + sess.intent_context = {"tea.skill:flag": {"value": None, + "turns_remaining": 1}} + SessionManager.update(sess) def _drive(): msg = Message("recognizer_loop:utterance", data={"utterances": ["hello"]}, - context={"session": sess.serialize()}) - with patch.object(svc, "get_pipeline", return_value=[]), \ - patch("ovos_core.intent_services.service.SessionManager.get", - return_value=sess), \ - patch.object(svc, "_validate_session", return_value=sess), \ - patch("ovos_core.intent_services.service.SessionManager.sync"): - svc.handle_utterance(msg) + context={"session": + SessionManager.sessions["turn-sess"].serialize()}) + # no pipelines loaded -> no match, decay still runs + svc.handle_utterance(msg) # turn 1: flag is live during the round, decremented to 0 after _drive() self.assertEqual( - svc.intent_context.get(sess.session_id)["tea.skill:flag"]["turns_remaining"], + SessionManager.sessions["turn-sess"].intent_context["tea.skill:flag"]["turns_remaining"], 0) # turn 2: pre-match prune removes the now-dead flag _drive() - self.assertNotIn("tea.skill:flag", - svc.intent_context.get(sess.session_id)) + self.assertNotIn( + "tea.skill:flag", + SessionManager.sessions["turn-sess"].intent_context or {}) + + def test_midispatch_sync_survives_decay(self): + # §4.1: an entry merged onto the managed session mid-dispatch (via + # the real SessionManager handler) is NOT decremented by the + # dispatch it arrived in — it lands alive for exactly the next match + # round. Exercises core's post-match decrement (only_keys) against a + # session the real singleton has merged into. + sess = Session("mid-sess") + sess.intent_context = {"old.skill:flag": {"value": None, + "turns_remaining": 1}} + SessionManager.update(sess) + + # pre-match snapshot, as core captures it before the match round + pre_match_keys = set(sess.intent_context.keys()) + + # mid-dispatch: a skill emits ovos.session.sync; the REAL handler + # merges the disjoint new key onto the managed session. + snap = sess.serialize() + snap[INTENT_CONTEXT_FIELD] = { + "new.skill:flag": {"value": None, "turns_remaining": 1}} + SessionManager.handle_session_sync( + Message("ovos.session.sync", context={"session": snap})) + + # core's post-match decrement: re-read authoritative session, skip + # mid-dispatch keys (only_keys) so they are not decremented. + managed = SessionManager.sessions["mid-sess"] + post_ctx = dict(managed.intent_context or {}) + decrement(post_ctx, only_keys=pre_match_keys) + managed.intent_context = post_ctx or None + SessionManager.update(managed) + + ctx = SessionManager.sessions["mid-sess"].intent_context + # the pre-existing entry was decremented (present at match time) + self.assertEqual(ctx["old.skill:flag"]["turns_remaining"], 0) + # the mid-dispatch entry was NOT decremented (arrived after prune) + self.assertEqual(ctx["new.skill:flag"]["turns_remaining"], 1) if __name__ == "__main__": diff --git a/test/unittests/test_intent_service_extended.py b/test/unittests/test_intent_service_extended.py index 226cb7dc794..5f17d80da4f 100644 --- a/test/unittests/test_intent_service_extended.py +++ b/test/unittests/test_intent_service_extended.py @@ -36,9 +36,9 @@ def _make_service(config=None) -> IntentService: svc.config = config or {} svc.pipeline_plugins = {} svc._deactivations = defaultdict(list) - # OVOS-CONTEXT-1 — orchestrator-resident intent_context store - from ovos_core.intent_services.intent_context import IntentContextStore - svc.intent_context = IntentContextStore() + # OVOS-CONTEXT-1 — the session.intent_context map is owned by the + # SessionManager singleton (bus-client #239); the orchestrator holds + # no store, so nothing to seed here. # Minimal stub objects for transformer services ut = MagicMock() From 4b5ebb0c7f8d4cdd49c3645dd4aa839225567187 Mon Sep 17 00:00:00 2001 From: JarbasAi Date: Thu, 2 Jul 2026 20:29:22 +0100 Subject: [PATCH 3/3] =?UTF-8?q?feat:=20OVOS-CONTEXT-1=20=C2=A76/=C2=A76.1?= =?UTF-8?q?=20orchestrator=20gate=20+=20manifest-sourced=20declarations?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The orchestrator drops a matched candidate whose requires_context is unmet or whose excludes_context is present, re-checking the gate so a misbehaving matcher cannot dispatch a context-gated intent. The declared gates and slot names are read from the passive INTENT-4 §10 manifest (the single source of an intent's declaration) — never off the Match. - IntentManifest.get_context_requirements / get_slot_names: union an intent's requires_context / excludes_context / slot names across its registration definitions. - match loop: gate backstop via gate_satisfied against session.intent_context. - _apply_context_slots (§7): sources requires_context + slot_names from the manifest instead of the Match. Additive: an intent that declares no gates is unaffected. Co-Authored-By: Claude Opus 4.8 (1M context) --- ovos_core/intent_services/manifest.py | 53 +++++++++++++++++ ovos_core/intent_services/service.py | 40 ++++++++++--- test/unittests/test_intent_context.py | 79 ++++++++++++++++++++++++++ test/unittests/test_intent_manifest.py | 53 +++++++++++++++++ 4 files changed, 217 insertions(+), 8 deletions(-) diff --git a/ovos_core/intent_services/manifest.py b/ovos_core/intent_services/manifest.py index 6a469ef9349..5601c6b37bf 100644 --- a/ovos_core/intent_services/manifest.py +++ b/ovos_core/intent_services/manifest.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +from typing import Optional + from ovos_bus_client.message import Message from ovos_spec_tools import standardize_lang from ovos_utils.log import LOG @@ -183,3 +185,54 @@ def _on_describe(self, message: Message): self.bus.emit(message.reply("ovos.intent.describe.response", {"ok": False, "error": f"unknown intent {skill_id}:{intent_name}:{lang}"})) + + # ------------------------------------------------------------------ + # orchestrator lookups — the passive index is the single source of an + # intent's declared context gates and slot names (OVOS-CONTEXT-1); + # the orchestrator reads them from here, never off the Match. + # ------------------------------------------------------------------ + + def _matching_definitions(self, session_id: str, skill_id: str, + intent_name: str, lang: Optional[str]) -> list: + lang = standardize_lang(lang) if lang else None + out = [] + for entry in self._effective_pool(session_id): + if entry["skill_id"] != skill_id or entry["intent_name"] != intent_name: + continue + if lang and entry["lang"] != lang: + continue + out.append(entry.get("definition") or {}) + return out + + def get_context_requirements(self, session_id: str, skill_id: str, + intent_name: str, lang: Optional[str] = None): + """OVOS-CONTEXT-1 §6/§6.1 — the ``requires_context`` / ``excludes_context`` + declared for an intent, unioned across its registration definitions. + + @return: a ``(requires, excludes)`` tuple of declaration lists (each a + bare-key string or a ``{key, scope}`` mapping); empty when the intent + declares no gates or is unknown. + """ + requires, excludes = [], [] + for d in self._matching_definitions(session_id, skill_id, intent_name, lang): + for r in (d.get("requires_context") or []): + if r not in requires: + requires.append(r) + for e in (d.get("excludes_context") or []): + if e not in excludes: + excludes.append(e) + return requires, excludes + + def get_slot_names(self, session_id: str, skill_id: str, + intent_name: str, lang: Optional[str] = None) -> list: + """The intent's declared slot / keyword names, unioned across its + registration definitions — the keyword model's ``required``/``optional``/ + ``one_of`` names (OVOS-INTENT-4 §5) and any template ``slots``. Used by + the §7 context-supplied slot rule.""" + names = [] + for d in self._matching_definitions(session_id, skill_id, intent_name, lang): + for field in ("required", "optional", "one_of", "slots"): + for name in (d.get(field) or []): + if name not in names: + names.append(name) + return names diff --git a/ovos_core/intent_services/service.py b/ovos_core/intent_services/service.py index d98dff63a6e..e9a54a71b77 100644 --- a/ovos_core/intent_services/service.py +++ b/ovos_core/intent_services/service.py @@ -38,6 +38,7 @@ from ovos_plugin_manager.templates.pipeline import IntentHandlerMatch, ConfidenceMatcherPipeline from ovos_core.intent_services.intent_context import ( + gate_satisfied, context_supplied_slots, prune as prune_intent_context, decrement as decrement_intent_context, @@ -570,6 +571,24 @@ def handle_utterance(self, message: Message): LOG.debug( f"ignoring match, intent '{match.match_type}' blacklisted by Session '{sess.session_id}'") continue + # OVOS-CONTEXT-1 §6/§6.1 — orchestrator gate backstop. A + # matcher SHOULD drop a candidate whose requires_context + # is unmet or whose excludes_context is present; core + # re-checks it so a misbehaving matcher cannot dispatch a + # context-gated intent. The declared gates are read from + # the passive INTENT-4 §10 manifest (the single source of + # an intent's declaration), never off the Match. Absent + # => ungated => unaffected. + if isinstance(match, IntentHandlerMatch) and match.skill_id: + intent_name = match.match_type.split(":", 1)[-1] + requires, excludes = self.intent_manifest.get_context_requirements( + sess.session_id, match.skill_id, intent_name, intent_lang) + if (requires or excludes) and not gate_satisfied( + sess.intent_context or {}, requires, excludes, + owner_id=match.skill_id): + LOG.debug( + f"ignoring match, context gate unsatisfied for '{match.match_type}'") + continue try: self._dispatch_match(match, message, intent_lang, pipeline_id=pipeline) @@ -638,19 +657,24 @@ def _apply_context_slots(self, match, sess, reply) -> None: match before its dispatch is emitted. The rule needs the matched intent's ``requires_context`` list and - its slot / vocabulary names. An engine that implements §7 fills - these slots itself; this orchestrator-resident pass is the - fallback for matches that surface the declaration on the Match - (``requires_context`` + ``slot_names`` attributes) but leave the - fill to core. It is a no-op for matches that expose neither, so - it never disturbs engines that already conform. + its slot / vocabulary names. Both are read from the passive INTENT-4 + §10 manifest — the single source of an intent's declaration — never + off the Match. An engine that implements §7 fills these slots itself; + this orchestrator-resident pass is the fallback. It is a no-op for an + intent that declares no context-gated slot, so it never disturbs + engines that already conform. @param match: the IntentHandlerMatch being dispatched. @param sess: the session whose intent_context is consulted. @param reply: the dispatch Message whose ``data`` slots are filled. """ - requires = getattr(match, "requires_context", None) - slot_names = getattr(match, "slot_names", None) + if not (isinstance(match, IntentHandlerMatch) and match.skill_id): + return + intent_name = match.match_type.split(":", 1)[-1] + requires, _ = self.intent_manifest.get_context_requirements( + sess.session_id, match.skill_id, intent_name, sess.lang) + slot_names = self.intent_manifest.get_slot_names( + sess.session_id, match.skill_id, intent_name, sess.lang) if not requires or not slot_names: return supplied = context_supplied_slots( diff --git a/test/unittests/test_intent_context.py b/test/unittests/test_intent_context.py index 646466cc199..b10706e9d61 100644 --- a/test/unittests/test_intent_context.py +++ b/test/unittests/test_intent_context.py @@ -396,3 +396,82 @@ def test_midispatch_sync_survives_decay(self): if __name__ == "__main__": unittest.main() + + +# --------------------------------------------------------------------------- +# §6/§6.1 — orchestrator gate backstop in the match loop +# --------------------------------------------------------------------------- + +from unittest.mock import patch # noqa: E402 +from ovos_plugin_manager.templates.pipeline import IntentHandlerMatch # noqa: E402 +from ovos_core.intent_services.manifest import IntentManifest # noqa: E402 + + +class TestOrchestratorGate(unittest.TestCase): + """handle_utterance drops a context-gated match whose gate — declared in + the passive §10 manifest, not on the Match — is unsatisfied, and dispatches + it once the required context is live.""" + + def _gated_service(self, match, requires=("kitchen",)): + svc = _make_service() + svc._handle_transformers = lambda m: m + svc.disambiguate_lang = lambda m: "en-US" + svc.send_complete_intent_failure = MagicMock() + svc._dispatch_match = MagicMock() + svc.get_pipeline = lambda session: [("fake-high", lambda utts, lang, msg: match)] + # declare the intent's requires_context in the passive manifest + svc.intent_manifest = IntentManifest(svc.bus) + if requires: + svc.intent_manifest._on_register(Message( + "ovos.intent.register.keyword", + {"skill_id": "lights.skill", "intent_name": "on", + "lang": "en-US", "requires_context": list(requires)}, {})) + return svc + + def _session(self, intent_context=None): + sess = Session("s1") + sess.lang = "en-US" + sess.pipeline = ["fake-high"] + sess.intent_context = intent_context + SessionManager.update(sess) + return sess + + def _match(self): + return IntentHandlerMatch(match_type="lights:on", + match_data={"conf": 1.0}, + skill_id="lights.skill", + utterance="turn on") + + def _utterance(self, sess): + return Message("ovos.utterance.handle", + {"utterances": ["turn on"], "lang": "en-US"}, + {"session": sess.serialize()}) + + def test_gate_unsatisfied_drops_match(self): + match = self._match() + svc = self._gated_service(match) + sess = self._session(intent_context=None) # no 'kitchen' context + with patch.object(svc, "_validate_session", return_value=sess): + svc.handle_utterance(self._utterance(sess)) + svc._dispatch_match.assert_not_called() + svc.send_complete_intent_failure.assert_called_once() + + def test_gate_satisfied_dispatches(self): + match = self._match() + svc = self._gated_service(match) + # private 'kitchen' under the declaring skill_id, live + ctx = {"lights.skill:kitchen": {"value": "kitchen", "turns_remaining": 2}} + sess = self._session(intent_context=ctx) + with patch.object(svc, "_validate_session", return_value=sess): + svc.handle_utterance(self._utterance(sess)) + svc._dispatch_match.assert_called_once() + svc.send_complete_intent_failure.assert_not_called() + + def test_ungated_match_unaffected(self): + match = IntentHandlerMatch(match_type="lights:on", match_data={"conf": 1.0}, + skill_id="lights.skill", utterance="turn on") + svc = self._gated_service(match, requires=None) # nothing declared + sess = self._session(intent_context=None) + with patch.object(svc, "_validate_session", return_value=sess): + svc.handle_utterance(self._utterance(sess)) + svc._dispatch_match.assert_called_once() diff --git a/test/unittests/test_intent_manifest.py b/test/unittests/test_intent_manifest.py index 97ac19d96e1..71f18674506 100644 --- a/test/unittests/test_intent_manifest.py +++ b/test/unittests/test_intent_manifest.py @@ -203,3 +203,56 @@ def test_describe_unknown_returns_error(self): def test_describe_missing_fields_returns_error(self): resp = self._query(skill_id="skill.a") self.assertFalse(resp["ok"]) + + +def _reg_ctx(skill_id, intent_name, requires=None, excludes=None, slots=None, + lang="en-US", method="keyword", session_id="default"): + data = {"skill_id": skill_id, "intent_name": intent_name, "lang": lang} + if requires is not None: + data["requires_context"] = requires + if excludes is not None: + data["excludes_context"] = excludes + if slots is not None: + data["required"] = slots + return Message(f"ovos.intent.register.{method}", data=data, + context={"session": {"session_id": session_id}, "skill_id": skill_id}) + + +class TestManifestContextLookups(unittest.TestCase): + def setUp(self): + self.m = _manifest() + + def test_context_requirements(self): + self.m._on_register(_reg_ctx("s.skill", "on", requires=["kitchen"], + excludes=["modal"])) + req, exc = self.m.get_context_requirements("default", "s.skill", "on", "en-US") + self.assertEqual(req, ["kitchen"]) + self.assertEqual(exc, ["modal"]) + + def test_context_requirements_empty_when_undeclared(self): + self.m._on_register(_reg_ctx("s.skill", "on")) + self.assertEqual(self.m.get_context_requirements("default", "s.skill", "on", "en-US"), + ([], [])) + + def test_context_requirements_unknown_intent(self): + self.assertEqual(self.m.get_context_requirements("default", "x", "y", "en-US"), + ([], [])) + + def test_context_requirements_union_across_methods(self): + self.m._on_register(_reg_ctx("s.skill", "on", requires=["a"], method="keyword")) + self.m._on_register(_reg_ctx("s.skill", "on", requires=["b"], method="template")) + req, _ = self.m.get_context_requirements("default", "s.skill", "on", "en-US") + self.assertEqual(sorted(req), ["a", "b"]) + + def test_slot_names(self): + self.m._on_register(_reg_ctx("s.skill", "on", slots=["room", "device"])) + self.assertEqual(self.m.get_slot_names("default", "s.skill", "on", "en-US"), + ["room", "device"]) + + def test_session_scoped_visible_via_effective_pool(self): + self.m._on_register(_reg_ctx("s.skill", "on", requires=["k"], session_id="sat-1")) + req, _ = self.m.get_context_requirements("sat-1", "s.skill", "on", "en-US") + self.assertEqual(req, ["k"]) + # a different session does not see the satellite-scoped declaration + self.assertEqual(self.m.get_context_requirements("other", "s.skill", "on", "en-US"), + ([], []))