[KM-567][AI] Planner agent: fix PII leak + placeholder/retry robustness

Review follow-ups on the Planner agent:

- inputs.py: suppress a PII column's top_values when building CatalogSummary.
Previously only sample_values were nulled, so a low-cardinality PII column's
top_values (same class of data) could reach the planner prompt. Now both are
suppressed. (Regression test added.)
- validator.py: a ${tN} placeholder is now valid when its target is a
transitive ancestor, not only a direct depends_on. Cycle detection moved
ahead of the placeholder check so ancestors are well-defined on a DAG. This
stops valid plans (target reached via an intermediate task) being rejected.
- service.py: floor max_retries at 1 (mirrors QueryService) so a misconfigured
value can't skip the LLM call entirely.

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

src/agents/planner/inputs.py

@@ -77,8 +77,15 @@ class CatalogSummary(BaseModel):
                             name=col.name,
                             data_type=col.data_type,
                             pii_flag=col.pii_flag,
+                            # PII columns leak nothing into the prompt: both
+                            # sample_values and (low-cardinality) top_values are
+                            # suppressed — top_values are the same class of data.
                             sample_values=None if col.pii_flag else col.sample_values,
-                            top_values=col.stats.top_values if col.stats else None,
+                            top_values=(
+                                None
+                                if col.pii_flag or col.stats is None
+                                else col.stats.top_values
+                            ),
                         )
                         for col in table.columns
                     ],

src/agents/planner/service.py

@@ -87,7 +87,7 @@ class PlannerService:
     ) -> None:
         self._chain = structured_chain
         self._validator = validator or PlannerValidator()
-        self._max_retries = max_retries
+        self._max_retries = max(1, max_retries)
 
     def _ensure_chain(self) -> Runnable:
         if self._chain is None:

src/agents/planner/validator.py

@@ -148,22 +148,28 @@ class PlannerValidator:
                     raise PlannerValidationError(
                         f"task {task.id}: depends_on includes itself"
                     )
-            # Placeholders must reference an existing, declared dependency.
+
+        cycle = _find_cycle(tasks_by_id)
+        if cycle:
+            raise PlannerValidationError(f"cycle detected in depends_on: {' -> '.join(cycle)}")
+
+        # On an acyclic graph, a placeholder is safe iff its target is a
+        # transitive ancestor — i.e. guaranteed to have completed before this
+        # task runs. Requiring a *direct* depends_on would wrongly reject valid
+        # plans that depend on the target through an intermediate task.
+        ancestors = _all_ancestors(tasks_by_id)
+        for task in tasks_by_id.values():
             for ref in _placeholder_refs(task):
                 if ref not in id_set:
                     raise PlannerValidationError(
                         f"task {task.id}: placeholder '${{{ref}}}' references unknown task"
                     )
-                if ref not in task.depends_on:
+                if ref not in ancestors[task.id]:
                     raise PlannerValidationError(
                         f"task {task.id}: placeholder '${{{ref}}}' used but {ref!r} is "
-                        f"not in depends_on"
+                        f"not a (transitive) dependency — add it to depends_on"
                     )
 
-        cycle = _find_cycle(tasks_by_id)
-        if cycle:
-            raise PlannerValidationError(f"cycle detected in depends_on: {' -> '.join(cycle)}")
-
     @staticmethod
     def _validate_parallelism(tasks_by_id: dict, id_set: set[str]) -> None:
         ancestors = _all_ancestors(tasks_by_id)