[NOTICKET] anomaly: name the anomalous rows + a flag-rate quality verdict (D-3)

analyze_anomaly's records carried only the scored numeric columns plus a score, so
"which units look unusual?" could never name a unit; and it was the only ML tool
with no quality block, so an 86%-flagged run reached the assembler as silence.

- Add optional label_columns: identifier columns carried into each record
(identifiers first), never fed to the model. Planner guidance to pass the id
column follows on Rifqi's side (decided shape (a), tool-side).
- Add the reserved quality block: metric anomaly_rate, weak only ABOVE a 0.5
ceiling. A20's flag-rate finding was re-measured, not re-assumed -- clean noise
flags 0.14-0.22 (mean 0.17, p95 0.22), a genuine 30%-outlier frame 0.30, the
demo's degenerate run 0.86 -- so 0.5 clears the noise floor and a real finding
yet still catches "most rows flagged"; the note states the rate plainly either
way. Logic in _flag_rate_quality() so both branches are unit-testable (the fit
can't be pushed past 0.5 via contamination).

PR-0 from docs/plans/PR0_PR1_TOOL_FIXES.md; plan doc marks D-1 + D-3 done.

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

docs/plans/PR0_PR1_TOOL_FIXES.md

@@ -1,6 +1,7 @@
 # PR-0 + PR-1 — the tools do what they claim
 
 🔄 **LIVE PLAN** · opened 2026-09-21 · branch `pr/22` · **owner: Sofhia**
+· **PR-0 (D-1, D-3) done 2026-09-21 — awaiting commit; PR-1 (D-4, D-6, D-7, D-10) not started**
 
 Six fixes, all inside `src/tools/analytics/`. They come from the 2026-09-18 design review
 ([docs/reviews/AGENTIC_DESIGN_REVIEW_2026-09-18.md](../reviews/AGENTIC_DESIGN_REVIEW_2026-09-18.md))
@@ -43,6 +44,12 @@ analytical correctness would mostly rediscover them.
 
 ## D-1 · `analyze_data_quality` advertises duplicate detection and computes none
 
+✅ **Done 2026-09-21.** Chose *compute it*: added `duplicate_rows {count, rate}` over the
+profiled columns (`df[cols].duplicated()`, `keep="first"` = redundant copies; a column
+subset is a candidate-key check) plus `issues.duplicate_row_count`. Always present (0 when
+none, never `None`/absent), so `use_when`'s `duplikat`/`duplicates` claim is now backed.
+3 local fixtures (zero-when-none, planted 3-identical→count 2, subset candidate-key).
+
 **Verified** `src/tools/analytics/data_quality.py:103-104`
 
 ```
@@ -77,6 +84,17 @@ but a narrower honest tool beats a broad dishonest one.
 
 ## D-3 · `analyze_anomaly` cannot name what is anomalous
 
+✅ **Done 2026-09-21.** Shape **(a) `label_columns`** (tool-side, decided with Rifqi):
+optional id columns carried into each record (identifiers first), never fed to the model —
+Rifqi updates planner guidance to pass the id column. `quality` block added: metric
+`anomaly_rate`, `weak` only ABOVE ceiling **0.5**. The A20 finding was **re-measured**
+(2026-09-21): clean gaussian noise flags 0.14–0.22 (mean 0.17, p95 0.22), a genuine
+30%-outlier frame flags 0.30, the demo's degenerate run flagged 0.86 — so 0.5 clears the
+noise floor and a real high-outlier finding yet still catches "most rows flagged"; the
+`note` reports the rate plainly either way. Logic split into `_flag_rate_quality()` so both
+branches are unit-tested (the fit can't be pushed past 0.5 via `contamination`). 4 local
+fixtures (names rows, missing label raises, quality present/not-weak, both branches).
+
 **Verified** `src/tools/analytics/anomaly.py:157-159`
 
 ```python

src/tools/analytics/anomaly.py

@@ -33,6 +33,14 @@ _MIN_ROWS = 10
 _MAX_ANOMALY_ROWS = 50
 # Fixed so the same data always flags the same rows (family-B determinism promise).
 _RANDOM_STATE = 42
+# Above this share of rows flagged, the result is not "a few anomalies" — the notion
+# has collapsed and the answer is noise. The reserved `quality` block reads `weak` off
+# it. NOT a naive low ceiling: A20 reverted one because `contamination="auto"` flags a
+# large slice of CLEAN data. Re-measured 2026-09-21 — clean gaussian noise flags 0.14–0.22
+# (mean 0.17, p95 0.22), a genuine 30%-outlier frame flags 0.30; the demo's degenerate
+# run flagged 0.86. 0.5 clears both the noise floor and a real high-outlier finding while
+# still catching "most rows flagged".
+_ANOMALY_RATE_CEILING = 0.5
 # Internal column names for the highlighted scatter (never collide with user data).
 _FLAG_COL = "__is_anomaly__"
 _INDEX_COL = "__row__"
@@ -52,6 +60,11 @@ class AnomalyInput(BaseModel):
     column_ids: list[str] = Field(
         ..., description="One or more NUMERIC columns to score rows for anomalies."
     )
+    label_columns: list[str] | None = Field(
+        None, description="Identifier columns (e.g. unit/equipment no.) to carry into each "
+        "flagged record so the anomaly can be NAMED. Never fed to the model — pass the id "
+        "columns whenever the question asks 'which X is unusual'."
+    )
     contamination: float | None = Field(
         None, description="Expected fraction of anomalies (0-0.5); omit for automatic."
     )
@@ -66,7 +79,8 @@ DESCRIPTION = build_description(
         "the user wants movement over time or a projection -> analyze_trend / analyze_forecast",
         "the user wants what DRIVES a metric -> analyze_importance",
     ],
-    output="a summary (row/anomaly counts, rate) + the flagged rows + a scatter with "
+    output="a summary (row/anomaly counts, rate) + the flagged rows (named by any "
+    "label_columns given) + a quality verdict on the flag rate + a scatter with "
     "anomaly points highlighted.",
     examples=[
         "ada anomali nggak di data bulan ini?",
@@ -86,6 +100,28 @@ def _fit_iforest(x: np.ndarray, contamination, random_state: int):
     return model
 
 
+def _flag_rate_quality(anomaly_rate: float) -> dict[str, object]:
+    """Reserved `quality` block from the flag rate (conventions §3).
+
+    Split out so both branches are unit-testable without a model run — the fit can't be
+    coaxed above a 0.5 flag rate through `contamination` (sklearn caps it there), so the
+    weak branch has no organic fixture.
+    """
+    weak = anomaly_rate > _ANOMALY_RATE_CEILING
+    return {
+        "metric": "anomaly_rate",
+        "value": round(anomaly_rate, 4),
+        "floor": _ANOMALY_RATE_CEILING,
+        "weak": weak,
+        "note": (
+            f"{anomaly_rate:.0%} of rows flagged — above half the data, so this is "
+            "noise, not a few outliers"
+            if weak
+            else f"{anomaly_rate:.0%} of rows flagged as anomalous"
+        ),
+    }
+
+
 def _anomaly_chart(
     df: pd.DataFrame, column_ids: list[str], is_anomaly: pd.Series
 ) -> dict[str, object]:
@@ -105,6 +141,7 @@ async def analyze_anomaly(
     df: pd.DataFrame,
     column_ids: list[str],
     contamination: float | None = None,
+    label_columns: list[str] | None = None,
 ) -> dict[str, object]:
     """Flag anomalous rows across `column_ids` with an Isolation Forest.
 
@@ -112,17 +149,23 @@ async def analyze_anomaly(
     synchronous compute. `data` is materialised upstream (Pattern A) and arrives as
     `df`.
 
+    `label_columns` are identifier columns carried into each flagged record so the
+    anomaly can be NAMED (e.g. `eqm_no`). They are NEVER fed to the model — only the
+    numeric `column_ids` are scored.
+
     Returns a dict with: method, column_ids, contamination, n_rows, n_anomalies,
     anomaly_rate, anomalies_total (= n_anomalies) / anomalies_shown (how many the
     capped sample actually carries), `anomalies` (the flagged rows, top by score,
-    capped at _MAX_ANOMALY_ROWS), and `charts` (the reserved auto-chart key).
+    capped at _MAX_ANOMALY_ROWS), a reserved `quality` block (flag-rate verdict), and
+    `charts` (the reserved auto-chart key).
 
     Raises (all wrapped by the invoker's never-throw seam):
         ColumnNotFoundError, EmptyColumnSelectionError, NotEnoughDataError.
     """
     if not column_ids:
         raise EmptyColumnSelectionError("column_ids must name at least one numeric column")
-    missing = [c for c in column_ids if c not in df.columns]
+    labels = list(label_columns) if label_columns else []
+    missing = [c for c in column_ids + labels if c not in df.columns]
     if missing:
         raise ColumnNotFoundError(f"columns not found: {missing}")
 
@@ -153,22 +196,33 @@ async def analyze_anomaly(
     anomalies: list[dict[str, object]] = []
     for pos in order[:_MAX_ANOMALY_ROWS]:
         idx = flagged_idx[pos]
-        record = {c: _clean(scored.at[idx, c]) for c in column_ids}
+        # Identifiers FIRST so a reader sees WHICH row before its numbers. Pulled from
+        # the original df (labels are non-numeric and never entered `scored`).
+        record: dict[str, object] = {c: _clean(df.at[idx, c]) for c in labels}
+        record.update({c: _clean(scored.at[idx, c]) for c in column_ids})
         record["anomaly_score"] = round(float(flagged_scores[pos]), 4)
         anomalies.append(record)
 
+    anomaly_rate = float(flags.mean()) if scored.shape[0] else 0.0
     return {
         "method": "isolation_forest",
         "column_ids": column_ids,
         "contamination": contam,
         "n_rows": int(scored.shape[0]),
         "n_anomalies": int(flags.sum()),
-        "anomaly_rate": float(flags.mean()) if scored.shape[0] else 0.0,
+        "anomaly_rate": anomaly_rate,
         # `anomalies` is a bounded top-by-score SAMPLE, not the full set. These two
         # keep it honest: total says how many there really are, shown says how many
         # the sample carries. When total > shown the caller knows it's truncated.
         "anomalies_total": int(flags.sum()),
         "anomalies_shown": len(anomalies),
         "anomalies": anomalies,
+        # Reserved `quality` block (conventions §3): the tool owns its own verdict; CK7
+        # surfaces `weak`. Isolation Forest has no accuracy score, so the honest signal
+        # is the flag RATE — an unsupervised detector that flags most of the data found
+        # nothing. `weak` fires only ABOVE _ANOMALY_RATE_CEILING (0.5), clear of the
+        # ~0.22 clean-data floor A20 measured, so a normal run is never marked weak; the
+        # note reports the rate plainly either way (A20's fallback, kept as the message).
+        "quality": _flag_rate_quality(anomaly_rate),
         "charts": [_anomaly_chart(df, column_ids, is_anomaly)],
     }