[NOTICKET] fix(assembler): a top-N list's ends are not the overall highest/lowest

- 9 Nov re-test (Rifqi): for "10 lowest-PA units" the order line said "highest: <unit>", and the answer called it the highest PA overall (83.743% vs a real 99.479%)
- retrieve_data meta now carries `cut` (limit, first sort column, direction)
- The order line marks a cut list ("10 rows shown — a cut list: the 10 lowest by avg_pa, not all rows") and names its ends "highest/lowest of these 10"; a limit not reached keeps the old wording
- assembler.md: a cut list's ends are never the overall highest/lowest
- Local chat_sim (5 turns, demo xlsx): 5/5 right, including "the highest in that list" and the MTBS site comparison (ADT 1st of 5)

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

src/agents/slow_path/prompt.py

@@ -133,9 +133,14 @@ def _order_lines(output: ToolOutput) -> list[str]:
     - exactly one text label column, not a time column (a trend keeps the stub-period
       rules; a two-label table would rank across groups);
     - numeric columns only, not time/id-like, no missing values, not all equal.
+
+    A cut list ("the 10 lowest": `limit` reached, or the executor truncated) says so
+    and names its ends "of these N" — on 9 Nov the top row of the 10 lowest-PA units
+    was reported as the highest PA overall.
     """
     try:
         columns, rows = output.columns or [], output.rows or []
+        scope, top, bottom = _cut_words(output, len(rows))
         if output.tool != "retrieve_data" or not 2 <= len(rows) <= _MAX_ROWS:
             return []
         if any(len(row) != len(columns) for row in rows):
@@ -164,15 +169,36 @@ def _order_lines(output: ToolOutput) -> list[str]:
                 zip(cols[label], values, strict=True), key=lambda pair: pair[1], reverse=True
             )
             lines.append(
-                f"order by {columns[i]} ({len(rows)} rows, highest first): "
-                f"{_ranked_text(ranked)} · highest: {_ends(ranked, ranked[0][1])}"
-                f" · lowest: {_ends(ranked, ranked[-1][1])}"
+                f"order by {columns[i]} ({scope}, highest first): "
+                f"{_ranked_text(ranked)} · {top}: {_ends(ranked, ranked[0][1])}"
+                f" · {bottom}: {_ends(ranked, ranked[-1][1])}"
             )
         return lines
     except Exception:  # noqa: BLE001 — a render detail must never cost the answer
         return []
 
 
+def _cut_words(output: ToolOutput, n: int) -> tuple[str, str, str]:
+    """(row-count text, top label, bottom label) — "of these N" when the list is cut.
+
+    Cut = the executor truncated, or the query's `limit` was reached (fewer rows than
+    the limit means every row came back). Without a known cut the words are as before.
+    """
+    meta = output.meta or {}
+    cut = meta.get("cut") if isinstance(meta.get("cut"), dict) else {}
+    limit = cut.get("limit")
+    is_cut = bool(meta.get("truncated")) or (isinstance(limit, int) and n >= limit)
+    if not is_cut:
+        return f"{n} rows", "highest", "lowest"
+    end = {"asc": "lowest", "desc": "highest"}.get(cut.get("dir") or "")
+    kept = f"the {n} {end} by {cut['column']}" if end and cut.get("column") else "a part"
+    return (
+        f"{n} rows shown — a cut list: {kept}, not all rows",
+        f"highest of these {n}",
+        f"lowest of these {n}",
+    )
+
+
 def _is_number(value: object) -> bool:
     if isinstance(value, bool) or not isinstance(value, numbers.Real):
         return False

src/config/prompts/assembler.md

@@ -97,6 +97,9 @@ When no such section is present, write as you otherwise would.
   every count ("of the 5 sites") comes from that line, never from the row position in
   the table, which may be unsorted. No sentence may contradict it. If the user asks
   lowest first, reverse that order; do not re-rank.
+  When the line says `a cut list` (e.g. "the 10 lowest by pa, not all rows"), the
+  table is only part of the data: its ends are "highest/lowest **of these** N", never
+  the highest or lowest overall — say "the 10 lowest", not "the highest is …".
   Example (made-up names): focus Site B; rows arrive as C 71.0, A 92.1, B 90.4 with
   `order by avg_uptime (3 rows, highest first): A 92.1 > B 90.4 > C 71.0`.
   Right: "Site B, the focus, is 2nd of 3 sites: below A, above C." Wrong: "B is the

src/tools/data_access.py

@@ -300,6 +300,10 @@ class DataAccessToolInvoker:
                 # How each averaged measure was computed, for the scope line
                 # (2026-11-09, RATIO_MEASURES_REQUIREMENTS M5).
                 "methods": _method_summary(catalog, ir),
+                # The row cut and the sort it kept, for the Assembler's order line
+                # (2026-11-09): "the 10 lowest" must not have its top row called
+                # the overall highest.
+                "cut": _cut_summary(catalog, ir),
             },
         )
 
@@ -420,6 +424,36 @@ def _method_summary(catalog: Catalog, ir: QueryIR) -> list[dict[str, Any]]:
         return []
 
 
+def _cut_summary(catalog: Catalog, ir: QueryIR) -> dict[str, Any] | None:
+    """`{"limit", "column", "dir"}` when the query kept only the first N rows, else None.
+
+    `column` / `dir` are the first sort key as it appears in the result (a select
+    alias, else the catalog column name), or None without a sort. Never raises.
+    """
+    try:
+        if ir.limit is None:
+            return None
+        column: str | None = None
+        direction: str | None = None
+        if ir.order_by:
+            key = ir.order_by[0]
+            direction = key.dir
+            aliases = {
+                getattr(item, "alias", None) for item in ir.select
+            } - {None}
+            if key.column_id in aliases:
+                column = key.column_id
+            else:
+                names = {
+                    c.column_id: c.name
+                    for s in catalog.sources for t in s.tables for c in t.columns
+                }
+                column = names.get(key.column_id, key.column_id)
+        return {"limit": ir.limit, "column": column, "dir": direction}
+    except Exception:
+        return None
+
+
 def _pii_column_names(
     catalog: Catalog, ir: QueryIR, output_columns: list[str]
 ) -> list[str]: