orchestrata_ultimate.orchestrator.intent_classifier

Reference › orchestrata_ultimate › orchestrator › intent_classifier

This is where a human sentence becomes a typed work order. Concierge passes the user's text to the intent_broker (also reached from the gateway RPC layer and the intent_demo script), and the broker hands free text to the IntentEvaluator in this module. It is traced to spec 003 story US-001 and depends on nlp, intent_normalizer, intent_vocabulary, intent_types and the work-type tables of pipeline_router.

NLIntentClassifier is a two-tier cascade. classify first runs named-entity recognition to find a resource path and an assigned volk. If a Prolog bridge is wired, it loads intent_grammar.pl once from the Prolog directory (_ensure_grammar_loaded) and queries parse_intent_evidence with the sanitized text; a successful parse becomes a ClassifiedIntent with tier set to grammar. If the bridge is absent, the grammar file is missing, the query fails or no parse is found, _classify_via_vocabulary scores the tokens against the keyword registries and picks a work type, action, target and priority. classify_async puts the normalizer in front: the turn, plus the preceding turns as context, is rewritten to a clean sentence first, and the question flag is re-read from what the user actually typed. Normalization failure leaves the text unchanged. to_intent converts the result into an Intent.

WorkflowMapper maps the classified work type to a workflow (with overrides) and to its StageSpec list. IntentEvaluator.evaluate_async (and the synchronous evaluate) returns an EvaluationResult with the intent, work type, confidence, reasons and stages. A turn is marked as needing disambiguation, and gets no stages, when its confidence is below min_confidence or when it is a question: a question is answered, not dispatched as work (REQ-003-US-001-070). The reasons list records the work type, confidence, tier and the decisive-word evidence for the audit trail.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

ClassifiedIntent (dataclass)

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-046', 'REQ-003-US-001-049']), dataclass · Line 87

Result of classifying a natural-language string.

NLIntentClassifier

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-042', 'REQ-003-US-001-043', 'REQ-003-US-001-046', 'REQ-003-US-… · Line 252

Cascade NL classifier: Prolog DCG grammar tier, vocabulary fallback tier.

WorkflowMapper

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-044']) · Line 540

Maps a ClassifiedIntent to a workflow WorkType + its PipelineRouter stage chain.

EvaluationResult (dataclass)

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-045']), dataclass · Line 570

Result of :meth:IntentEvaluator.evaluate.

IntentEvaluator

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-045']) · Line 586

The evaluator step: NL text -> confidence-gated EvaluationResult.

Functions

Source

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

"""Extensible NL intent classifier + workflow mapper + evaluator.

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

Borrows the cascade shape of dr-hannah's ``NaturalLanguageQueryParser``
(~/workspace/dr-hannah/src/dr_hannah/nlq.py): a Prolog DCG grammar tier
(preferred, ``intent_grammar.pl``) falling back to vocabulary keyword
scoring (``intent_vocabulary.py``) when no Prolog bridge is wired or the
grammar returns ``unknown``. ``NamedEntityRecognizer`` (orchestrator/nlp.py)
supplies ``resource_path``/``assigned_volk``/entities in both tiers — never
fabricated.

Composed with a ``WorkflowMapper`` (category -> ``PipelineRouter`` stage
chain) and an ``IntentEvaluator`` (the evaluator step: NL text ->
confidence-gated ``EvaluationResult``).
"""

from __future__ import annotations

import logging
from collections.abc import Sequence
from dataclasses import dataclass, field
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.paths import get_prolog_dir
from orchestrata_ultimate.orchestrator.intent_normalizer import IntentNormalizer
from orchestrata_ultimate.orchestrator.intent_types import (
    Intent,
    IntentAction,
    IntentPriority,
    IntentTarget,
)
from orchestrata_ultimate.orchestrator.intent_vocabulary import IntentVocabulary
from orchestrata_ultimate.orchestrator.nlp import (
    EntityType,
    NamedEntity,
    NamedEntityRecognizer,
    Tokenizer,
)
from orchestrata_ultimate.orchestrator.pipeline_router import (
    PipelineRouter,
    StageSpec,
    WorkType,
)

