Rifqi Hafizuddin
Claude Opus 5.5
[NOTICKET] feat(scope): code-owned conversation scope — focus from value lists, follow-up carry, scope line
111ef6b | """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 | |