orchestrata_ultimate.orchestrator.intent_vocabulary

Reference › orchestrata_ultimate › orchestrator › intent_vocabulary

The keyword tables behind the fallback tier of intent_classifier. It is also read by intent_broker_factory and by the gen_intent_lexicon and check_lexicon_collisions scripts. It is traced to spec 003 story US-001 (REQ-003-US-001-040, 041 and 047).

The module-level constants are the seed data: work-type keywords (fix, bug, broken and so on, keyed by WorkType from pipeline_router), generic verbs that map to a work type (add and build to feature, for example), and keyword sets for IntentAction, IntentTarget and IntentPriority. IntentVocabulary is a dataclass that holds one registry per category. default builds the seeded instance; add_keywords and remove_keywords extend or shrink a category at runtime, merge folds another vocabulary into this one, and to_dict and from_dict serialize it so extensions can be stored and reloaded. The classifier scores the tokens of a sentence against these registries when the grammar tier does not decide.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

IntentVocabulary (dataclass)

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-040', 'REQ-003-US-001-041', 'REQ-003-US-001-047']), dataclass · Line 456

Runtime-extensible keyword vocabulary for NL intent classification.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/orchestrator/intent_vocabulary.py

"""Extensible NL intent vocabulary — keyword → category registries.

Spec: 003-intent / US-001 extension (see plans/intent-nl-classifier-design.md).

Data-driven, runtime-mutable keyword vocabulary that powers the
fallback (vocabulary/keyword-scoring) tier of the NL intent
classification cascade in ``intent_classifier.py``. Callers add domain
vocabulary at runtime without any code edits — that extensibility is
the entire point of this module.
"""

from __future__ import annotations

from dataclasses import dataclass, field

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.orchestrator.intent_types import (
    IntentAction,
    IntentPriority,
    IntentTarget,
)
from orchestrata_ultimate.orchestrator.pipeline_router import WorkType

VocabCategory = WorkType | IntentAction | IntentTarget | IntentPriority

# ── Seed keyword sets for IntentVocabulary.default() ────────────────────
#
# DELIBERATELY ABSENT, measured against the held-out corpus
# (tests/orchestrator/data/intent_corpus_heldout.json) — four words that read
# as work-type signals in the abstract but name SUBSYSTEMS in this codebase,
# so each one misclassified real sentences:
#
#   memory       PERFORMANCE  "wire the memory engine into the gateway" is
#                             feature work; memory/ is a package.
#   trace        ANALYSIS     "add tracing to the dispatcher" is feature work;
#                             core/tracing/ is a package.
#   integration  TEST         "the fireflies integration is broken" is fix
#                             work; integrations/ is a package.
#   version      DEPENDENCY   "the protocol version field" is CXP code, not a
#                             dependency bump.
#
# The sentences that motivated each still classify correctly without them
# ("bump the pinned version of msgpack" resolves on `bump`; "we need
# integration tests for the kernel" resolves on `tests`). Do not re-add these
# without a corpus case that needs them and does not regress the four above.