logger = logging.getLogger(__name__)

_GRAMMAR_FILE = "intent_grammar.pl"

# WorkType (from the grammar/vocabulary tiers) -> a representative IntentAction.
_WORK_TYPE_ACTION_MAP: dict[WorkType, IntentAction] = {
    WorkType.FIX: IntentAction.MODIFY,
    WorkType.FEATURE: IntentAction.CREATE,
    WorkType.REFACTOR: IntentAction.MODIFY,
    WorkType.DOCUMENTATION: IntentAction.WRITE,
    WorkType.TEST: IntentAction.EXECUTE,
    WorkType.REVIEW: IntentAction.REVIEW,
    WorkType.ANALYSIS: IntentAction.ANALYZE,
    WorkType.PERFORMANCE: IntentAction.ANALYZE,
    WorkType.DEPENDENCY: IntentAction.MODIFY,
    WorkType.SPEC: IntentAction.PLAN,
}

# Grammar-tier target atom (lowercase, matches intent_grammar.pl's lexicon) -> IntentTarget.
_TARGET_ATOM_MAP: dict[str, IntentTarget] = {
    "file": IntentTarget.FILE,
    "code": IntentTarget.CODE,
    "database": IntentTarget.DATABASE,
    "api": IntentTarget.API,
    "system": IntentTarget.SYSTEM,
    "directory": IntentTarget.DIRECTORY,
    "command": IntentTarget.COMMAND,
    "git": IntentTarget.GIT,
    "network": IntentTarget.NETWORK,
    "impact": IntentTarget.IMPACT,
    "architecture": IntentTarget.ARCHITECTURE,
}


