Rifqi Hafizuddin
Claude Opus 5.5
[NOTICKET] feat(scope): code-owned conversation scope — focus from value lists, follow-up carry, scope line
111ef6b
Raw History Blame
51.9 kB
"""Scope engine — merge, apply and render conversation scope in code (2026-10-29).
The planner only DECLARES scope (`ScopeChange`, and once per analysis the
objective's `focus`). Everything that decides what an answer actually covers lives
here, because text the planner "should" apply was applied on 7 of 27 turns:
- `resolve_scope` merges pinned focus / previous scope / this turn's change into
the effective scope. A value is only accepted when the user actually typed it
(`grounded`) — the planner may misread a message, it may not invent a code.
- `apply_scope_to_plan` injects the effective scope into every `retrieve_data`
IR, adding the lookup join when exactly one foreign key reaches the column.
- `render_scope_line` describes what RAN (the executed filters in each output's
meta), not what was intended, so the line can never name a scope its query
never filtered.
Pure and never-throw (house rule: these sit on the answer path). Generic: an
entry is any non-PII string column, matched by NAME case-insensitively; nothing
here knows what a site or a section is. Integer columns (year/month/week) are
periods, not scope.
`TaskList` / `RunState` / `ToolOutput` are read duck-typed and imported only for
type checking: `planner.schemas` imports `scope.models`, so a runtime import back
into the planner/slow-path packages would be circular.
"""
from __future__ import annotations
import re
from typing import TYPE_CHECKING, Any
from src.catalog.models import Catalog, Column, Source, Table
from src.middlewares.logging import get_logger
from src.query.ir.models import FilterClause, Join, QueryIR
from .models import ScopeChange, ScopeEntry, TurnScope
if TYPE_CHECKING:
from src.agents.planner.schemas import TaskList
from src.agents.slow_path.schemas import RunState
logger = get_logger("scope_engine")
_EQ_OPS = ("=", "in")
_MAX_OTHER_FILTERS = 4
# --------------------------------------------------------------------------- #
# Small helpers
# --------------------------------------------------------------------------- #
def _norm(s: str) -> str:
return re.sub(r"\s+", " ", s).strip().lower()
def grounded(value: str, text: str) -> bool:
"""True when `value` appears in `text` as a whole token run.
Case-insensitive and whitespace-collapsed ("COAL LOADER" matches "coal loader").
The value must be bounded by non-alphanumeric characters or the string ends, so
"IPR" matches "kalo di ipr?" but not "iprx" — a code that merely prefixes another
word is not something the user asked for.
"""
try:
v = _norm(str(value or ""))
if not v:
return False
t = _norm(str(text or ""))
# [^\W_] = an alphanumeric char (unicode-aware), so the lookarounds forbid a
# letter/digit on either side but allow punctuation, spaces and string ends.
pattern = r"(?<![^\W_])" + re.escape(v) + r"(?![^\W_])"
return re.search(pattern, t) is not None
except Exception as e:
logger.warning("scope_grounded_failed", extra={"error": repr(e)})
return False
def _eligible(col: Column) -> bool:
return col.data_type == "string" and not col.pii_flag
def _eligible_columns(catalog: Catalog) -> dict[str, list[Column]]:
"""Eligible columns across the catalog, keyed by lower-cased NAME."""
out: dict[str, list[Column]] = {}
for s in catalog.sources:
for t in s.tables:
for c in t.columns:
if _eligible(c):
out.setdefault(c.name.lower(), []).append(c)
return out
def _canonical_values(cols: list[Column]) -> dict[str, str]:
"""lower-cased value -> the catalog's spelling, from samples and top values."""
out: dict[str, str] = {}
for c in cols:
known: list[Any] = list(c.sample_values or [])
if c.stats is not None and c.stats.top_values:
known.extend(c.stats.top_values)
for v in known:
if v is None:
continue
out.setdefault(_norm(str(v)), str(v))
return out
def _canon(value: str, canon: dict[str, str]) -> str:
return canon.get(_norm(value), value)
def _dedupe(items: list[str]) -> list[str]:
seen: set[str] = set()
out: list[str] = []
for i in items:
if i not in seen:
seen.add(i)
out.append(i)
return out
def _upsert(entries: list[ScopeEntry], entry: ScopeEntry) -> None:
"""Replace the entry for the same column (case-insensitive), else append."""
for i, e in enumerate(entries):
if e.column.lower() == entry.column.lower():
entries[i] = entry
return
entries.append(entry)
def _filter_values(value: Any) -> list[str]:
if isinstance(value, list | tuple):
return [str(v) for v in value]
return [str(value)]
def _retrieve_calls(task_list: Any) -> list[Any]:
calls: list[Any] = []
for task in getattr(task_list, "tasks", None) or []:
for call in getattr(task, "tool_calls", None) or []:
if getattr(call, "tool", None) == "retrieve_data":
calls.append(call)
return calls
#: A value handoff from an earlier task: "${t1}" / "${t2.site_id}" (planner
#: `PLACEHOLDER_RE` + the validator's task-id rule). Mirrored, not imported: the
#: engine must not import planner modules at runtime (circular via planner.schemas).
_HANDOFF_RE = re.compile(r"^\$\{t\d+(\.[A-Za-z0-9_]+)?\}$")
def _is_handoff(value: Any) -> bool:
return isinstance(value, str) and bool(_HANDOFF_RE.match(value.strip()))
#: Name parts too generic to say WHICH column a user means ("site_code" → "site").
_GENERIC_NAME_PARTS = frozenset({"code", "id", "no", "name", "nm", "kd"})
def _column_named(column: str, text: str) -> bool:
"""True when the message names the column itself: its full name or a
non-generic part of it ("section" for `section`, "site" for `site_code`)."""
if grounded(column.replace("_", " "), text) or grounded(column, text):
return True
parts = [p for p in column.lower().split("_") if p and p not in _GENERIC_NAME_PARTS]
return any(
len(p) >= 3 and (grounded(p, text) or grounded(p + "s", text)) for p in parts
)
def _quantified(column: str, text: str) -> bool:
"""True when a quantifier sits right before the column's word: "semua site",
"per section", "all sites", "tiap section" — the user asked for every value of
it, whatever the planner declared (test 3, 2026-10-30: "… di semua site" stayed on
the focus site because the planner declared nothing)."""
parts = {p for p in column.lower().split("_") if len(p) >= 3 and p not in _GENERIC_NAME_PARTS}
words = re.findall(r"[a-z0-9]+", (text or "").lower())
for i, w in enumerate(words):
if w in _QUANTIFIERS:
for n in words[i + 1:i + 3]:
if any(n == p or n == p + "s" for p in parts):
return True
return False
def _cued(value: str, column: str, text: str) -> bool:
"""True when a non-generic part of `column` is one of the two words right before
`value` in `text` ("di site SDJ" for `site_code`)."""
parts = {p for p in column.lower().split("_") if len(p) >= 3 and p not in _GENERIC_NAME_PARTS}
if not parts:
return False
words = re.findall(r"[a-z0-9]+", text.lower())
target = _norm(value)
for i, w in enumerate(words):
if w == target and any(p in words[max(0, i - 2):i] for p in parts):
return True
return False
def _misplaced(value: str, column: str, value_index: dict[str, list[str]] | None) -> bool:
"""True when `value` is not in `column`'s list but IS in another column's list."""
if not value_index:
return False
lists = {k.lower(): {_norm(x) for x in v} for k, v in value_index.items()}
own = lists.get(column.lower())
if own is None or _norm(value) in own:
return False
return any(_norm(value) in vals for k, vals in lists.items() if k != column.lower())
def _focus_from_index(
value_index: dict[str, list[str]] | None,
objective: str,
eligible: dict[str, list[Column]],
notes: list[str],
require_cue: bool = False,
) -> list[ScopeEntry]:
"""Values of `objective` (any text) found in the column value lists, one entry
per column.
A value found under two DIFFERENT column names is ambiguous and dropped from
both; never picks between them. With `require_cue` (the objective, which pins the
focus for the whole analysis), a code of ≤3 letters counts only right after its
column word ("site SDJ"): review 2026-10-29 pinned "HO" from "review ke HO". The
user's follow-ups stay bare ("lat?", "kalo di bin?" — the real 7 Sep session).
"""
if not value_index or not objective:
return []
hits: dict[str, list[str]] = {}
owners: dict[str, set[str]] = {}
for name, values in value_index.items():
cols = eligible.get(name.lower())
if not cols:
continue
for v in values:
if not isinstance(v, str) or len(v.strip()) < 2:
continue
if require_cue and len(v.strip()) <= 3 and not _cued(v, cols[0].name, objective):
continue
if grounded(v, objective):
hits.setdefault(cols[0].name, []).append(v)
owners.setdefault(_norm(v), set()).add(cols[0].name.lower())
out: list[ScopeEntry] = []
for name, values in hits.items():
keep = [v for v in values if len(owners.get(_norm(v), set())) == 1]
dropped = [v for v in values if v not in keep]
# Longest match wins: "OB LOADER" is not also in scope when the objective
# says "OB LOADER / DRILL PUMP".
keep = [
v for v in keep
if not any(o is not v and _norm(v) != _norm(o) and grounded(v, o) for o in keep)
]
if dropped:
notes.append(f"focus {name}: ambiguous across columns {dropped}")
if keep:
out.append(ScopeEntry(column=name, values=keep))
return out
# --------------------------------------------------------------------------- #
# 1. Merge
# --------------------------------------------------------------------------- #
#: Universal quantifiers — grammar, not domain vocabulary. A planner "every value of
#: X" on a focus column is accepted when the user named the column OR used one of
#: these ("semua jobsite", "per lokasi": review 2026-10-29, synonyms of `site`).
_QUANTIFIERS = ("semua", "seluruh", "setiap", "tiap", "per", "all", "every", "each", "across")
#: A column whose values identify ONE entity (unit numbers: 885-2,334 distinct on the
#: production catalog; models 32-73). Naming one sets the other dimensions by the
#: entity's own attributes (requirement #4) — the focus is not forced onto it.
ENTITY_MIN_DISTINCT = 200
_NEG_OPS = ("!=", "not_in")
_TASK_REF_RE = re.compile(r"\$\{(t\d+)")
def _has_quantifier(text: str) -> bool:
return any(grounded(q, text) for q in _QUANTIFIERS)
def _entity_columns(catalog: Catalog) -> set[str]:
out: set[str] = set()
for s in catalog.sources:
for t in s.tables:
for c in t.columns:
dc = c.stats.distinct_count if c.stats else None
if _eligible(c) and dc is not None and dc > ENTITY_MIN_DISTINCT:
out.add(c.name.lower())
return out
class _PlanScan:
"""What the plan's retrieve_data IRs do with each eligible column (by name)."""
def __init__(self, task_list: Any, catalog: Catalog) -> None:
cols_by_id = {
c.column_id: c for s in catalog.sources for t in s.tables for c in t.columns
}
# A group-by on a foreign-key id groups by its dimension too: `site_id`
# (an int, never scope) → the site table's `site_code` (test 3, 2026-10-30:
# "MTTR … di semua site" grouped by site_id and the site rule never fired).
via_fk: dict[str, set[str]] = {}
tables = {t.table_id: t for s in catalog.sources for t in s.tables}
for t in tables.values():
for fk in t.foreign_keys:
target = tables.get(fk.target_table_id)
if target is not None:
via_fk.setdefault(fk.column_id, set()).update(
c.name.lower() for c in target.columns if _eligible(c)
)
self.grouped: set[str] = set()
self.filtered: set[str] = set()
self.eq: dict[str, list[str]] = {}
self.neg: dict[str, list[str]] = {}
for call in _retrieve_calls(task_list):
try:
ir = QueryIR.model_validate((call.args or {}).get("ir"))
except Exception as e:
logger.info("scope_ir_unparseable", extra={"error": repr(e)})
continue
for gid in ir.group_by:
col = cols_by_id.get(gid)
if col is not None and _eligible(col):
self.grouped.add(col.name.lower())
self.grouped |= via_fk.get(gid, set())
for f in ir.filters:
col = cols_by_id.get(f.column_id)
if col is None or not _eligible(col):
continue
name = col.name.lower()
self.filtered.add(name)
bucket = (
self.eq if f.op in _EQ_OPS else self.neg if f.op in _NEG_OPS else None
)
if bucket is None:
continue
values = bucket.setdefault(name, [])
for v in _filter_values(f.value):
if _norm(v) not in {_norm(x) for x in values}:
values.append(v)
def resolve_scope(
*,
pinned_focus: list[ScopeEntry] | None,
previous: list[ScopeEntry] | None,
change: ScopeChange | None,
objective: str,
raw_message: str,
task_list: TaskList,
catalog: Catalog,
value_index: dict[str, list[str]] | None = None,
rewritten: str = "",
) -> tuple[list[ScopeEntry], list[ScopeEntry], str, list[str]]:
"""Merge this turn's scope → `(focus, effective, base, log_notes)`.
Code decides scope; the planner contributes only language it alone can judge
(gate run 2, 2026-10-29: the more scope the planner was asked to handle, the more
it failed). Order, every step deterministic:
a. focus — pinned, else the objective matched against the column value lists (a
code of ≤3 letters only right after its column word: "site SDJ", never "ke HO");
b. negation — a planner `!=`/`not_in` on values the user typed ("selain SDJ")
opens that column (ALL); the planner's exclusion is kept as written;
c. the user's own words against the FOCUS columns' lists (a value lands only on
a column whose list holds it; "SDJ vs IPR" → site [SDJ, IPR]);
d. the planner's `scope_change.set`: ALL when the user named the column or used a
quantifier; values of columns too large to list, if typed;
e. the router's carry of the previous answer's values into a follow-up — added to
the user's value for a column the plan groups by ("bandingkan dengan SDJ");
f. the planner's filters on typed values (a code in no list is kept: "not found",
never swapped for the focus; one that belongs to another column is not);
g. a breakdown the user asked for ("per section") as ALL;
h. an entity (a unit) named or carried lifts the focus columns for the turn.
Never raises: a failure degrades to the pinned focus as the effective scope.
"""
try:
notes: list[str] = []
eligible = _eligible_columns(catalog)
indexed = {k.lower() for k, v in (value_index or {}).items() if v}
scan = _PlanScan(task_list, catalog)
def canon_col(name: str) -> str:
cols = eligible.get(name.lower())
return cols[0].name if cols else name
# a. focus
if pinned_focus is not None:
focus = [e.model_copy(deep=True) for e in pinned_focus]
else:
focus = _focus_from_index(value_index, objective, eligible, notes, require_cue=True)
focus_cols = {e.column.lower() for e in focus}
effective = [e.model_copy(deep=True) for e in focus]
accepted: set[str] = set() # columns this message changed
# b. negation
negated: set[str] = set()
for name, values in scan.neg.items():
if name in eligible and values and all(grounded(v, raw_message) for v in values):
negated.add(name)
notes.append(f"negated {name}: {values}")
_upsert(effective, ScopeEntry(column=canon_col(name), values=None))
accepted.add(name)
# c. the user's words against the focus columns' value lists
focus_index = {
k: v
for k, v in (value_index or {}).items()
if k.lower() in focus_cols and k.lower() not in negated
}
for entry in _focus_from_index(focus_index, raw_message, eligible, notes):
notes.append(f"message {entry.column}: {entry.values}")
_upsert(effective, entry)
accepted.add(entry.column.lower())
# c2. "semua site" / "per section": every value of a focus column the user
# did not give a value for — decided from the words, not the planner.
for name in focus_cols - accepted:
if _quantified(name, raw_message):
notes.append(f"quantified {name}: ALL")
_upsert(effective, ScopeEntry(column=canon_col(name), values=None))
accepted.add(name)
# d. the planner's declaration
for e in change.set if change is not None else []:
name = e.column.lower()
cols = eligible.get(name)
if not cols or name in accepted:
continue
if e.values is None:
# Widening a FOCUS column leaves the focus: the user must have named
# it or said "every"; gate 1 saw section widened on "kalo IPR 2025?".
if (
name in focus_cols
and not _column_named(e.column, raw_message)
and not _has_quantifier(raw_message)
):
notes.append(f"set {e.column}=ALL: not asked for in the message")
continue
_upsert(effective, ScopeEntry(column=canon_col(e.column), values=None))
accepted.add(name)
continue
if name in focus_cols and name in indexed:
continue # listed focus columns come from (c) only
if not e.values or not all(grounded(v, raw_message) for v in e.values):
notes.append(f"set {e.column}: values not in message {e.values}")
continue
if any(_misplaced(v, name, value_index) for v in e.values):
notes.append(f"set {e.column}: values belong to another column {e.values}")
continue
canon = _canonical_values(cols)
_upsert(effective, ScopeEntry(column=canon_col(e.column),
values=[_canon(v, canon) for v in e.values]))
accepted.add(name)
# e. the router's carry. Only the router sees the chat history; it restates
# the previous scope in a short follow-up ("terus komponennya?" → "… di site
# IPR …") and, on the real sessions, never restated a switched site in the
# complete question after it — so go-home holds unless the rewrite carries.
base = "focus"
carried: set[str] = set()
for e in previous or []:
name = e.column.lower()
if not e.values or name in negated:
continue
if not all(grounded(v, rewritten) for v in e.values):
continue
current = next((x for x in effective if x.column.lower() == name), None)
if name in accepted:
# "bandingkan dengan SDJ" after IPR: a plan grouped by the column
# compares the user's value WITH the previous one (review: the carry
# was blocked and IPR vanished from the comparison).
if name in scan.grouped and current is not None and current.values:
merged = list(current.values)
merged += [v for v in e.values if _norm(v) not in {_norm(m) for m in merged}]
if len(merged) > len(current.values):
notes.append(f"compared {e.column}: {merged}")
_upsert(effective, ScopeEntry(column=e.column, values=merged))
carried.add(name)
base = "previous"
continue
if current is None or current.values != e.values:
notes.append(f"carried {e.column}: {e.values} (restated by the rewrite)")
_upsert(effective, e.model_copy(deep=True))
carried.add(name)
base = "previous"
# f. the planner's filters on typed values
adopted: set[str] = set()
for name, values in scan.eq.items():
if name in accepted or name not in eligible:
continue
if not values or not all(grounded(v, raw_message) for v in values):
continue
# A typed value on a column whose list does not hold it but ANOTHER list
# does is misplaced (gate 2: IPR on `section`) — not adopted. A value in
# no list stays: an unknown code ("KMB") must be answered as "not found".
if name in indexed and any(_misplaced(v, name, value_index) for v in values):
continue
canon = _canonical_values(eligible.get(name, []))
entry = ScopeEntry(column=canon_col(name), values=[_canon(v, canon) for v in values])
notes.append(f"adopted {entry.column}: {entry.values}")
_upsert(effective, entry)
adopted.add(name)
# g. a breakdown the user asked for ("per section") is ALL for that column —
# only when the user named the column (plans also group by descriptive
# columns) and the plan does not filter it (a comparison grouped by sides).
# A filter whose values neither the user nor the rewrite gave (the planner
# copying the objective's SDJ onto a per-site question) does not make a
# breakdown a comparison.
said = f"{raw_message}\n{rewritten}"
intended = {
n for n, vals in scan.eq.items() if vals and all(grounded(v, said) for v in vals)
} | set(scan.neg)
for name in scan.grouped - intended - accepted:
if not _column_named(name, raw_message):
continue
if any(e.column.lower() == name and e.values is not None for e in effective):
notes.append(f"grouped {name}: ALL")
_upsert(effective, ScopeEntry(column=canon_col(name), values=None))
# h. an entity (one unit) sets the other dimensions by its own attributes: a
# unit outside the focus answered "not found" three times over (review).
entities = _entity_columns(catalog) - focus_cols
if any(
e.column.lower() in entities and e.values
and e.column.lower() in (accepted | carried | adopted)
for e in effective
):
lifted = [
e.column for e in effective
if e.column.lower() in focus_cols and e.column.lower() not in accepted | carried
]
if lifted:
notes.append(f"entity named: focus lifted for {lifted}")
effective = [e for e in effective if e.column not in lifted]
return focus, effective, base, notes
except Exception as e:
logger.warning(
"scope_resolve_failed",
extra={"error": repr(e), "degraded_seam": "scope_resolve"},
)
try:
fallback = [x.model_copy(deep=True) for x in pinned_focus or []]
except Exception:
fallback = []
return fallback, [x.model_copy(deep=True) for x in fallback], "focus", [
f"resolve_scope failed: {e!r}"
]
# --------------------------------------------------------------------------- #
# 2. Apply
# --------------------------------------------------------------------------- #
def _find_source_table(catalog: Catalog, ir: QueryIR) -> tuple[Source, Table] | None:
for s in catalog.sources:
if s.source_id != ir.source_id:
continue
for t in s.tables:
if t.table_id == ir.table_id:
return s, t
return None
def _lookup_path(
name: str, in_play: list[Table], tables_by_id: dict[str, Table]
) -> list[tuple[Table, Any, Column]]:
"""(target table, fk, column) for each FK from a table in play to a table holding
an eligible column called `name`."""
out: list[tuple[Table, Any, Column]] = []
for t in in_play:
for fk in t.foreign_keys:
target = tables_by_id.get(fk.target_table_id)
if target is None:
continue
col = next(
(c for c in target.columns if c.name.lower() == name and _eligible(c)), None
)
if col is not None:
out.append((target, fk, col))
return out
def _strip_broken_id_filters(
ir: QueryIR, owner: Table, in_play: list[Table], column: str, table_name: str
) -> bool:
"""Drop the planner's own handling of `owner`'s dimension through an id column —
a typed id, a string on the int key, a made-up `${scope.site_id}` (gate run 2: 6 of
31 turns refused on these). A real handoff from a lookup task (`${t1.site_id}`)
stays. True when something was dropped."""
id_cols = {
fk.column_id
for t in in_play
if t is not owner
for fk in t.foreign_keys
if fk.target_table_id == owner.table_id
}
kept = [f for f in ir.filters if f.column_id not in id_cols or _is_handoff(f.value)]
if len(kept) == len(ir.filters):
return False
logger.info("scope_id_filter_replaced", extra={"column": column, "table": table_name})
ir.filters = kept
return True
def _apply_to_ir(
ir: QueryIR,
source: Source,
base_table: Table,
effective: list[ScopeEntry],
not_applied: list[str],
respected: set[str] | None = None,
grounded_text: str = "",
sources_of: dict[str, set[str]] | None = None,
) -> bool:
"""Apply every concrete entry to one IR in place; True when it changed.
Code fills gaps, corrects what is invalid or ungrounded, and respects what is
grounded (2026-10-29, after gate run 2):
- a filter on the column whose values are in the scope or in the user's / the
rewrite's words is the planner's deliberate choice — kept;
- any other filter on it (an invented value) is replaced by the scope;
- no filter: the scope is added — unless `respected` says another query on the
same table filters this column, i.e. the planner built a subset-vs-total
structure ("IPR's share of all sites", "COAL LOADER vs every section"), where
adding it would compare the subject with itself;
- a column the query cannot reach (no column, no single FK) is noted, and so is a
same-named column in a source whose value list was never read (a file's own
`section` is not scoped by the database's values — review 2026-10-29).
"""
tables_by_id = {t.table_id: t for t in source.tables}
in_play: list[Table] = [base_table]
for j in ir.joins:
t = tables_by_id.get(j.target_table_id)
if t is not None and t not in in_play:
in_play.append(t)
changed = False
for entry in effective:
if entry.values is None:
continue
name = entry.column.lower()
values = list(entry.values)
known_in = (sources_of or {}).get(name)
if known_in and source.source_id not in known_in:
not_applied.append(f"{entry.column}: {base_table.name}")
continue
target_col = next(
(c for t in in_play for c in t.columns if c.name.lower() == name and _eligible(c)),
None,
)
candidates = [] if target_col is not None else _lookup_path(name, in_play, tables_by_id)
if target_col is not None:
owner = next((t for t in in_play if target_col in t.columns), None)
else:
owner = candidates[0][0] if len(candidates) == 1 else None
if owner is not None and _strip_broken_id_filters(
ir, owner, in_play, entry.column, base_table.name
):
changed = True
existing = (
[f for f in ir.filters if f.column_id == target_col.column_id]
if target_col is not None
else []
)
if existing:
wanted = {_norm(v) for v in values}
if all(
f.op in _EQ_OPS + _NEG_OPS
and all(
_norm(v) in wanted or grounded(v, grounded_text)
for v in _filter_values(f.value)
)
for f in existing
):
# The planner's grounded choice: a comparison side, a switch, or an
# exclusion the user asked for ("selain SDJ" — review: replacing a
# grounded `not_in [SDJ]` with `in [SDJ]` inverted the question).
continue
logger.info(
"scope_filter_conflict_replaced",
extra={
"column": entry.column,
"table": base_table.name,
"replaced": [f.model_dump() for f in existing],
"values": values,
},
)
ir.filters = [f for f in ir.filters if f.column_id != target_col.column_id]
elif respected and name in respected:
continue # subset-vs-total: another query on this table carries the filter
if target_col is None:
# The column lives in a lookup table: reach it through the ONE foreign key
# that leads there, or report it — never pick between two paths.
if len(candidates) != 1 or source.source_type != "schema":
not_applied.append(f"{entry.column}: {base_table.name}")
continue
target, fk, target_col = candidates[0]
join = Join(
target_table_id=fk.target_table_id,
left_column_id=fk.column_id,
right_column_id=fk.target_column_id,
type="inner",
)
if join not in ir.joins:
ir.joins.append(join)
if target not in in_play:
in_play.append(target)
# No group-by exemption: plans group by descriptive columns too (unit + model
# + section on a ranking); a breakdown the user asked for is ALL by now.
ir.filters.append(
FilterClause(
column_id=target_col.column_id, op="in", value=values, value_type="string"
)
)
changed = True
return changed
def _drop_misplaced_filters(
ir: QueryIR, catalog: Catalog, value_index: dict[str, list[str]] | None
) -> bool:
"""Remove a filter whose values belong to ANOTHER listed column (gate run 2: the
planner put the site code IPR on `section`). The right column gets the value from
the user's words. True when something was removed."""
if not value_index:
return False
cols_by_id = {c.column_id: c for s in catalog.sources for t in s.tables for c in t.columns}
kept = []
for f in ir.filters:
col = cols_by_id.get(f.column_id)
if (
col is not None
and _eligible(col)
and f.op in _EQ_OPS
and any(_misplaced(v, col.name, value_index) for v in _filter_values(f.value))
):
logger.info("scope_misplaced_filter_dropped", extra={"column": col.name})
continue
kept.append(f)
if len(kept) == len(ir.filters):
return False
ir.filters = kept
return True
def _filtered_columns(ir: QueryIR, catalog: Catalog) -> set[str]:
cols_by_id = {c.column_id: c for s in catalog.sources for t in s.tables for c in t.columns}
return {
cols_by_id[f.column_id].name.lower()
for f in ir.filters
if f.column_id in cols_by_id and _eligible(cols_by_id[f.column_id]) and f.op in _EQ_OPS
}
def _open_in_ir(
ir: QueryIR, source: Source, base_table: Table, opened: list[ScopeEntry],
grounded_text: str,
) -> bool:
"""For a column the scope opened (ALL), drop the planner's `=`/`in` filters on
values neither the user nor the rewrite gave — the objective's SDJ copied onto
"MTTR … di semua site" (test 3, 2026-10-30) — and broken id filters on its
lookup path. Grounded filters and exclusions stay. True when something changed."""
tables_by_id = {t.table_id: t for t in source.tables}
in_play: list[Table] = [base_table]
for j in ir.joins:
t = tables_by_id.get(j.target_table_id)
if t is not None and t not in in_play:
in_play.append(t)
changed = False
for entry in opened:
name = entry.column.lower()
ids = {c.column_id for t in in_play for c in t.columns
if c.name.lower() == name and _eligible(c)}
kept = [
f for f in ir.filters
if f.column_id not in ids
or f.op not in _EQ_OPS
or all(grounded(v, grounded_text) for v in _filter_values(f.value))
]
if len(kept) != len(ir.filters):
logger.info("scope_ungrounded_filter_opened", extra={"column": entry.column})
ir.filters = kept
changed = True
candidates = _lookup_path(name, in_play, tables_by_id)
if len(candidates) == 1 and _strip_broken_id_filters(
ir, candidates[0][0], in_play, entry.column, base_table.name
):
changed = True
return changed
def _widen_comparisons(
parsed: list[Any], answering: list[int], concrete: list[ScopeEntry],
catalog: Catalog, grounded_text: str,
) -> set[int]:
"""The user typed several values of a column ("SDJ dibanding IPR") but the
answering queries together cover fewer — the planner fetched one side only (test
3: an A/B step treating IPR as a section over SDJ-only data; run 2: a lookup of
SDJ alone). Every answering filter on that column is widened to all the values,
so the data holds both sides. Returns the indexes of the IRs changed."""
cols_by_id = {c.column_id: c for s in catalog.sources for t in s.tables for c in t.columns}
changed: set[int] = set()
for entry in concrete:
if len(entry.values or []) < 2 or not all(
grounded(v, grounded_text) for v in entry.values
):
continue
name = entry.column.lower()
want = {_norm(v) for v in entry.values}
covered: set[str] = set()
for i in answering:
for f in parsed[i][2].filters:
col = cols_by_id.get(f.column_id)
if col is not None and col.name.lower() == name and f.op in _EQ_OPS:
covered |= {_norm(v) for v in _filter_values(f.value)}
if not covered or covered >= want:
continue
for i in answering:
ir = parsed[i][2]
hit = False
for f in ir.filters:
col = cols_by_id.get(f.column_id)
if col is not None and col.name.lower() == name and f.op in _EQ_OPS:
f.op, f.value = "in", list(entry.values)
hit = True
if hit:
logger.info("scope_comparison_widened", extra={"column": entry.column})
changed.add(i)
return changed
def _lookup_task_ids(task_list: Any) -> set[str]:
"""Tasks whose output another task reads through a `${tN…}` handoff — lookups
(a site code → its id), not queries that answer the question."""
out: set[str] = set()
for task in getattr(task_list, "tasks", None) or []:
for call in getattr(task, "tool_calls", None) or []:
out.update(_TASK_REF_RE.findall(repr(getattr(call, "args", None))))
return out
def apply_scope_to_plan(
task_list: TaskList,
effective: list[ScopeEntry],
catalog: Catalog,
*,
value_index: dict[str, list[str]] | None = None,
grounded_text: str = "",
partial: list[str] | None = None,
) -> list[str]:
"""Apply the effective scope to every `retrieve_data` IR, in place.
Three passes: drop filters whose values belong to another listed column; note
which columns each table's queries filter (a subset-vs-total structure); then
apply each entry per `_apply_to_ir`. `grounded_text` is the user's words plus
the router's rewrite — a planner filter on a value found there is kept.
Coverage is judged over the queries that ANSWER the question (a lookup feeding
a handoff does not): an entry none of them filters is returned as a "not
applied" note (`"<column>: <table>"`); one only some of them filter is appended
to `partial` — the line says so instead of claiming the whole answer (review
2026-10-29: "Cakupan: … site SDJ" over a second number covering every site).
An IR that does not parse, or that fails mid-way, is left exactly as the planner
wrote it. Never raises.
"""
notes: list[str] = []
try:
concrete = [e for e in effective or [] if e.values is not None]
opened = [e for e in effective or [] if e.values is None]
lookups = _lookup_task_ids(task_list)
sources_of = getattr(value_index, "sources", None)
parsed: list[tuple[str, Any, QueryIR, Source, Table, bool]] = []
for task in getattr(task_list, "tasks", None) or []:
for call in getattr(task, "tool_calls", None) or []:
if getattr(call, "tool", None) != "retrieve_data":
continue
args = getattr(call, "args", None)
if not isinstance(args, dict):
continue
try:
ir = QueryIR.model_validate(args.get("ir"))
except Exception as e:
logger.info("scope_ir_unparseable", extra={"error": repr(e)})
continue
found = _find_source_table(catalog, ir)
if found is None:
continue # the validator reports an unknown table
changed = _drop_misplaced_filters(ir, catalog, value_index)
parsed.append((str(task.id), args, ir, found[0], found[1], changed))
if not parsed:
return [] # nothing to apply to, so nothing to report as missed
filtered = [_filtered_columns(ir, catalog) for _, _, ir, _, _, _ in parsed]
local: list[list[str]] = []
for i, (_tid, args, ir, source, base_table, changed) in enumerate(parsed):
local_notes: list[str] = []
try:
respected: set[str] = set()
for j, other in enumerate(parsed):
if j != i and other[4].table_id == base_table.table_id:
respected |= filtered[j]
if concrete and _apply_to_ir(
ir, source, base_table, concrete, local_notes,
respected=respected - filtered[i], grounded_text=grounded_text,
sources_of=sources_of,
):
changed = True
if opened and _open_in_ir(ir, source, base_table, opened, grounded_text):
changed = True
if changed:
args["ir"] = ir.model_dump()
except Exception as e:
logger.warning(
"scope_apply_ir_failed",
extra={
"error": repr(e), "table": base_table.name,
"degraded_seam": "scope_apply",
},
)
local_notes = [f"{x.column}: {base_table.name}" for x in concrete]
local.append(local_notes)
answering = [i for i, p in enumerate(parsed) if p[0] not in lookups] or list(
range(len(parsed))
)
for i in _widen_comparisons(parsed, answering, concrete, catalog, grounded_text):
parsed[i][1]["ir"] = parsed[i][2].model_dump()
after = [_filtered_columns(p[2], catalog) | _negated_columns(p[2], catalog) for p in parsed]
for entry in concrete:
name = entry.column.lower()
took = [i for i in answering if name in after[i]]
if not took:
note = next(
(n for i in answering for n in local[i] if n.lower().startswith(name + ":")),
entry.column,
)
notes.append(note)
elif len(took) < len(answering) and partial is not None:
partial.append(entry.column)
return _dedupe(notes)
except Exception as e:
logger.warning(
"scope_apply_failed", extra={"error": repr(e), "degraded_seam": "scope_apply"}
)
return _dedupe(notes)
def _negated_columns(ir: QueryIR, catalog: Catalog) -> set[str]:
cols_by_id = {c.column_id: c for s in catalog.sources for t in s.tables for c in t.columns}
return {
cols_by_id[f.column_id].name.lower()
for f in ir.filters
if f.column_id in cols_by_id and _eligible(cols_by_id[f.column_id]) and f.op in _NEG_OPS
}
# --------------------------------------------------------------------------- #
# 3. Render
# --------------------------------------------------------------------------- #
def _entry_map(entries: list[ScopeEntry]) -> dict[str, frozenset[str] | None]:
return {
e.column.lower(): (None if e.values is None else frozenset(_norm(v) for v in e.values))
for e in entries
}
def focus_changed(
focus: list[ScopeEntry],
effective: list[ScopeEntry],
lifted: set[str] | frozenset[str] = frozenset(),
) -> bool:
"""True when the effective scope covers a different slice than the focus.
Compared per FOCUS column as value sets (case-insensitive): a focus column
missing from `effective`, widened to ALL, or set to other values is different.
A column the focus does not name is NOT: one unit or model inside the focus is
a narrower look at the focus, not a detour away from it (lead, 2026-10-29), so
neither the chat line nor the report labels a drill-down as outside the focus.
`lifted`: focus columns a named unit set aside for the turn (it carries its own
site and section, usually inside the focus) — not counted as leaving it.
Never raises (False on garbage).
"""
try:
f = _entry_map(focus or [])
e = _entry_map(effective or [])
skip = {c.lower() for c in lifted}
return any(
col not in e or e[col] != vals for col, vals in f.items() if col not in skip
)
except Exception as ex:
logger.warning("scope_focus_changed_failed", extra={"error": repr(ex)})
return False
def _render_entry(entry: ScopeEntry, id_: bool) -> str:
if entry.values is None:
return f"{entry.column}: {'semua' if id_ else 'all'}"
return f"{entry.column} {', '.join(entry.values)}"
def _render_filter(f: dict[str, Any]) -> str:
col, op, value = f.get("column"), f.get("op"), f.get("value")
if op == "=":
return f"{col} {value}"
if op == "in" and isinstance(value, list | tuple):
return f"{col} {', '.join(str(v) for v in value)}"
if op == "between" and isinstance(value, list | tuple) and len(value) == 2:
return f"{col} {value[0]}–{value[1]}"
if value is None:
return f"{col} {op}"
return f"{col} {op} {value}"
def _as_int(v: Any) -> int | None:
if v is None or isinstance(v, bool):
return None
try:
if isinstance(v, float):
return int(v) if v.is_integer() else None
return int(str(v).strip())
except (TypeError, ValueError):
return None
def _meta_filters(out: Any) -> list[dict[str, Any]]:
meta = getattr(out, "meta", None) or {}
filters = meta.get("filters") if isinstance(meta, dict) else None
return [f for f in filters or [] if isinstance(f, dict)]
def _coverage(outputs: list[Any]) -> str | None:
periods: list[tuple[int, int]] = []
for out in outputs:
cols = [str(c).lower() for c in getattr(out, "columns", None) or []]
if "year" not in cols or "month" not in cols:
continue
yi, mi = cols.index("year"), cols.index("month")
for row in getattr(out, "rows", None) or []:
if not isinstance(row, list | tuple) or len(row) <= max(yi, mi):
continue
y, m = _as_int(row[yi]), _as_int(row[mi])
if y is not None and m is not None:
periods.append((y, m))
if not periods:
return None
lo, hi = min(periods), max(periods)
if lo == hi:
return f"data {lo[0]:04d}-{lo[1]:02d}"
return f"data {lo[0]:04d}-{lo[1]:02d}–{hi[0]:04d}-{hi[1]:02d}"
def render_scope_line(
scope: TurnScope, run_state: RunState, reply_language: str | None
) -> TurnScope:
"""Fill `line`, `not_applied` and `unknown_values` from what actually RAN.
Every part of the line is read from the executed `retrieve_data` outputs'
`meta.filters`, so an entry the queries never filtered is marked "not applied"
instead of being claimed. A lookup filtered only on scope values that returned
no rows means the user named a code that does not exist. Indonesian when the
reply language says so, else English. Never raises: on failure the scope comes
back unchanged.
"""
try:
id_ = "indonesia" in (reply_language or "").lower()
outputs: list[Any] = []
for result in (getattr(run_state, "results", None) or {}).values():
for out in getattr(result, "outputs", None) or []:
if getattr(out, "tool", None) == "retrieve_data" and getattr(
out, "kind", None
) == "table":
outputs.append(out)
effective = list(scope.effective)
eff_cols = {e.column.lower() for e in effective}
filtered_cols = {
str(f.get("column", "")).lower() for out in outputs for f in _meta_filters(out)
}
# The values that actually RAN, per column, not the intended ones: the label
# must never claim more than the data (replay of gate run 2: a comparison
# whose second side the plan never fetched would otherwise read "SDJ, IPR").
ran: dict[str, list[str]] = {}
for out in outputs:
for f in _meta_filters(out):
col = str(f.get("column", "")).lower()
if col in eff_cols and f.get("op") in _EQ_OPS:
bucket = ran.setdefault(col, [])
for v in _filter_values(f.get("value")):
if _norm(v) not in {_norm(b) for b in bucket}:
bucket.append(v)
excluded: dict[str, list[str]] = {}
for out in outputs:
for f in _meta_filters(out):
col = str(f.get("column", "")).lower()
if col in eff_cols and f.get("op") in _NEG_OPS:
excluded.setdefault(col, []).extend(_filter_values(f.get("value")))
parts: list[str] = []
not_applied = list(scope.not_applied)
partial = {p.lower() for p in scope.partial}
for entry in effective:
shown = entry
if entry.values is not None and ran.get(entry.column.lower()):
shown = ScopeEntry(column=entry.column, values=ran[entry.column.lower()])
text = _render_entry(shown, id_)
if excluded.get(entry.column.lower()):
ex = ", ".join(_dedupe(excluded[entry.column.lower()]))
text += f" kecuali {ex}" if id_ else f" except {ex}"
if entry.column.lower() in partial:
text += " (sebagian kueri)" if id_ else " (some queries)"
if entry.values is not None and entry.column.lower() not in filtered_cols:
text += " (tidak diterapkan)" if id_ else " (not applied)"
if not any(n.lower().startswith(entry.column.lower() + ":") for n in not_applied):
not_applied.append(entry.column)
parts.append(text)
others: list[str] = []
for out in outputs:
for f in _meta_filters(out):
if str(f.get("column", "")).lower() in eff_cols:
continue
rendered = _render_filter(f)
if rendered not in others:
others.append(rendered)
parts.extend(others[:_MAX_OTHER_FILTERS])
coverage = _coverage(outputs)
if coverage:
parts.append(coverage)
if scope.focus and focus_changed(scope.focus, effective, set(scope.lifted)):
rendered_focus = "; ".join(_render_entry(e, id_) for e in scope.focus)
parts.append(
f"di luar fokus ({rendered_focus})" if id_ else f"outside focus ({rendered_focus})"
)
unknown: list[str] = []
for out in outputs:
meta = getattr(out, "meta", None) or {}
filters = _meta_filters(out)
if (
meta.get("row_count") == 0
and filters
and all(str(f.get("column", "")).lower() in eff_cols for f in filters)
):
for f in filters:
unknown.extend(_filter_values(f.get("value")))
# Only a value the user TYPED can be "not found": an empty slice of the focus
# is "no data", never "code SDJ not found" (review: a unit outside the focus
# reported the focus's own codes as missing).
typed = {_norm(t) for t in scope.typed}
unknown = [u for u in _dedupe(unknown) if _norm(u) in typed]
not_applied_word = "tidak diterapkan" if id_ else "not applied"
notes = [f"{n} ({not_applied_word})" for n in _dedupe(scope.not_applied)]
if unknown:
notes.extend(
(f"kode {u} tidak ditemukan" if id_ else f"code {u} not found") for u in unknown
)
elif outputs and all(
(getattr(o, "meta", None) or {}).get("row_count") == 0 for o in outputs
):
notes.append("tidak ada data untuk cakupan ini" if id_ else "no rows for this scope")
lines: list[str] = []
if not effective and not outputs:
parts, notes = [], []
if parts:
lines.append(f"_{'Cakupan' if id_ else 'Scope'}: " + " · ".join(parts) + "_")
if notes:
lines.append(f"_{'Catatan' if id_ else 'Note'}: " + "; ".join(notes) + "_")
return scope.model_copy(
update={
"line": "\n".join(lines),
"not_applied": _dedupe(not_applied),
"unknown_values": unknown,
},
deep=True,
)
except Exception as e:
logger.warning(
"scope_render_failed", extra={"error": repr(e), "degraded_seam": "scope_render"}
)
return scope