[KM-629] AnalyticsToolInvoker (wrapper layer)

Implement the ToolInvoker Protocol (src/agents/slow_path/invoker.py) for the
analyze_* family: invoke(tool_name, args) -> ToolOutput.

- Dispatch tool_name -> (compute fn, output_kind); unknown name -> error envelope.
- Materialize the Pattern A `data` arg (already resolved by the TaskRunner to the
upstream ToolOutput kind="table") into a DataFrame; also accepts a raw
DataFrame or {columns, rows} dict.
- Pass remaining args as kwargs (names match compute signatures 1:1); wrap the
result in ToolOutput with the tool's declared kind.
- Never-throw: unknown tool, bad/missing data, or a compute exception all return
ToolOutput(kind="error"), honoring the §8.4 seam guarantee.

Verified locally (tests are gitignored): 10 invoker tests + full tools suite
82 passed; ruff + mypy strict clean.

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

src/tools/invoker.py

@@ -0,0 +1,105 @@
+"""AnalyticsToolInvoker — the runtime seam implementation (KM-465).
+
+Implements the `ToolInvoker` Protocol the slow-path TaskRunner calls
+(src/agents/slow_path/invoker.py). One method, `invoke(tool_name, args)`, does the
+whole job for the `analyze_*` family:
+
+1. Look the tool up in a name -> (compute fn, output_kind) dispatch map; an unknown
+   name returns an error envelope (never an exception).
+2. Materialize the Pattern A `data` argument — which the TaskRunner has already
+   resolved to the upstream task's `ToolOutput` (kind="table") — into a DataFrame.
+3. Call the pure compute function with the remaining args as keyword arguments
+   (their names match the compute signatures one-to-one).
+4. Wrap the result in a `ToolOutput` with the tool's declared `kind`.
+
+Frozen guarantee (§8.4): **never throws.** Any failure — unknown tool, bad data,
+or an exception from compute (e.g. GroupNotFoundError) — comes back as
+`ToolOutput(kind="error", error=...)`, so the TaskRunner's degrade-and-continue
+keeps working.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+import pandas as pd
+
+from src.tools.analytics import (
+    aggregation,
+    comparison,
+    decomposition,
+    descriptive,
+    quality,
+    relationship,
+    segmentation,
+    temporal,
+)
+from src.tools.contracts import ToolOutput
+
+# tool name -> (compute callable, ToolOutput.kind it produces). Kept in lockstep
+# with src/tools/registry.py output_kind values.
+_DISPATCH: dict[str, tuple[Callable[..., Any], str]] = {
+    "analyze_descriptive": (descriptive.analyze_descriptive, "stats"),
+    "analyze_aggregate": (aggregation.analyze_aggregate, "table"),
+    "analyze_comparison": (comparison.analyze_comparison, "stats"),
+    "analyze_contribution": (decomposition.analyze_contribution, "table"),
+    "analyze_profile": (quality.analyze_profile, "stats"),
+    "analyze_correlation": (relationship.analyze_correlation, "stats"),
+    "analyze_segment": (segmentation.analyze_segment, "table"),
+    "analyze_trend": (temporal.analyze_trend, "series"),
+}
+
+
+class AnalyticsToolInvoker:
+    """Never-throwing invoker for the `analyze_*` tools (implements ToolInvoker)."""
+
+    async def invoke(self, tool_name: str, args: dict[str, Any]) -> ToolOutput:
+        entry = _DISPATCH.get(tool_name)
+        if entry is None:
+            return ToolOutput(
+                tool=tool_name, kind="error", error=f"unknown tool {tool_name!r}"
+            )
+        fn, kind = entry
+
+        df, err = _materialize(args.get("data"))
+        if err is not None:
+            return ToolOutput(tool=tool_name, kind="error", error=err)
+
+        kwargs = {k: v for k, v in args.items() if k != "data"}
+        try:
+            result = fn(df, **kwargs)
+        except Exception as exc:  # noqa: BLE001 — never-throw seam (§8.4)
+            return ToolOutput(
+                tool=tool_name,
+                kind="error",
+                error=f"{type(exc).__name__}: {exc}",
+            )
+
+        return ToolOutput(tool=tool_name, kind=kind, value=result)
+
+
+def _materialize(data: Any) -> tuple[pd.DataFrame, None] | tuple[None, str]:
+    """Turn the resolved `data` argument into a DataFrame.
+
+    Accepts the upstream `ToolOutput` (kind="table"), a raw DataFrame, or a
+    {"columns", "rows"} dict (a serialized table). Returns (df, None) on success
+    or (None, error_message) on failure — the caller wraps the message.
+    """
+    if data is None:
+        return None, "missing 'data' argument (no upstream table to analyze)"
+
+    if isinstance(data, pd.DataFrame):
+        return data, None
+
+    if isinstance(data, ToolOutput):
+        if data.kind == "error":
+            return None, f"upstream data unavailable: {data.error}"
+        if data.kind != "table" or data.columns is None:
+            return None, f"cannot materialize 'data' of kind {data.kind!r}"
+        return pd.DataFrame(data.rows or [], columns=data.columns), None
+
+    if isinstance(data, dict) and "columns" in data:
+        return pd.DataFrame(data.get("rows") or [], columns=data["columns"]), None
+
+    return None, f"unsupported 'data' type: {type(data).__name__}"