@traces(
    spec="003",
    story="US-001",
    reqs=["REQ-003-US-001-046", "REQ-003-US-001-049"],
)
@dataclass
class ClassifiedIntent:
    """Result of classifying a natural-language string."""

    work_type: WorkType
    action: IntentAction
    target: IntentTarget
    priority: IntentPriority = IntentPriority.NORMAL
    title: str = ""
    resource_path: str = ""
    assigned_volk: str | None = None
    entities: list[NamedEntity] = field(default_factory=list)
    confidence: float = 0.0
    scores: dict[str, float] = field(default_factory=dict)
    matched_keywords: dict[str, list[str]] = field(default_factory=dict)
    # The sentence the GRAMMAR actually read. Empty when classification ran
    # on the raw text (sync path, or the normalizer degraded to passthrough).
    # Stored so a verdict is reconstructable from the parser's input rather
    # than from what the user typed — the two differ by design once the
    # normalization stage is in front (REQ-003-US-001-062).
    normalized_text: str = ""

    # ── Evidence behind the verdict (REQ-003-US-001-068) ────────────────────
    # WHY THESE EXIST. `confidence` above carries one bit and no more: with
    # the grammar tier live it is 0.9 or 0.3 and nothing between, measured
    # over all 185 turns in the three corpora. The 0.5-0.7 keyword band in
    # `_confidence_from_scores` is unreachable while a Prolog bridge is
    # present (see the proof in that function's docstring). So a consumer
    # asking "how sure are we?" is told either `sure` or `not sure`, and any
    # dispatch threshold in (0.30, 0.85] is the same threshold.
    #
    # These fields carry what the parse actually knew, so a graded confidence
    # can be DERIVED later from production telemetry instead of stamped now
    # from a hand-picked constant. They are deliberately not folded into
    # `confidence` yet: the obvious candidate rule was measured on the corpus
    # and does not replicate (plans/intent-dcg-accuracy-design.md §2.7).
    #
    #   tier            "grammar" | "vocabulary" — which tier decided
    #   decisive_hits   decisive work-type words found in the sentence
    #   winning_hits    how many of them voted for the winning category
    #
    # decisive_hits == 0 on a "grammar" verdict means the parse rested on a
    # GENERIC VERB alone — 23 of 185 corpus turns — which today is
    # indistinguishable from a 3-word unanimous parse in the confidence value.
    tier: str = ""
    decisive_hits: int = 0
    winning_hits: int = 0

    # ── Speech act (REQ-003-US-001-070) ──────────────────────────────────────
    # Whether the user ASKED something rather than commissioned work.
    #
    # This is not cosmetic. Measured over 272 real user turns, 41% of the
    # questions reach a CONFIDENT work verdict: "why did you stop in that
    # state?" dispatches ANALYSIS at 0.9, "are you observing?" dispatches
    # TEST. The grammar is behaving correctly on its own terms — `why` and
    # `test` are decisive work-type words — but the turn is a question and
    # answering it with a pipeline is the wrong act entirely.
    #
    # Detected from the RAW text, because the signal is the user's own
    # punctuation and the normalizer rewrite may not preserve it.
    is_question: bool = False


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-070"])
def is_question(text: str) -> bool:
    """Did the user ASK something rather than commission work?

    A terminal ``?`` and nothing else. That is not a simplification for its own
    sake — it is what measured best on 272 labelled real turns:

        rule                          precision   recall   (held-out split)
        ends with '?'                     0.862    0.833
        wh/modal opening word             0.762    0.533
        either of the two                 0.784    0.967
        both together                     0.923    0.400

    Recall matters more than the last points of precision here, because the
    cost of the two errors is asymmetric and not in the direction one first
    assumes. Missing a question means dispatching a work pipeline at someone
    who asked a question; over-firing means asking to confirm a request that
    was already going to be confirmed by a human anyway.

    Wh-word openings are deliberately NOT used. `why`, `what` and `how` are
    already decisive work-type words in the lexicon (`why` -> analysis), so
    matching on them would double-count the exact signal the grammar is
    reading, and it measured worse on its own.

    Applied to the corpus, this stops 35 currently-dispatching turns: 29 are
    questions or statements (82.9%) and 6 are requests that will now ask for
    confirmation rather than dispatch unattended.
    """
    return text.strip().endswith("?")


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-068"])
def _as_int(value: object) -> int:
    """Coerce a janus-marshalled Prolog integer, defaulting to 0.

    Prolog bindings arrive as whatever janus could convert; an evidence count
    that failed to marshal must read as "no evidence" rather than crash a
    classification that is otherwise fine.
    """
    if isinstance(value, int):  # bool is an int subclass; already the answer
        return value
    if isinstance(value, float):
        return int(value)
    if isinstance(value, str):
        try:
            return int(value)
        except ValueError:
            return 0
    return 0


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-048"])
def _confidence_from_scores(scores: dict[str, float]) -> float:
    """Map keyword-hit scores to a ~0.5-0.7 confidence band.

    ``confidence = 0.5 + 0.2 * margin_ratio`` where ``margin_ratio`` is the
    (top - runner-up) hit-count gap normalized by the total hits. A single
    unambiguous category -> 0.7; a tie between categories -> 0.5; no
    matches at all -> 0.3 (below the default IntentEvaluator threshold).

    THE 0.5-0.7 BAND IS UNREACHABLE WHENEVER A PROLOG BRIDGE IS PRESENT, and
    that is a property of the cascade, not an accident of the corpus. Proof:

    1. Both tiers read the SAME work-type lexicon — ``intent_lexicon.gen.pl``
       is generated from this vocabulary (``gen_intent_lexicon.py``, gated by
       `lexicon-drift`), so a token is a decisive hit for the grammar exactly
       when it scores for the keyword tier.
    2. ``work_type_of/3`` succeeds on ANY decisive hit, and ``target_of/4``
       ends in a catch-all clause that cannot fail, so ``classify_tokens/2``
       returns a term whenever one decisive hit exists.
    3. Therefore the keyword tier is reached only when there were no decisive
       hits at all — in which case ``scores`` is empty and this returns 0.3.

    Measured, consistent with the proof: 0 of 185 turns across the three
    corpora landed between 0.3 and 0.9. Confidence carries one bit, so every
    ``IntentEvaluator`` threshold in (0.30, 0.85] behaves identically. Making
    it graded is REQ-003-US-001-069; the evidence fields on ``ClassifiedIntent``
    are step one of it.

    The band is kept, not deleted: it is live and correct for the
    bridge-less configuration (``prolog_bridge=None``), which is the sync
    path several callers still use.
    """
    if not scores:
        return 0.3
    ordered = sorted(scores.values(), reverse=True)
    top = ordered[0]
    runner_up = ordered[1] if len(ordered) > 1 else 0.0
    total = sum(ordered)
    margin_ratio = (top - runner_up) / total if total else 0.0
    return round(0.5 + 0.2 * margin_ratio, 4)