_DEFAULT_WORK_TYPE_KEYWORDS: dict[WorkType, set[str]] = {
    WorkType.FIX: {
        "fix",
        "bug",
        "broken",
        "break",
        "error",
        "crash",
        "regression",
        "defect",
        "fail",
        "failure",
        "exception",
        "traceback",
        "repair",
        "hang",
        "deadlock",
        "leak",
        "corrupt",
        "incorrect",
        "wrong",
        "segfault",
        "flaky",
    },
    WorkType.FEATURE: {
        "add",
        "create",
        "implement",
        "build",
        "feature",
        "new",
        "introduce",
        "support",
        "enable",
        "scaffold",
        "generate",
        "extend",
        # Mined from the conversation history (swarm, 2026-08-08).
        "deploy",
    },
    WorkType.REFACTOR: {
        "refactor",
        "restructure",
        "rename",
        "extract",
        "cleanup",
        "clean",
        "split",
        "simplify",
        "decompose",
        "consolidate",
        "deduplicate",
        "inline",
        "modularize",
        "untangle",
        "tidy",
        "reorganize",
        "collapse",
        # Mined from the conversation history (swarm, 2026-08-08).
        "decouple",
        "retire",
        "migrate",
    },
    WorkType.DOCUMENTATION: {
        "document",
        "docs",
        "readme",
        "docstring",
        "changelog",
        "guide",
        "tutorial",
        "comment",
        "annotate",
        "manual",
    },
    WorkType.TEST: {
        "test",
        "coverage",
        "unit",
        "e2e",
        "pytest",
        "fixture",
        "assertion",
        "tdd",
        # Mined from the conversation history (swarm, 2026-08-08). RUL-009
        # makes observation part of what a test run IS here, so "observe the
        # suite" is test work rather than analysis.
        "observe",
        # `suite` was tried here and REJECTED on measurement. It is a genuine
        # test artifact ("run the full suite", "the suite is green") and it
        # gains +1 on raw turns, but it costs exactly 1 on the labelled corpus
        # because "the suite is slow under load" then ties `suite`(test)
        # against `slow`(performance) and the leftmost wins.
        #
        # That is a PERFORMANCE COMPLAINT ABOUT A TEST ARTIFACT, and the
        # artifact noun should not outrank the property word. A precedence
        # rule ("performance outranks other decisive categories") was measured
        # across all three corpora and scores IDENTICALLY — 110/185 either way
        # — because it fixes "the suite is slow" and breaks "diagnose the slow
        # startup". So there is no free win here; `suite` stays out until the
        # artifact-vs-property distinction is modelled properly.
    },
    WorkType.REVIEW: {
        "review",
        "audit",
        "inspect",
        "critique",
        "lint",
        "feedback",
        "approve",
        "verify",
    },
    WorkType.ANALYSIS: {
        "analyze",
        "analysis",
        "investigate",
        "why",
        "explain",
        "diagnose",
        "understand",
        "compare",
        "examine",
        # Mined from the conversation history (swarm, 2026-08-08).
        "research",
        "triage",
        "calculate",
    },
    WorkType.PERFORMANCE: {
        "optimize",
        "slow",
        "slower",
        "perf",
        "latency",
        "benchmark",
        "throughput",
        "speed",
        "faster",
        "bottleneck",
        "profile",
        "cpu",
        "allocation",
    },
    WorkType.DEPENDENCY: {
        "dependency",
        "upgrade",
        "bump",
        "pin",
        "package",
        "library",
        "vendored",
        "lockfile",
        # Mined from the conversation history (swarm, 2026-08-08).
        "wheel",
    },
    WorkType.SPEC: {
        "spec",
        "specification",
        "requirement",
        "contract",
        "acceptance",
        "criteria",
        "adr",
        "rfc",
    },
}

# ── Generic change verbs — grammar precedence, NOT a fifth registry ─────
#
# A generic verb says only that SOMETHING is being changed; it never names
# the work artifact. "add tests for the parser" is test work, not feature
# work, even though `add` sits at position 0 and `tests` at position 1 --
# English puts the verb first and the OBJECT names the artifact.
#
# The DCG grammar therefore resolves a decisive work-type word ahead of a
# generic verb REGARDLESS of position (REQ-003-US-001-052), and falls back to
# the generic verb's default work type only when no decisive word appears.
#
# Words here that ALSO appear in _DEFAULT_WORK_TYPE_KEYWORDS keep scoring
# exactly as before in the vocabulary/keyword-scoring tier -- this map only
# changes the GRAMMAR tier's precedence. Words here that appear in NO
# work-type set (`update`, `make`, `change`, `modify`, `write`, `set`) are
# grammar-only fallbacks; adding them to the scoring registries would
# create ties in the fallback tier, which is a separate behaviour change.
_DEFAULT_GENERIC_VERBS: dict[str, WorkType] = {
    "add": WorkType.FEATURE,
    "build": WorkType.FEATURE,
    "change": WorkType.FEATURE,
    "create": WorkType.FEATURE,
    "enable": WorkType.FEATURE,
    "extend": WorkType.FEATURE,
    "generate": WorkType.FEATURE,
    "implement": WorkType.FEATURE,
    "introduce": WorkType.FEATURE,
    "make": WorkType.FEATURE,
    "modify": WorkType.FEATURE,
    "new": WorkType.FEATURE,
    "scaffold": WorkType.FEATURE,
    "set": WorkType.FEATURE,
    "support": WorkType.FEATURE,
    "update": WorkType.FEATURE,
    "write": WorkType.FEATURE,
    # Mined from the conversation history (swarm, 2026-08-08).
    "clarify": WorkType.DOCUMENTATION,
    "clone": WorkType.FEATURE,
    "deny": WorkType.FEATURE,
    "draw": WorkType.FEATURE,
}

# REJECTED by the swarm-consolidation review, recorded so they are not
# re-proposed. Each was mined with real evidence and still fails the
# subsystem test that produced the four DELIBERATELY ABSENT words above:
#
#   measure     ANALYSIS    35 hits, but its evidence is "measure run time" —
#                           performance, not analysis, and this project says
#                           "measure" about everything.
#   validate    TEST        evidence is `wiki-kimi validate`, a COMMAND name;
#                           validate_ci.py / validate_env.py are gates.
#   patch       FIX         evidence is a sentence about NOT patching; also
#                           collides with the banned unittest.mock.patch.
#   purge       FIX         evidence names the purge TOOL
#                           (purge_stray_kg_databases.py).
#   unblock     REFACTOR    miscategorised — unblocking is fix work.
#   promote     FEATURE     promote_env.py is the dev->test->prod script.
#   provision   FEATURE     names the deploy/ provisioner.

_DEFAULT_ACTION_KEYWORDS: dict[IntentAction, set[str]] = {
    IntentAction.READ: {"read", "view", "show", "get"},
    IntentAction.WRITE: {"write", "save", "document"},
    IntentAction.DELETE: {"delete", "remove", "drop"},
    IntentAction.EXECUTE: {"execute", "run", "start"},
    IntentAction.QUERY: {"query", "search", "find"},
    IntentAction.CREATE: {"create", "add", "new", "build", "implement"},
    IntentAction.MODIFY: {"modify", "update", "change", "edit", "fix", "refactor"},
    IntentAction.ANALYZE: {"analyze", "investigate"},
    IntentAction.REVIEW: {"review", "audit", "inspect"},
    IntentAction.PLAN: {"plan", "design"},
}

_DEFAULT_TARGET_KEYWORDS: dict[IntentTarget, set[str]] = {
    IntentTarget.FILE: {"file", "toggle", "path", "config", "configuration", "yaml"},
    IntentTarget.CODE: {
        "code",
        "parser",
        "function",
        "class",
        "module",
        "method",
        "variable",
        "import",
        "symbol",
        "type",
        "interface",
        "decorator",
        "helper",
        # Mined from the conversation history (swarm, 2026-08-08) — the
        # code-level concepts this project's own prose is built out of.
        "adapter",
        "closure",
        "consumer",
        "coroutine",
        "dealloc",
        "factory",
        "fallback",
        "generator",
        "hook",
        "lifecycle",
        "loop",
        "mixin",
        "namespace",
        "ontology",
        "protocol",
        "recursion",
        "sentinel",
        "signature",
        "wrapper",
    },
    IntentTarget.DATABASE: {
        "database",
        "db",
        "table",
        "schema",
        "migration",
        "cypher",
        "sql",
        "graph",
        "redis",
        "memgraph",
        # Mined from the conversation history (swarm, 2026-08-08) — the
        # repository-pattern vocabulary.
        "aggregate",
        "entity",
        "index",
        "ledger",
        "store",
        "transaction",
    },
    IntentTarget.API: {
        "api",
        "endpoint",
        "route",
        "handler",
        "http",
        "rest",
        "grpc",
        "websocket",
        "gateway",
        # Mined from the conversation history (swarm, 2026-08-08) — the
        # transport/messaging surface.
        "channel",
        "client",
        "codec",
        # `dispatch` was mined and accepted here on "message dispatch", then
        # rejected by the corpus: `dispatcher` stems to `dispatch`, and
        # "refactor the dispatcher into smaller functions" started resolving
        # to api instead of code. kernel/dispatcher.py and cxp/dispatcher.py
        # are code, not an API surface.
        "packet",
        "payload",
        "stream",
        "transport",
    },
    IntentTarget.SYSTEM: {
        "system",
        "service",
        "daemon",
        "kernel",
        "process",
        "runtime",
        "container",
        "infrastructure",
        "infra",
        # Mined from the conversation history (swarm, 2026-08-08).
        "gate",
        "pool",
        "scheduler",
        "server",
        "thread",
        "worker",
    },
    IntentTarget.DIRECTORY: {"directory", "folder", "dir", "tree", "workspace"},
    IntentTarget.COMMAND: {
        "command",
        "cli",
        "script",
        "shell",
        "terminal",
        "subcommand",
        "flag",
        "argument",
    },
    IntentTarget.GIT: {
        "git",
        "branch",
        "commit",
        "merge",
        "rebase",
        "push",
        "tag",
        "worktree",
        "diff",
        "repo",
        "repository",
    },
    IntentTarget.NETWORK: {
        "network",
        "socket",
        "port",
        "dns",
        "tcp",
        "firewall",
        "proxy",
        "connection",
    },
    IntentTarget.IMPACT: {"impact", "blast", "radius", "downstream", "ripple"},
    IntentTarget.ARCHITECTURE: {
        "architecture",
        "pr",
        "layer",
        "boundary",
        "component",
        "diagram",
        "pattern",
        "topology",
        # Mined from the conversation history (swarm, 2026-08-08) — the
        # structural-quality vocabulary the user named explicitly (layers,
        # patterns, coupling).
        "cohesion",
        "coupling",
        "facade",
        "invariant",
        "seam",
        "tier",
    },
}

_DEFAULT_PRIORITY_KEYWORDS: dict[IntentPriority, set[str]] = {
    IntentPriority.CRITICAL: {"urgent", "critical", "asap", "immediately"},
    IntentPriority.HIGH: {"important", "high", "soon"},
    IntentPriority.LOW: {"low", "minor", "whenever"},
    IntentPriority.BACKGROUND: {"background", "someday", "eventually"},
}


