fix(analyze_aggregate): reject non-string group_by (CASE/binning) keys

When asked to bucket a numeric column ("group uptime into high/medium/low"), the
planner emitted a CASE expression as a group_by entry (a dict). df.groupby then
crashed with a bare "TypeError: unhashable type: 'dict'", surfaced to the user as
a misleading "technical issue". Guard both layers, same as the data-handoff fix:
the validator (new Check 8d) rejects any non-string group_by key so the retry
loop re-prompts the planner, and analyze_aggregate raises a clear
UnsupportedGroupByError instead of the opaque TypeError if one still reaches it.
Derived bucketing belongs to analyze_segment, not analyze_aggregate.

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

src/agents/planner/validator.py

@@ -142,6 +142,23 @@ class PlannerValidator:
                                 "use analyze_descriptive instead."
                             )
 
+                    # Check 8d — group_by keys must be plain column-name strings. A
+                    # derived grouping (a CASE/binning expression emitted as a dict)
+                    # is unhashable and crashes df.groupby at execution ("unhashable
+                    # type: 'dict'"). Reject it here so the planner is re-prompted;
+                    # bucketing a numeric column into ranges belongs to
+                    # analyze_segment, not smuggled through analyze_aggregate.
+                    group_by = call.args.get("group_by")
+                    if isinstance(group_by, list):
+                        bad_keys = [g for g in group_by if not isinstance(g, str)]
+                        if bad_keys:
+                            raise PlannerValidationError(
+                                f"task {task.id}: analyze_aggregate group_by must be "
+                                f"column names (strings); got non-string entr(ies) "
+                                f"{bad_keys}. Derived groupings (CASE/binning) are not "
+                                "supported — group by an existing column."
+                            )
+
                 # Check 3 — concrete source_id args must exist in the catalog.
                 src = call.args.get("source_id")
                 if isinstance(src, str) and not _is_placeholder(src):

src/tools/analytics/aggregation.py

@@ -27,6 +27,17 @@ class UnsupportedAggregationError(ValueError):
     """Requested aggregation is not in SUPPORTED_AGGS (maps to UNSUPPORTED_AGG)."""
 
 
+class UnsupportedGroupByError(ValueError):
+    """A group_by entry is not a plain column name (maps to UNSUPPORTED_GROUP_BY).
+
+    group_by keys must be column-name strings. A derived grouping — e.g. a CASE /
+    binning expression the planner emits as a dict — is unhashable and would crash
+    `df.groupby` with a bare "unhashable type: 'dict'"; raising here turns it into a
+    clear, actionable error instead. Bucketing a numeric column into ranges belongs
+    to analyze_segment, not analyze_aggregate.
+    """
+
+
 def _clean(value: object) -> object:
     """Convert numpy/pandas scalars to plain Python so the output is JSON-clean.
 
@@ -103,6 +114,17 @@ def analyze_aggregate(
     """
     group_by = group_by or []
 
+    # group_by keys must be plain column names. A non-string entry (e.g. a CASE /
+    # binning dict) is unhashable and would otherwise crash the membership check or
+    # df.groupby with an opaque "unhashable type: 'dict'".
+    non_str = [g for g in group_by if not isinstance(g, str)]
+    if non_str:
+        raise UnsupportedGroupByError(
+            f"group_by entries must be column names (strings); got non-string "
+            f"entr(ies) {non_str}. Derived groupings (e.g. CASE/binning expressions) "
+            "are not supported by analyze_aggregate — group by an existing column."
+        )
+
     # Validate columns first (fail-fast on caller mistakes).
     referenced = list(group_by) + list(aggregations.keys())
     missing = [c for c in referenced if c not in df.columns]