@traces(
    spec="003",
    story="US-001",
    reqs=[
        "REQ-003-US-001-042",
        "REQ-003-US-001-043",
        "REQ-003-US-001-046",
        "REQ-003-US-001-048",
    ],
)
class NLIntentClassifier:
    """Cascade NL classifier: Prolog DCG grammar tier, vocabulary fallback tier.

    Mirrors dr-hannah's ``NaturalLanguageQueryParser(prolog_engine=None)``:
    try the grammar backend first when a ``prolog_bridge`` is wired, fall
    back gracefully to keyword scoring otherwise (or when the grammar
    returns ``unknown``).
    """

    def __init__(
        self,
        vocabulary: IntentVocabulary | None = None,
        prolog_bridge: Any = None,
        tokenizer: Tokenizer | None = None,
        ner: NamedEntityRecognizer | None = None,
        normalizer: IntentNormalizer | None = None,
    ) -> None:
        self.vocabulary = vocabulary or IntentVocabulary.default()
        self.prolog_bridge = prolog_bridge
        self.tokenizer = tokenizer or Tokenizer()
        self.ner = ner or NamedEntityRecognizer()
        self.normalizer = normalizer or IntentNormalizer()
        self._grammar_loaded = False

    @traces(
        spec="003", story="US-001", reqs=["REQ-003-US-001-043", "REQ-003-US-001-046"]
    )
    def classify(self, text: str) -> ClassifiedIntent:
        """Classify natural-language text into a ``ClassifiedIntent``.

        Cascade: Prolog DCG grammar (confidence ~0.9) -> vocabulary keyword
        scoring (confidence ~0.5-0.7). NER supplies resource_path/
        assigned_volk/entities in both tiers — never fabricated.
        """
        entities = self.ner.recognize(text)
        resource_path = self._resource_path_from_entities(entities)
        assigned_volk = self._assigned_volk_from_entities(entities)

        if self.prolog_bridge is not None:
            grammar_result = self._classify_via_grammar(text)
            if grammar_result is not None:
                grammar_result.entities = entities
                grammar_result.resource_path = resource_path
                grammar_result.assigned_volk = assigned_volk
                grammar_result.is_question = is_question(text)
                return grammar_result

        result = self._classify_via_vocabulary(text)
        result.entities = entities
        result.resource_path = resource_path
        result.assigned_volk = assigned_volk
        result.is_question = is_question(text)
        return result

    @traces(
        spec="003",
        story="US-001",
        reqs=["REQ-003-US-001-060", "REQ-003-US-001-062", "REQ-003-US-001-063"],
    )
    async def classify_async(
        self, text: str, context: Sequence[str] | None = None
    ) -> ClassifiedIntent:
        """Normalize the turn, then classify — the full production cascade.

        The grammar is a keyword parser and people do not type keywords.
        Measured over 76 real user turns from this project's own history
        (``tests/orchestrator/data/intent_corpus_raw_turns.json``), the grammar
        scored 19/76 on the raw text and 29/76 once the same turns were
        rewritten as plain sentences — the gap is spelling and phrasing, not
        vocabulary. So the normalizer runs FIRST, unconditionally, and Prolog
        only ever parses clean input (design:
        plans/intent-dcg-accuracy-design.md §2.5).

        The shipped prompt currently captures about 3 of those 10 points
        (26/92 independently measured); closing the rest is open work, not a
        solved problem.

        ``context`` is the preceding turns, oldest first, and is what makes
        "it hangs again when I drain" resolvable at all — no amount of
        vocabulary reaches a sentence whose subject is a pronoun.

        Normalization degrades to passthrough on any failure, so this method
        is never worse than :meth:`classify`; it is the same cascade with a
        better input.
        """
        normalized = await self.normalizer.normalize(text, context)
        result = self.classify(normalized)
        # Attribute the title to what the user actually said; record the
        # parsed sentence separately.
        result.normalized_text = normalized
        # Re-read the speech act from what the USER typed. The normalizer is
        # told to produce a declarative sentence for the parser, so it strips
        # the question mark that is the whole signal — classifying the rewrite
        # would lose it every time.
        result.is_question = is_question(text)
        if normalized != text:
            result.title = text.strip()
        return result

    # ── Prolog grammar tier ─────────────────────────────────────────────

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-042"])
    def _ensure_grammar_loaded(self) -> bool:
        """Load intent_grammar.pl into the wired bridge (once)."""
        if self._grammar_loaded:
            return True
        if self.prolog_bridge is None:
            return False
        grammar_path = get_prolog_dir() / _GRAMMAR_FILE
        if not grammar_path.exists():
            return False
        loaded: bool = self.prolog_bridge.load_rules(grammar_path)
        self._grammar_loaded = loaded
        return loaded

    @traces(
        spec="003", story="US-001", reqs=["REQ-003-US-001-042", "REQ-003-US-001-048"]
    )
    def _classify_via_grammar(self, text: str) -> ClassifiedIntent | None:
        """Try the DCG grammar tier. Returns None on unknown/unavailable."""
        if not self._ensure_grammar_loaded():
            return None

        sanitized = text.replace("\\", "\\\\").replace("'", "\\'")
        # parse_intent_evidence/4, not parse_intent_string/2: same parse, but
        # it also returns the vote tally the grammar already computed and
        # previously discarded. Proved term-for-term identical to the /2 form
        # over all 185 corpus turns before the switch — the evidence rides
        # along, it does not steer.
        query_str = (
            f"parse_intent_evidence('{sanitized}', "
            "work_intent(WorkType, Target, Rest), Wins, Total)"
        )
        try:
            result = self.prolog_bridge.query(query_str)
        except Exception as exc:  # pragma: no cover - defensive, mirrors dr-hannah
            logger.warning("intent_grammar_query_failed: %s", exc)
            return None

        if not result.success or not result.solutions:
            return None

        return self._prolog_term_to_classified(text, result.solutions[0])

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-042"])
    def _prolog_term_to_classified(
        self, text: str, solution: dict[str, Any]
    ) -> ClassifiedIntent | None:
        """Convert a ``work_intent(WorkType, Target, Rest)`` solution to a
        ``ClassifiedIntent``. Mirrors dr-hannah's ``_prolog_term_to_query``.
        """
        work_type_raw = str(solution.get("WorkType", ""))
        target_raw = str(solution.get("Target", ""))

        try:
            work_type = WorkType(work_type_raw)
        except ValueError:
            return None

        target = _TARGET_ATOM_MAP.get(target_raw, IntentTarget.CODE)
        action = _WORK_TYPE_ACTION_MAP.get(work_type, IntentAction.MODIFY)

        # Evidence is reported, never scored — `confidence` stays the flat 0.9
        # this tier has always returned. Grading it is REQ-003-US-001-069 and
        # needs production telemetry; the one candidate rule available from the
        # corpus (single decisive hit predicts correctness) reverses direction
        # on held-out data and was rejected.
        return ClassifiedIntent(
            work_type=work_type,
            action=action,
            target=target,
            priority=self._priority_from_vocabulary(text),
            title=text.strip(),
            confidence=0.9,
            scores={work_type.value: 1.0},
            matched_keywords={"grammar": [work_type_raw]},
            tier="grammar",
            decisive_hits=_as_int(solution.get("Total")),
            winning_hits=_as_int(solution.get("Wins")),
        )

    # ── Vocabulary tier ──────────────────────────────────────────────────

    @traces(
        spec="003", story="US-001", reqs=["REQ-003-US-001-043", "REQ-003-US-001-041"]
    )
    def _classify_via_vocabulary(self, text: str) -> ClassifiedIntent:
        tokens = {t.normalized for t in self.tokenizer.tokenize(text)}

        work_type, work_scores, work_matched = self._score(
            tokens, self.vocabulary.work_type_keywords, WorkType.FEATURE
        )
        action, _, action_matched = self._score(
            tokens, self.vocabulary.action_keywords, IntentAction.MODIFY
        )
        target, _, target_matched = self._score(
            tokens, self.vocabulary.target_keywords, IntentTarget.CODE
        )
        priority = self._priority_from_vocabulary(text, tokens)

        return ClassifiedIntent(
            work_type=work_type,
            action=action,
            target=target,
            priority=priority,
            title=text.strip(),
            confidence=_confidence_from_scores(work_scores),
            scores=work_scores,
            matched_keywords={
                "work_type": work_matched,
                "action": action_matched,
                "target": target_matched,
            },
            tier="vocabulary",
            # This tier has no vote to report: it scores hits, it does not
            # tally them per category. `work_matched` is the raw evidence.
            decisive_hits=len(work_matched),
            winning_hits=len(work_matched),
        )

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-041"])
    def _priority_from_vocabulary(
        self, text: str, tokens: set[str] | None = None
    ) -> IntentPriority:
        toks = (
            tokens
            if tokens is not None
            else {t.normalized for t in self.tokenizer.tokenize(text)}
        )
        priority: IntentPriority
        priority, _, _ = self._score(
            toks, self.vocabulary.priority_keywords, IntentPriority.NORMAL
        )
        return priority

    @staticmethod
    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-043"])
    def _score(
        tokens: set[str], registry: dict[Any, set[str]], default: Any
    ) -> tuple[Any, dict[str, float], list[str]]:
        """Score each registry category by keyword hit count; argmax wins."""
        scores: dict[str, float] = {}
        matched: dict[Any, list[str]] = {}
        for category, keywords in registry.items():
            hits = sorted(tokens & keywords)
            if hits:
                scores[category.value] = float(len(hits))
                matched[category] = hits

        if not matched:
            return default, {}, []

        best_category = max(matched, key=lambda c: len(matched[c]))
        return best_category, scores, matched[best_category]

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-046"])
    def _resource_path_from_entities(self, entities: list[NamedEntity]) -> str:
        for entity in entities:
            if entity.entity_type == EntityType.FILE_PATH:
                return entity.text
        return ""

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-046"])
    def _assigned_volk_from_entities(self, entities: list[NamedEntity]) -> str | None:
        for entity in entities:
            if entity.entity_type == EntityType.AGENT:
                return entity.text
        return None

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-049"])
    def to_intent(
        self, classified: ClassifiedIntent, requester: str = "U:Ox"
    ) -> Intent:
        """Build a broker ``Intent`` from a ``ClassifiedIntent``."""
        return Intent(
            action=classified.action,
            target=classified.target,
            resource_path=classified.resource_path,
            priority=classified.priority,
            requester=requester,
            parameters={
                "work_type": classified.work_type.value,
                "title": classified.title,
            },
        )


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-044"])
class WorkflowMapper:
    """Maps a ``ClassifiedIntent`` to a workflow ``WorkType`` + its
    ``PipelineRouter`` stage chain.

    Extensible via :meth:`register` — callers can override the
    category -> workflow mapping without code edits.
    """

    def __init__(self, overrides: dict[WorkType, WorkType] | None = None) -> None:
        self._overrides: dict[WorkType, WorkType] = dict(overrides or {})
        self._router = PipelineRouter()

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-044"])
    def register(self, category: WorkType, workflow: WorkType) -> None:
        """Register (or override) a category -> workflow mapping."""
        self._overrides[category] = workflow

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-044"])
    def workflow_for(self, classified: ClassifiedIntent) -> WorkType:
        """Resolve the workflow ``WorkType`` for a classified intent."""
        return self._overrides.get(classified.work_type, classified.work_type)

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-044"])
    def stages_for(self, work_type: WorkType, has_ui: bool = False) -> list[StageSpec]:
        """Resolve the real ``PipelineRouter`` stage chain for a work type."""
        return self._router.route(work_type, has_ui=has_ui)


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-045"])
@dataclass
class EvaluationResult:
    """Result of :meth:`IntentEvaluator.evaluate`."""

    intent: Intent
    work_type: WorkType
    stages: list[StageSpec]
    confidence: float
    reasons: list[str] = field(default_factory=list)
    needs_disambiguation: bool = False
    # The classification this verdict was built from, evidence included
    # (REQ-003-US-001-068). Without it the evidence fields are unreachable from
    # the only object callers actually receive, which makes them decoration.
    classified: ClassifiedIntent | None = None


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-045"])
class IntentEvaluator:
    """The evaluator step: NL text -> confidence-gated ``EvaluationResult``.

    Ties together ``IntentVocabulary`` (extensible keywords),
    ``NLIntentClassifier`` (the cascade), and ``WorkflowMapper``
    (category -> workflow + stages). Below ``min_confidence``, the result
    is flagged ``needs_disambiguation`` and stages are withheld.
    """

    def __init__(
        self,
        vocabulary: IntentVocabulary | None = None,
        classifier: NLIntentClassifier | None = None,
        mapper: WorkflowMapper | None = None,
        min_confidence: float = 0.34,
    ) -> None:
        self.vocabulary = vocabulary or IntentVocabulary.default()
        self.classifier = classifier or NLIntentClassifier(vocabulary=self.vocabulary)
        self.mapper = mapper or WorkflowMapper()
        self.min_confidence = min_confidence

    @traces(
        spec="003",
        story="US-001",
        reqs=["REQ-003-US-001-045", "REQ-003-US-001-063"],
    )
    async def evaluate_async(
        self,
        text: str,
        requester: str = "U:Ox",
        context: Sequence[str] | None = None,
    ) -> EvaluationResult:
        """Evaluate with the normalization stage in front — the path the
        broker uses in production.

        Identical to :meth:`evaluate` except the grammar reads a cleaned
        sentence instead of raw typing. ``context`` is the preceding turns,
        oldest first.
        """
        classified = await self.classifier.classify_async(text, context)
        return self._evaluation_from(classified, requester)

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-045"])
    def evaluate(self, text: str, requester: str = "U:Ox") -> EvaluationResult:
        """Classify, map to a workflow, and gate on confidence.

        The synchronous path: no normalization, so the grammar reads the raw
        text. Retained for callers that have no event loop; prefer
        :meth:`evaluate_async`.
        """
        classified = self.classifier.classify(text)
        return self._evaluation_from(classified, requester)

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-045"])
    def _evaluation_from(
        self, classified: ClassifiedIntent, requester: str
    ) -> EvaluationResult:
        """Shared tail of both evaluate paths — workflow mapping plus the
        confidence gate."""
        intent = self.classifier.to_intent(classified, requester=requester)
        work_type = self.mapper.workflow_for(classified)

        # A question does not auto-dispatch, however confident the parse.
        # "why did you stop in that state?" parses as ANALYSIS at 0.9 because
        # `why` is a decisive analysis word — the grammar is right and the
        # dispatch is still wrong. Confidence cannot express this: the parse is
        # good, the SPEECH ACT is not a work order (REQ-003-US-001-070).
        needs_disambiguation = (
            classified.confidence < self.min_confidence or classified.is_question
        )
        reasons = [
            f"work_type={work_type.value}",
            f"confidence={classified.confidence:.2f}",
            # The evidence, in the human-readable trail, because "confidence
            # 0.90" alone cannot distinguish a unanimous three-word parse from
            # a generic-verb guess — and both occur.
            f"tier={classified.tier}",
            f"evidence={classified.winning_hits}/{classified.decisive_hits}"
            " decisive words",
        ]
        if classified.is_question:
            # A distinct reason, because it is a distinct situation: the parse
            # succeeded and the turn still must not become work. A caller that
            # sees this should ANSWER, not ask the user to rephrase.
            reasons.append(
                "turn is a question — answer it, do not dispatch work for it"
            )
        elif needs_disambiguation:
            reasons.append(
                "confidence below min_confidence threshold — disambiguation needed"
            )

        stages: list[StageSpec] = (
            [] if needs_disambiguation else self.mapper.stages_for(work_type)
        )

        return EvaluationResult(
            intent=intent,
            work_type=work_type,
            stages=stages,
            confidence=classified.confidence,
            reasons=reasons,
            needs_disambiguation=needs_disambiguation,
            classified=classified,
        )