File size: 5,593 Bytes
e201ae4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
"""Tool groups β€” the planner sees a slice, not the whole registry (A17).

With every analytics tool active the Planner-visible set is **18 tools**, inside the
measured 15–20 band where tool-selection accuracy starts to degrade β€” and the driver is
semantic similarity, not raw count, which is exactly what this registry has
(`analyze_trend` vs `analyze_forecast`, `analyze_correlation` vs `analyze_importance`,
`analyze_segmentation` vs `analyze_cluster`). Rather than turn tools off, the router
names a GROUP and the Planner is shown only that group's tools plus the always-on core:
roughly 8, not 18.

**Three layers, and it matters which does what** (decided 2026-09-17):

    router (LLM, already runs) -> emits group_hint on its existing structured output
      -> THIS MODULE (deterministic, 0 LLM) -> filters the registry
        -> planner (LLM) -> plans against the slice, unchanged

The filter is deliberately NOT the planner's job: to choose a group it would have to be
shown every group, which is the thing being avoided, and a second LLM call would break
the two-call invariant for a benefit that only pays above ~60 tools. It is not the
router's job either β€” the router classifies, it does not filter.

**Fail open, always.** The router runs ~95 lines before the catalog is read
(`chat_handler`: classify at 334, catalog at 429), so it is CATALOG-BLIND: it can judge
what a question is *shaped* like, never what the data can support. Feasibility stays the
Planner's job (recipes R6 and R15). And because a filtered-out tool is one the Planner cannot
ask for, a wrong hint is unrecoverable β€” so an absent, unknown, or empty hint yields the
FULL registry. Slightly worse selection always beats an unrecoverable miss.

Groups are derived from each spec's existing `category` rather than stored on `ToolSpec`:
grouping is a planner concern (the 2026-09-15 discussion: *"ini tuh dari planner-nya"*),
`category` is already almost the taxonomy, and deriving keeps the tool layer untouched.
An unmapped category falls back to `explore`, so a new tool is visible-by-default rather
than silently invisible β€” the safe direction for this failure.
"""

from __future__ import annotations

from .contracts import ToolRegistry, ToolSpec

# --------------------------------------------------------------------------- #
# The taxonomy (A0, 2026-09-17)
# --------------------------------------------------------------------------- #

DATA_ACCESS = "data_access"
EXPLORE = "explore"
RELATIONSHIP = "relationship"
TIMESERIES = "timeseries"
GROUPING = "grouping"
EVALUATE = "evaluate"
UTILITY = "utility"

#: Groups the router may name. `data_access` and `utility` are never hinted β€” they are
#: always included (see `ALWAYS_ON`), so hinting them would mean nothing.
HINTABLE_GROUPS: tuple[str, ...] = (EXPLORE, RELATIONSHIP, TIMESERIES, GROUPING, EVALUATE)

#: Always in scope whatever the hint: the four data-access tools (nothing can run
#: without them) and `render_chart` (any group's answer may carry a viz tail).
ALWAYS_ON = (DATA_ACCESS, UTILITY)

#: One line per group, for the router prompt. Written for a classifier that has the
#: QUESTION and no catalog β€” phrased by what the user is asking for, not by method.
GROUP_PURPOSE: dict[str, str] = {
    EXPLORE: "summarising, totalling or checking the data itself",
    RELATIONSHIP: "what relates to, explains or contributes to a measure",
    TIMESERIES: "movement over time, including projecting it forward",
    GROUPING: "splitting rows into segments or discovering groups",
    EVALUATE: "comparing two groups, or finding unusual rows",
}

_CATEGORY_TO_GROUP: dict[str, str] = {
    # data access β€” always on
    "catalog.introspection": DATA_ACCESS,
    "analytics.query": DATA_ACCESS,
    "retrieval.documents": DATA_ACCESS,
    # explore
    "analytics.descriptive": EXPLORE,
    "analytics.aggregation": EXPLORE,
    "analytics.quality": EXPLORE,
    # relationship
    "analytics.relationship": RELATIONSHIP,  # correlation + importance
    "analytics.pareto": RELATIONSHIP,
    # timeseries
    "analytics.timeseries": TIMESERIES,  # trend + forecast
    # grouping
    "analytics.segmentation": GROUPING,
    "analytics.clustering": GROUPING,
    # evaluate
    "analytics.comparison": EVALUATE,  # ab_test
    "analytics.anomaly": EVALUATE,
    # utility β€” always on
    "analytics.combine": UTILITY,
    "analytics.visualization": UTILITY,
}

#: An unmapped category lands here. Visible-by-default: a tool nobody grouped should be
#: selectable, not silently withheld.
_UNGROUPED = EXPLORE


def group_of(spec: ToolSpec) -> str:
    """The group a spec belongs to, derived from its `category`."""
    return _CATEGORY_TO_GROUP.get(spec.category, _UNGROUPED)


def scope(registry: ToolRegistry, hints: list[str] | None) -> ToolRegistry:
    """The slice of `registry` a turn hinted at `hints` should plan against.

    Returns the FULL registry when `hints` is empty, None, or names nothing known β€”
    the fail-open rule. Otherwise: everything in `ALWAYS_ON` plus every hinted group,
    preserving the registry's own ordering so the planner prompt stays stable.
    """
    wanted = {h for h in (hints or []) if h in HINTABLE_GROUPS}
    if not wanted:
        return ToolRegistry(tools=list(registry.tools))
    keep = wanted | set(ALWAYS_ON)
    return ToolRegistry(tools=[t for t in registry.tools if group_of(t) in keep])


def render_group_menu() -> str:
    """The group list for the router prompt, one line each."""
    return "\n".join(f"- `{g}` β€” {GROUP_PURPOSE[g]}" for g in HINTABLE_GROUPS)