[NOTICKET] check: schema-detail view for column/type questions

"kolom apa aja + tipe datanya?" now resolves straight to the data — no
filename required — instead of a bare source listing. A schema cue
(kolom/tipe/struktur/field) drills into the named source(s), or ALL structured
sources when none is named (documents excluded — no columns).

The rendered schema is humanized, ChatGPT-style and fully deterministic:
- lean per-table columns (Column · Data type); friendly type words
(teks/bilangan bulat/desimal); row count moved to the header line.
- adaptive columns: "Boleh kosong" only when a table mixes nullable values,
"PII" only when a column is flagged.
- per-source Ringkasan bucketing columns (Categorical / Numeric integer /
Numeric decimal / Date) — only buckets that exist; int vs float split.
- heuristic notes: a date-named string column → "convert to datetime"; a
numeric id-named column → "likely an identifier".
- big-database guard: a DB with >3 tables is summarised at the table level
(name + column/row counts) instead of dumping every column.

Supersedes the old name-match drill-down (a source referred to as "dataset
itu" now resolves via the single-/all-source fallback).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/agents/handlers/check.py

@@ -14,7 +14,7 @@ from __future__ import annotations
 
 import asyncio
 import re
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from src.tools.contracts import ToolOutput
 
@@ -166,7 +166,7 @@ def render_tool_output(out: ToolOutput, reply_language: str = "English") -> str:
     type_labels = _SOURCE_TYPE_LABELS.get(reply_language, _SOURCE_TYPE_LABELS["English"])
     type_idx = columns.index("source_type") if "source_type" in columns else -1
 
-    def _cell(i: int, row: list) -> str:
+    def _cell(i: int, row: list[Any]) -> str:
         if i == type_idx:
             return type_labels.get(str(row[i]), str(row[i]))
         return str(row[i])
@@ -267,6 +267,250 @@ def _render_helicopter(
     return opener + "\n\n" + "\n\n".join(parts)
 
 
+# ---------------------------------------------------------------------------
+# Schema-detail view (columns + types)
+# ---------------------------------------------------------------------------
+
+# Cues meaning "show me the columns/types", not just "what sources exist". When
+# present, the handler drills into the schema of the named source(s) — or ALL
+# structured sources when none is named — instead of listing. This is what lets
+# "kolom apa aja yang aku miliki dan tipenya?" resolve straight to the data
+# without the user having to type a full filename.
+_SCHEMA_CUES = (
+    "kolom", "column", "tipe data", "data type", "tipe", "dtype",
+    "struktur", "structure", "schema", "skema", "field",
+)
+
+# A database with more tables than this is summarised (table names + counts)
+# rather than dumping every column — otherwise a big DB is a wall of text.
+_DB_TABLE_THRESHOLD = 3
+
+# Friendly, localized data-type words (normalized from the catalog's raw type).
+_TYPE_WORDS = {
+    "English": {"string": "text", "integer": "integer", "float": "decimal",
+                "date": "date", "boolean": "boolean", "other": "other"},
+    "Indonesian": {"string": "teks", "integer": "bilangan bulat", "float": "desimal",
+                   "date": "tanggal", "boolean": "boolean", "other": "lainnya"},
+}
+
+# Localized strings for the schema view + the per-source Ringkasan.
+_SCHEMA_STR = {
+    "English": {
+        "lead": "Here are the columns and data types across your {n} source{s}:",
+        "summary": "Summary:",
+        "cat": "Categorical", "int": "Numeric (integer)",
+        "float": "Numeric (decimal)", "date": "Date",
+        "note_date": "still text, consider converting to datetime",
+        "note_id": "likely an identifier",
+        "rows_word": "rows", "table_word": "Table",
+        "yes": "Yes", "no": "—",
+        "db_tables": "{name} has {n} tables:",
+        "db_item": "- {table} ({cols} columns{rows})",
+        "db_hint": "Name a table to see its columns.",
+    },
+    "Indonesian": {
+        "lead": "Berikut kolom dan tipe data di {n} sumber yang kamu punya:",
+        "summary": "Ringkasan:",
+        "cat": "Kategorikal", "int": "Numerik (bilangan bulat)",
+        "float": "Numerik (desimal)", "date": "Tanggal",
+        "note_date": "masih teks, sebaiknya dikonversi ke datetime",
+        "note_id": "kemungkinan identifier",
+        "rows_word": "baris", "table_word": "Tabel",
+        "yes": "Ya", "no": "—",
+        "db_tables": "{name} punya {n} tabel:",
+        "db_item": "- {table} ({cols} kolom{rows})",
+        "db_hint": "Sebut nama tabelnya untuk lihat kolomnya.",
+    },
+}
+
+
+def _sc(reply_language: str) -> dict[str, str]:
+    """Localized schema/summary string bundle; defaults to English."""
+    return _SCHEMA_STR.get(reply_language, _SCHEMA_STR["English"])
+
+
+def _wants_schema(message: str) -> bool:
+    """True when the message asks about columns/types (schema detail)."""
+    lowered = message.lower()
+    return any(cue in lowered for cue in _SCHEMA_CUES)
+
+
+def _type_category(raw: str) -> str:
+    """Normalize a catalog `data_type` string into a coarse category."""
+    r = (raw or "").lower()
+    if "bool" in r:
+        return "boolean"
+    if any(k in r for k in ("float", "double", "decimal", "numeric", "real")):
+        return "float"
+    if "int" in r:
+        return "integer"
+    if any(k in r for k in ("date", "time", "timestamp")):
+        return "date"
+    if any(k in r for k in ("char", "text", "string", "object", "varchar", "str")):
+        return "string"
+    return "other"
+
+
+def _looks_date(name: str) -> bool:
+    return any(k in name for k in ("date", "tanggal", "tgl"))
+
+
+def _looks_id(name: str) -> bool:
+    tokens = re.split(r"[^a-z0-9]+", name)
+    return "id" in tokens or "email" in tokens or name.endswith("id")
+
+
+def _bucket_and_note(name: str, data_type: str, reply_language: str) -> tuple[str, str]:
+    """Assign a column to a Ringkasan bucket + an optional heuristic note.
+
+    Grouping is by data type, with two name-based overrides that mirror how a
+    human reads a schema: a date-named column goes to Date even when stored as
+    text (with a 'convert to datetime' note), and a numeric column whose name
+    looks like an id gets an 'identifier' note (shouldn't be treated as a
+    quantity). Deterministic — heuristics only, no LLM.
+    """
+    sc = _sc(reply_language)
+    cat = _type_category(data_type)
+    lname = name.lower()
+    if _looks_date(lname) or cat == "date":
+        return "date", (sc["note_date"] if cat != "date" else "")
+    if cat == "integer":
+        return "int", (sc["note_id"] if _looks_id(lname) else "")
+    if cat == "float":
+        return "float", (sc["note_id"] if _looks_id(lname) else "")
+    return "cat", ""
+
+
+def _schema_rows(out: ToolOutput) -> list[dict[str, Any]]:
+    """Flatten a check_data(source_id) ToolOutput into per-column dicts."""
+    cols = out.columns or []
+    idx = {c: i for i, c in enumerate(cols)}
+
+    def _get(row: list[Any], key: str, default: object = None) -> object:
+        return row[idx[key]] if key in idx else default
+
+    rows: list[dict[str, Any]] = []
+    for r in out.rows or []:
+        rows.append({
+            "table": str(_get(r, "table_name", "")),
+            "table_rows": _get(r, "table_row_count"),
+            "column": str(_get(r, "column_name", "")),
+            "data_type": str(_get(r, "data_type", "")),
+            "nullable": bool(_get(r, "nullable", False)),
+            "pii": bool(_get(r, "pii_flag", False)),
+        })
+    return rows
+
+
+def _columns_table(cols: list[dict[str, Any]], reply_language: str) -> str:
+    """Lean per-table markdown: Column · Data type (+ Nullable if mixed, + PII if any)."""
+    labels = _HEADER_LABELS.get(reply_language, _HEADER_LABELS["English"])
+    types = _TYPE_WORDS.get(reply_language, _TYPE_WORDS["English"])
+    sc = _sc(reply_language)
+    # Nullable only carries signal when a table mixes nullable + non-nullable;
+    # PII only when at least one column is flagged. Otherwise the column is noise.
+    show_nullable = len({c["nullable"] for c in cols}) > 1
+    show_pii = any(c["pii"] for c in cols)
+
+    headers = [labels["column_name"], labels["data_type"]]
+    if show_nullable:
+        headers.append(labels["nullable"])
+    if show_pii:
+        headers.append(labels["pii_flag"])
+    lines = [
+        "| " + " | ".join(headers) + " |",
+        "| " + " | ".join("---" for _ in headers) + " |",
+    ]
+    for c in cols:
+        cells = [c["column"], types.get(_type_category(c["data_type"]), c["data_type"])]
+        if show_nullable:
+            cells.append(sc["yes"] if c["nullable"] else sc["no"])
+        if show_pii:
+            cells.append(sc["yes"] if c["pii"] else sc["no"])
+        lines.append("| " + " | ".join(cells) + " |")
+    return "\n".join(lines)
+
+
+def _render_summary(cols: list[dict[str, Any]], reply_language: str) -> str:
+    """Adaptive per-source Ringkasan: only buckets that actually have columns."""
+    sc = _sc(reply_language)
+    buckets: dict[str, list[str]] = {"cat": [], "date": [], "int": [], "float": []}
+    for c in cols:
+        bucket, note = _bucket_and_note(c["column"], c["data_type"], reply_language)
+        buckets[bucket].append(f"{c['column']} ({note})" if note else c["column"])
+    lines = [sc["summary"]]
+    emitted = False
+    for key in ("cat", "date", "int", "float"):
+        if buckets[key]:
+            emitted = True
+            lines.append(f"- {sc[key]}: " + ", ".join(buckets[key]))
+    return "\n".join(lines) if emitted else ""
+
+
+def _render_schema_source(out: ToolOutput, reply_language: str) -> str:
+    """Render one source's schema: columns per table + a Ringkasan.
+
+    A database with more than `_DB_TABLE_THRESHOLD` tables is summarised at the
+    table level instead (name + column/row counts) so it doesn't become a wall
+    of text; the user names a table to drill into its columns.
+    """
+    if out.kind == "error":
+        return _s(reply_language)["lookup_error"].format(error=out.error)
+    rows = _schema_rows(out)
+    if not rows:
+        return ""
+    meta = out.meta or {}
+    name = str(meta.get("source_name") or "source")
+    source_type = str(meta.get("source_type") or "")
+    sc = _sc(reply_language)
+
+    tables: dict[str, list[dict[str, Any]]] = {}
+    for r in rows:
+        tables.setdefault(r["table"], []).append(r)
+
+    def _rows_suffix(rc: object) -> str:
+        return f", {rc} {sc['rows_word']}" if rc else ""
+
+    if source_type == "schema" and len(tables) > _DB_TABLE_THRESHOLD:
+        lines = [sc["db_tables"].format(name=name, n=len(tables))]
+        for tname, tcols in tables.items():
+            rc = tcols[0]["table_rows"]
+            lines.append(
+                sc["db_item"].format(table=tname, cols=len(tcols), rows=_rows_suffix(rc))
+            )
+        lines.append(sc["db_hint"])
+        return "\n".join(lines)
+
+    parts: list[str] = []
+    if len(tables) == 1:
+        tname, tcols = next(iter(tables.items()))
+        rc = tcols[0]["table_rows"]
+        head = f"**{name}**" + (f" ({rc} {sc['rows_word']})" if rc else "")
+        parts.append(f"{head}\n{_columns_table(tcols, reply_language)}")
+    else:
+        parts.append(f"**{name}**")
+        for tname, tcols in tables.items():
+            rc = tcols[0]["table_rows"]
+            sub = f"{sc['table_word']} {tname}" + (f" ({rc} {sc['rows_word']})" if rc else "") + ":"
+            parts.append(f"{sub}\n{_columns_table(tcols, reply_language)}")
+
+    summary = _render_summary(rows, reply_language)
+    if summary:
+        parts.append(summary)
+    return "\n\n".join(parts)
+
+
+def _render_schemas(schemas: list[ToolOutput], reply_language: str) -> str:
+    """Stitch one or more source schemas under a single lead-in sentence."""
+    blocks = [b for b in (_render_schema_source(o, reply_language) for o in schemas) if b]
+    if not blocks:
+        return ""
+    n = len(blocks)
+    suffix = "s" if (reply_language == "English" and n != 1) else ""
+    lead = _sc(reply_language)["lead"].format(n=n, s=suffix)
+    return lead + "\n\n" + "\n\n".join(blocks)
+
+
 async def run_check(
     message: str, invoker: ToolInvoker, reply_language: str = "English"
 ) -> str:
@@ -287,36 +531,35 @@ async def run_check(
             return _no_match
         return _lead("documents", reply_language, len(out.rows or [])) + "\n\n" + listing
 
-    if intent == "data":
+    # Schema-detail question ("kolom apa aja + tipe datanya"): drill into the
+    # schema of the named source(s), or ALL structured sources when none is
+    # named — so the user never has to type a full filename to see their columns
+    # and types. This supersedes name-matched drill-down (a source referred to as
+    # "dataset itu" now resolves via the single-/all-source fallback).
+    if _wants_schema(message):
         inventory = await invoker.invoke("check_data", {})
         if inventory.kind == "error":
             return render_tool_output(inventory, reply_language)
-        # Drill down to the schema of each source the user named; if they named
-        # none, return the source listing.
-        source_ids = _matched_source_ids(message, inventory)
-        if not source_ids:
-            listing = _render_source_list(inventory, reply_language)
-            if not listing:
-                return _no_match
-            n = len(inventory.rows or [])
-            return _lead("structured", reply_language, n) + "\n\n" + listing
+        if not (inventory.rows or []):
+            return _no_match
+        named = _matched_source_ids(message, inventory)
+        cols = inventory.columns or []
+        sid_i = cols.index("source_id") if "source_id" in cols else 0
+        ids = named or [str(r[sid_i]) for r in (inventory.rows or [])]
         schemas = await asyncio.gather(
-            *(invoker.invoke("check_data", {"source_id": sid}) for sid in source_ids)
+            *(invoker.invoke("check_data", {"source_id": sid}) for sid in ids)
         )
-        if len(schemas) == 1:
-            table = render_tool_output(schemas[0], reply_language)
-            if not table:
-                return _no_match
-            name = (schemas[0].meta or {}).get("source_name") or ""
-            return _lead("schema", reply_language, name=name) + "\n\n" + table
-        # Multiple named sources → one labelled section per source.
-        sections: list[str] = []
-        for out in schemas:
-            table = render_tool_output(out, reply_language)
-            if table:
-                label = (out.meta or {}).get("source_name") or "source"
-                sections.append(f"**{label}**\n{table}")
-        return "\n\n".join(sections) or _no_match
+        return _render_schemas(schemas, reply_language) or _no_match
+
+    if intent == "data":
+        inventory = await invoker.invoke("check_data", {})
+        if inventory.kind == "error":
+            return render_tool_output(inventory, reply_language)
+        listing = _render_source_list(inventory, reply_language)
+        if not listing:
+            return _no_match
+        n = len(inventory.rows or [])
+        return _lead("structured", reply_language, n) + "\n\n" + listing
 
     # broad / ambiguous → helicopter view: call both concurrently
     data_out, knowledge_out = await asyncio.gather(