fix(planner): reject inlined table-ref as analyze_* data (Pattern A)

Check 9 only guarded `data`/`data_right` when it was a string placeholder, so
when the planner inlined a {table_id, source_id} dict (skipping retrieve_data)
the plan slipped through validation and only failed at execution with the opaque
"unsupported 'data' type: dict" — surfaced to the user as a misleading "data type
not supported" answer. Now any non-placeholder `data`/`data_right` is rejected in
the validator so the retry loop re-prompts the planner to chain a retrieve_data
output. Also harden _materialize: a {table_id/source_id} dict returns an
actionable Pattern-A error instead of the bare type name, as a defensive net.

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

src/agents/planner/validator.py

@@ -212,23 +212,45 @@ class PlannerValidator:
     def _validate_data_source(
         task_id: str, call, tasks_by_id: dict, registry: ToolRegistry
     ) -> None:
-        """A `data` placeholder must reference a data-producing task, not a
-        metadata (check_data/check_knowledge) or documents (retrieve_knowledge) one.
-
-        Those pass the structural checks (check_* also returns kind="table"), but
-        their rows are catalog schema, so a downstream analyze_* fails to find the
-        requested columns. Resolving points at the referenced task's representative
-        output — its last tool call (matches TaskRunner's `outputs[-1]`).
+        """The `data`/`data_right` handoff (Pattern A) must be a '${t<id>}'
+        placeholder pointing at an upstream data-producing task.
+
+        Two failure modes are rejected here:
+
+        1. The arg is present but is NOT a placeholder — e.g. the planner inlined a
+           {table_id, source_id} table reference (or any literal) instead of
+           chaining a retrieve_data output. analyze_* tools never self-fetch by
+           source_id; they only consume an already-retrieved table, so such a plan
+           can only blow up at execution ("unsupported 'data' type: dict"). Reject
+           it here so the retry loop re-prompts the planner to add a retrieve_data
+           task and pass its placeholder.
+        2. The placeholder resolves to a metadata (check_data/check_knowledge) or
+           documents (retrieve_knowledge) task. Those pass the structural checks
+           (check_* also returns kind="table"), but their rows are catalog schema,
+           so a downstream analyze_* fails to find the requested columns. Resolving
+           points at the referenced task's representative output — its last tool
+           call (matches TaskRunner's `outputs[-1]`).
+
+        `data_right` is analyze_merge's second table input (KM-703) — same Pattern A
+        handoff, so it gets the same guard. An arg that is absent is left to Check 8a
+        (required-arg presence).
         """
-        # `data_right` is analyze_merge's second table input (KM-703) — same
-        # Pattern A handoff, so it gets the same guard.
         for arg_name in ("data", "data_right"):
-            data_arg = call.args.get(arg_name)
-            if not isinstance(data_arg, str):
-                continue
-            match = PLACEHOLDER_RE.fullmatch(data_arg.strip())
-            if not match:
+            if arg_name not in call.args:
                 continue
+            data_arg = call.args[arg_name]
+            match = (
+                PLACEHOLDER_RE.fullmatch(data_arg.strip())
+                if isinstance(data_arg, str)
+                else None
+            )
+            if match is None:
+                raise PlannerValidationError(
+                    f"task {task_id}: tool {call.tool!r} arg {arg_name!r} must be a "
+                    f"'${{t<id>}}' placeholder referencing a retrieve_data output "
+                    f"(Pattern A), got {data_arg!r}. analyze_* tools do not fetch data "
+                    "themselves — add a retrieve_data task and pass its output."
+                )
             ref_task = tasks_by_id.get(match.group(1))
             if ref_task is None or not ref_task.tool_calls:
                 continue  # a dangling placeholder is reported by the DAG check

src/tools/invoker.py

@@ -163,6 +163,18 @@ def _materialize(data: Any) -> tuple[pd.DataFrame, None] | tuple[None, str]:
         df = pd.DataFrame(data.get("rows") or [], columns=data["columns"])
         return _normalize_numeric(df), None
 
+    # A {table_id/source_id} dict is a raw catalog reference the planner inlined
+    # instead of chaining a retrieve_data output (Pattern A). analyze_* tools never
+    # self-fetch, so give an actionable message rather than the opaque type name.
+    # The planner validator should reject this upstream (Check 9); this is the
+    # defensive net if a bad plan still reaches execution.
+    if isinstance(data, dict) and ("table_id" in data or "source_id" in data):
+        return None, (
+            "'data' is a table reference (table_id/source_id), not a retrieved "
+            "table — analyze_* consumes the output of retrieve_data, it does not "
+            "fetch data by id (Pattern A)"
+        )
+
     return None, f"unsupported 'data' type: {type(data).__name__}"