@traces(
    spec="003",
    story="US-001",
    reqs=["REQ-003-US-001-040", "REQ-003-US-001-041", "REQ-003-US-001-047"],
)
@dataclass
class IntentVocabulary:
    """Runtime-extensible keyword vocabulary for NL intent classification.

    All four registries are plain mutable dicts of sets, so callers can
    add domain vocabulary without code edits (design:
    plans/intent-nl-classifier-design.md).
    """

    work_type_keywords: dict[WorkType, set[str]] = field(default_factory=dict)
    action_keywords: dict[IntentAction, set[str]] = field(default_factory=dict)
    target_keywords: dict[IntentTarget, set[str]] = field(default_factory=dict)
    priority_keywords: dict[IntentPriority, set[str]] = field(default_factory=dict)
    generic_verbs: dict[str, WorkType] = field(default_factory=dict)

    @classmethod
    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-041"])
    def default(cls) -> IntentVocabulary:
        """Build the seeded default vocabulary."""
        return cls(
            work_type_keywords={
                k: set(v) for k, v in _DEFAULT_WORK_TYPE_KEYWORDS.items()
            },
            action_keywords={k: set(v) for k, v in _DEFAULT_ACTION_KEYWORDS.items()},
            target_keywords={k: set(v) for k, v in _DEFAULT_TARGET_KEYWORDS.items()},
            priority_keywords={
                k: set(v) for k, v in _DEFAULT_PRIORITY_KEYWORDS.items()
            },
            generic_verbs=dict(_DEFAULT_GENERIC_VERBS),
        )

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-040"])
    def _registry_for(self, category: VocabCategory) -> dict:
        if isinstance(category, WorkType):
            return self.work_type_keywords
        if isinstance(category, IntentAction):
            return self.action_keywords
        if isinstance(category, IntentTarget):
            return self.target_keywords
        if isinstance(category, IntentPriority):
            return self.priority_keywords
        raise TypeError(f"Unknown vocabulary category type: {type(category)!r}")

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-040"])
    def add_keywords(self, category: VocabCategory, words: set[str]) -> None:
        """Add keywords to a category's registry — the extensibility surface."""
        registry = self._registry_for(category)
        existing = registry.setdefault(category, set())
        existing.update(w.lower() for w in words)

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-040"])
    def remove_keywords(self, category: VocabCategory, words: set[str]) -> None:
        """Remove keywords from a category's registry."""
        registry = self._registry_for(category)
        existing = registry.get(category)
        if existing:
            existing.difference_update(w.lower() for w in words)

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-040"])
    def merge(self, other: IntentVocabulary) -> None:
        """Merge another vocabulary's keywords into this one (set union)."""
        for work_type_category, words in other.work_type_keywords.items():
            self.add_keywords(work_type_category, words)
        for action_category, words in other.action_keywords.items():
            self.add_keywords(action_category, words)
        for target_category, words in other.target_keywords.items():
            self.add_keywords(target_category, words)
        for priority_category, words in other.priority_keywords.items():
            self.add_keywords(priority_category, words)
        self.generic_verbs.update(other.generic_verbs)

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-047"])
    def to_dict(self) -> dict:
        """Serialize to a JSON/YAML-round-trippable dict."""
        return {
            "work_type_keywords": {
                k.value: sorted(v) for k, v in self.work_type_keywords.items()
            },
            "action_keywords": {
                k.value: sorted(v) for k, v in self.action_keywords.items()
            },
            "target_keywords": {
                k.value: sorted(v) for k, v in self.target_keywords.items()
            },
            "priority_keywords": {
                k.value: sorted(v) for k, v in self.priority_keywords.items()
            },
            "generic_verbs": {
                word: work_type.value
                for word, work_type in sorted(self.generic_verbs.items())
            },
        }

    @classmethod
    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-047"])
    def from_dict(cls, data: dict) -> IntentVocabulary:
        """Deserialize from a dict produced by :meth:`to_dict`."""
        return cls(
            work_type_keywords={
                WorkType(k): set(v)
                for k, v in data.get("work_type_keywords", {}).items()
            },
            action_keywords={
                IntentAction(k): set(v)
                for k, v in data.get("action_keywords", {}).items()
            },
            target_keywords={
                IntentTarget(k): set(v)
                for k, v in data.get("target_keywords", {}).items()
            },
            priority_keywords={
                IntentPriority(k): set(v)
                for k, v in data.get("priority_keywords", {}).items()
            },
            generic_verbs={
                word: WorkType(value)
                for word, value in data.get("generic_verbs", {}).items()
            },
        )