Blame for src/agents/scope/engine.py

"""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")
_PERIOD_COLUMNS = {"year", "month", "week", "quarter", "date", "day", "period"}
_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")
# A VALUE handoff `${tN.col}` marks tN as a lookup. A Pattern-A data feed `${tN}` (the
# rows an analyze_* tool reads) does not: counting it made the answering query a
# "lookup", the line fell back to the site lookup and read "(sebagian kueri)" (P0-4).
_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."""
    fks = [
        fk for t in in_play if t is not owner
        for fk in t.foreign_keys if fk.target_table_id == owner.table_id
    ]
    # Both ends of the key: the measure table's `site_id` AND the lookup table's own
    # `site_id` — a typed `vw_dim_site.site_id = 2009` is the same guess (P0-4; Check 8j
    # rejected it, costing a retry, because only the FK side was stripped here).
    id_cols = {fk.column_id for fk in fks} | {fk.target_column_id for fk in fks}
    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 _stem(column: str) -> str:
    """`site_id`, `site_code`, `site_name` → `site`: one dimension, several columns."""
    return re.sub(r"_(id|code|kode|no|name|nama)$", "", column.lower())


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)

        # Period filters first and outside the cap: the line must always say which
        # period ran (D1a, 2026-11-03), however many other filters a query carries.
        period_parts: list[str] = []
        others: list[str] = []
        eff_stems = {_stem(c) for c in eff_cols}
        for out in outputs:
            for f in _meta_filters(out):
                col = str(f.get("column", "")).lower()
                # `site_id 2009` is the resolved id of `site_code IPR`, already shown (P0-4).
                # Only `*_id` columns: a `unit_no` next to `unit_name` may disagree.
                if col in eff_cols or (col.endswith("_id") and _stem(col) in eff_stems):
                    continue
                rendered = _render_filter(f)
                bucket = period_parts if col in _PERIOD_COLUMNS else others
                if rendered not in bucket:
                    bucket.append(rendered)
        parts.extend(period_parts)
        parts.extend(others[:_MAX_OTHER_FILTERS])

        coverage = _coverage(outputs)
        if coverage:
            parts.append(coverage)
        # D1a (2026-11-03): the line always says which period was used — a query with no
        # period filter read every year, and a silent line read as "the objective's year".
        if scope.periodic and not coverage and outputs and not any(
            str(f.get("column", "")).lower() in _PERIOD_COLUMNS
            for out in outputs for f in _meta_filters(out)
        ):
            parts.append("semua periode" if id_ else "all periods")

        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