[KM-567][AI] Planner agent: contracts, schemas, inputs, registry

Phase 3 slow-path Planner foundation (AGENT_ARCHITECTURE_CONTEXT_new.md §7.3).

- contracts.py: STUB BusinessContext / ToolSpec / ToolRegistry / ToolOutput
(to reconcile with lead + tool team KM-608).
- schemas.py: TaskList / Task / ToolCall output contract. No replan schemas.
- inputs.py: CatalogSummary (condensed, PII sample_values nulled, from_catalog
builder) + Constraints.
- registry.py: STUB v1 P0 tool registry (query_structured, retrieve_documents,
list/describe_source, compute_median/stddev/percentile/mode, date_trunc).
- errors.py: PlannerError / PlannerValidationError.

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

src/agents/planner/__init__.py

@@ -0,0 +1,8 @@
+"""Planner agent — slow-path CRISP-DM analysis planner.
+
+Single LLM call: BusinessContext + CatalogSummary + ToolRegistry + question +
+Constraints -> a validated, static `TaskList` (a DAG of fully-specified
+tool-call chains). No replanning (INV-6); tool-agnostic (INV-7).
+
+See AGENT_ARCHITECTURE_CONTEXT_new.md §7.3.
+"""

src/agents/planner/contracts.py

@@ -0,0 +1,104 @@
+"""STUB contracts the planner consumes from other teams.
+
+These models are LOCAL STUBS so the planner is buildable and testable now.
+They must be reconciled with their owners before integration:
+
+  - `BusinessContext` (+ KeyTerm / DataTableNote / DataColumnNote)
+      Owner: the lead (interview / Business Understanding).
+      Shape mirrors AGENT_ARCHITECTURE_CONTEXT_new.md §7.1.
+
+  - `ToolSpec` / `ToolRegistry`
+      Owner: the tool team (KM-608).
+      Shape mirrors §9.2. The concrete v1 P0 registry instance lives in
+      `registry.py`; this file only defines the contract types.
+
+  - `ToolOutput`
+      The tool -> agent output envelope (§8.1). Not produced by the planner
+      (tools return it at TaskRunner time); defined here so the planner layer
+      owns one definition of the contract it plans against.
+
+When the real modules land, delete the corresponding stub here and import from
+the shared location instead.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+# --------------------------------------------------------------------------- #
+# BusinessContext (lead's contract — §7.1)
+# --------------------------------------------------------------------------- #
+
+
+class KeyTerm(BaseModel):
+    term: str
+    meaning: str
+
+
+class DataTableNote(BaseModel):
+    table_name: str
+    row_represents: str
+
+
+class DataColumnNote(BaseModel):
+    column_name: str
+    meaning: str
+
+
+class BusinessContext(BaseModel):
+    project_id: str
+    industry: str
+    completeness: Literal["partial", "complete"]
+    business_description: str
+    scale_and_scope: str
+    key_terms: list[KeyTerm] = Field(default_factory=list)
+    data_overview: list[DataTableNote] = Field(default_factory=list)
+    data_column_notes: list[DataColumnNote] = Field(default_factory=list)
+    whats_normal: str = ""
+    recent_events: str = ""
+    things_to_watch_for: str = ""
+    open_questions: list[str] = Field(default_factory=list)
+
+
+# --------------------------------------------------------------------------- #
+# Tool registry (tool team's contract — §9.2)
+# --------------------------------------------------------------------------- #
+
+
+class ToolSpec(BaseModel):
+    name: str
+    category: str  # analytics.query | .aggregation | .timeseries | ...
+    input_schema: dict[str, Any]  # validates ToolCall.args
+    output_kind: str  # the ToolOutput.kind it returns
+    description: str  # prompt-style: what it does, edge cases, what NOT to use it for
+    phase: Literal["P0", "P1", "P2"] = "P0"
+
+
+class ToolRegistry(BaseModel):
+    tools: list[ToolSpec] = Field(default_factory=list)
+
+    def names(self) -> set[str]:
+        return {t.name for t in self.tools}
+
+    def get(self, name: str) -> ToolSpec | None:
+        for t in self.tools:
+            if t.name == name:
+                return t
+        return None
+
+
+# --------------------------------------------------------------------------- #
+# Tool output envelope (tool -> agent contract — §8.1)
+# --------------------------------------------------------------------------- #
+
+
+class ToolOutput(BaseModel):
+    tool: str
+    kind: Literal["scalar", "table", "stats", "series", "documents", "error"]
+    value: Any | None = None
+    columns: list[str] | None = None
+    rows: list[list[Any]] | None = None
+    meta: dict[str, Any] = Field(default_factory=dict)
+    error: str | None = None

src/agents/planner/errors.py

@@ -0,0 +1,15 @@
+"""Typed errors for the planner agent."""
+
+from __future__ import annotations
+
+
+class PlannerError(Exception):
+    """Base error for the planner agent."""
+
+
+class PlannerValidationError(PlannerError):
+    """A TaskList failed one of the planner validator's checks.
+
+    The message is specific enough that the planner can be re-prompted with it
+    to self-correct (max 3 attempts, see service.py).
+    """

src/agents/planner/inputs.py

@@ -0,0 +1,132 @@
+"""Planner input models — CatalogSummary and Constraints.
+
+`CatalogSummary` is a condensed, PII-safe view of the user's `Catalog`, built
+for the planner prompt. It carries every table + column id/type/PII flag + row
+counts + low-cardinality top_values, with `sample_values` nulled on PII columns
+(INV: no PII sample values into the prompt, see doc §13). It also lists the
+available unstructured sources so the planner can plan `retrieve_documents`.
+
+The planner *validator* still checks inline `query_structured` IRs against the
+full `Catalog` via the existing IRValidator — the summary is a prompt input, not
+the validation source of truth.
+
+See AGENT_ARCHITECTURE_CONTEXT_new.md §7.3.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+from ...catalog.models import Catalog, DataType
+
+
+class ColumnSummary(BaseModel):
+    column_id: str
+    name: str
+    data_type: DataType
+    pii_flag: bool = False
+    sample_values: list[Any] | None = None  # nulled when pii_flag is True
+    top_values: list[Any] | None = None
+
+
+class TableSummary(BaseModel):
+    table_id: str
+    name: str
+    row_count: int | None = None
+    columns: list[ColumnSummary] = Field(default_factory=list)
+
+
+class StructuredSourceSummary(BaseModel):
+    source_id: str
+    name: str
+    source_type: str  # "schema" | "tabular"
+    tables: list[TableSummary] = Field(default_factory=list)
+
+
+class UnstructuredSourceSummary(BaseModel):
+    source_id: str
+    name: str
+
+
+class CatalogSummary(BaseModel):
+    structured_sources: list[StructuredSourceSummary] = Field(default_factory=list)
+    unstructured_sources: list[UnstructuredSourceSummary] = Field(default_factory=list)
+
+    @classmethod
+    def from_catalog(cls, catalog: Catalog) -> CatalogSummary:
+        structured: list[StructuredSourceSummary] = []
+        unstructured: list[UnstructuredSourceSummary] = []
+
+        for source in catalog.sources:
+            if source.source_type == "unstructured":
+                unstructured.append(
+                    UnstructuredSourceSummary(source_id=source.source_id, name=source.name)
+                )
+                continue
+
+            tables = [
+                TableSummary(
+                    table_id=table.table_id,
+                    name=table.name,
+                    row_count=table.row_count,
+                    columns=[
+                        ColumnSummary(
+                            column_id=col.column_id,
+                            name=col.name,
+                            data_type=col.data_type,
+                            pii_flag=col.pii_flag,
+                            sample_values=None if col.pii_flag else col.sample_values,
+                            top_values=col.stats.top_values if col.stats else None,
+                        )
+                        for col in table.columns
+                    ],
+                )
+                for table in source.tables
+            ]
+            structured.append(
+                StructuredSourceSummary(
+                    source_id=source.source_id,
+                    name=source.name,
+                    source_type=source.source_type,
+                    tables=tables,
+                )
+            )
+
+        return cls(structured_sources=structured, unstructured_sources=unstructured)
+
+    def render(self) -> str:
+        """Render the summary as compact text for the planner prompt."""
+        if not self.structured_sources and not self.unstructured_sources:
+            return "(catalog is empty — the user has not registered any data yet)"
+
+        lines: list[str] = []
+        for source in self.structured_sources:
+            lines.append(f"Source: {source.name} ({source.source_type}) — id={source.source_id}")
+            for table in source.tables:
+                rc = f" ({table.row_count:,} rows)" if table.row_count is not None else ""
+                lines.append(f"  Table: {table.name}{rc} — id={table.table_id}")
+                for col in table.columns:
+                    samples = "PII (suppressed)" if col.pii_flag else (col.sample_values or [])
+                    top = f", top={col.top_values}" if col.top_values else ""
+                    lines.append(
+                        f"    - {col.name} [{col.data_type}]: "
+                        f"samples={samples}{top} — id={col.column_id}"
+                    )
+            lines.append("")
+
+        if self.unstructured_sources:
+            lines.append("Unstructured sources (for retrieve_documents):")
+            for src in self.unstructured_sources:
+                lines.append(f"  - {src.name} — id={src.source_id}")
+
+        return "\n".join(lines).rstrip()
+
+
+class Constraints(BaseModel):
+    max_tasks: int = 5
+    modeling_allowed: bool = False  # no modeling tools in v1
+    token_budget: int | None = None
+    time_budget_seconds: int | None = None
+    row_budget: int = 10_000

src/agents/planner/registry.py

@@ -0,0 +1,159 @@
+"""STUB v1 P0 tool registry.
+
+This is the agent team's local stand-in for the tool team's inventory (KM-608)
+so the planner is buildable and testable before the real tools land. The tools
+here are *contracts only* — there is no implementation behind them; the planner
+plans against the registry and never names a tool outside it (INV-7).
+
+`input_schema` is a lightweight JSON-schema-ish dict consumed by the planner
+validator (validator.py check #8): it carries `required` (list of arg names) and
+`properties` (allowed arg names). Arg *values* may be "${t<id>}" placeholders the
+TaskRunner resolves at execution time, so the validator checks arg *keys*, not
+value types — except `query_structured.args["ir"]`, whose inline QueryIR is
+validated against the catalog by the existing IRValidator.
+
+When KM-608 ships, replace `default_registry()` with the real registry import.
+See AGENT_ARCHITECTURE_CONTEXT_new.md §9.2 / §9.3.
+"""
+
+from __future__ import annotations
+
+from .contracts import ToolRegistry, ToolSpec
+
+_P0_TOOLS: list[ToolSpec] = [
+    ToolSpec(
+        name="query_structured",
+        category="analytics.query",
+        input_schema={"required": ["ir"], "properties": {"ir": {"type": "object"}}},
+        output_kind="table",
+        description=(
+            "Run one validated, single-table query against a structured source (DB "
+            "schema or tabular file) and return rows. The `ir` argument is an inline "
+            "QueryIR (the JSON intent: source_id, table_id, select, filters, group_by, "
+            "order_by, limit) — never SQL. Use this for any selection, filtering, "
+            "grouping, or built-in aggregation (count/sum/avg/min/max/count_distinct). "
+            "Do NOT use it for medians/percentiles/modes/stddev (use the compute_* "
+            "tools on its output) and do NOT use it to read documents (use "
+            "retrieve_documents)."
+        ),
+    ),
+    ToolSpec(
+        name="retrieve_documents",
+        category="retrieval.documents",
+        input_schema={
+            "required": ["query"],
+            "properties": {
+                "query": {"type": "string"},
+                "source_id": {"type": "string"},
+                "top_k": {"type": "integer"},
+            },
+        },
+        output_kind="documents",
+        description=(
+            "Dense-retrieve the most relevant chunks from the user's unstructured "
+            "sources (PDF/DOCX/TXT) for a natural-language `query`. Use this to pull "
+            "qualitative context into an analysis. Optionally scope to one `source_id`. "
+            "Do NOT use it for numbers in tables — that is query_structured's job."
+        ),
+    ),
+    ToolSpec(
+        name="list_sources",
+        category="catalog.introspection",
+        input_schema={"required": [], "properties": {}},
+        output_kind="table",
+        description=(
+            "List the user's available data sources (id, name, type, table count). Use "
+            "early in data_understanding when the plan must discover what exists before "
+            "querying. Cheap. Do NOT use it to read column details (use describe_source)."
+        ),
+    ),
+    ToolSpec(
+        name="describe_source",
+        category="catalog.introspection",
+        input_schema={
+            "required": ["source_id"],
+            "properties": {"source_id": {"type": "string"}},
+        },
+        output_kind="table",
+        description=(
+            "Return the tables and columns (names, types, row counts) of one source by "
+            "`source_id`. Use in data_understanding to confirm the shape of a source "
+            "before querying it. Do NOT use it to fetch data rows (use query_structured)."
+        ),
+    ),
+    ToolSpec(
+        name="compute_median",
+        category="analytics.aggregation",
+        input_schema={"required": ["values"], "properties": {"values": {"type": "array"}}},
+        output_kind="scalar",
+        description=(
+            "Compute the median of a numeric series. `values` is typically a "
+            "'${t<id>}' placeholder referencing an upstream query_structured output "
+            "column. Use this because SQL/pandas median is not exposed via the IR. Do "
+            "NOT use it on categorical data (use compute_mode)."
+        ),
+    ),
+    ToolSpec(
+        name="compute_stddev",
+        category="analytics.aggregation",
+        input_schema={"required": ["values"], "properties": {"values": {"type": "array"}}},
+        output_kind="scalar",
+        description=(
+            "Compute the standard deviation of a numeric series (`values`, usually a "
+            "'${t<id>}' placeholder from an upstream query). Use to quantify spread or "
+            "to flag outliers. Do NOT use on non-numeric data."
+        ),
+    ),
+    ToolSpec(
+        name="compute_percentile",
+        category="analytics.aggregation",
+        input_schema={
+            "required": ["values", "percentile"],
+            "properties": {
+                "values": {"type": "array"},
+                "percentile": {"type": "number"},
+            },
+        },
+        output_kind="scalar",
+        description=(
+            "Compute a given `percentile` (0-100) of a numeric series `values` "
+            "(usually a '${t<id>}' placeholder). Use for p90/p95-style thresholds. Do "
+            "NOT use for the median alone (use compute_median)."
+        ),
+    ),
+    ToolSpec(
+        name="compute_mode",
+        category="analytics.aggregation",
+        input_schema={"required": ["values"], "properties": {"values": {"type": "array"}}},
+        output_kind="scalar",
+        description=(
+            "Compute the most frequent value(s) of a series `values` (usually a "
+            "'${t<id>}' placeholder). Works on categorical or numeric data. Use to find "
+            "the typical category. Do NOT use it for an average (use query_structured "
+            "avg)."
+        ),
+    ),
+    ToolSpec(
+        name="date_trunc",
+        category="analytics.timeseries",
+        input_schema={
+            "required": ["values", "granularity"],
+            "properties": {
+                "values": {"type": "array"},
+                "granularity": {"type": "string"},
+            },
+        },
+        output_kind="series",
+        description=(
+            "Truncate a datetime series `values` (usually a '${t<id>}' placeholder) to "
+            "a `granularity` ('day' | 'week' | 'month' | 'quarter' | 'year') so results "
+            "can be grouped into time buckets for trend analysis. Do NOT use it to "
+            "filter by date — put a date filter in the query_structured IR instead."
+        ),
+    ),
+]
+
+
+def default_registry() -> ToolRegistry:
+    """The v1 P0 stub registry (a fresh instance per call)."""
+    return ToolRegistry(tools=list(_P0_TOOLS))

src/agents/planner/schemas.py

@@ -0,0 +1,54 @@
+"""Planner output schemas — the `TaskList` contract.
+
+The planner emits exactly one `TaskList`: a DAG of typed tasks, each an ordered
+chain of fully-specified tool calls. This is the static plan the TaskRunner
+executes verbatim. There is no replanning (INV-6), so there are no
+`ReplanRequest` / `ReplanResponse` schemas.
+
+See AGENT_ARCHITECTURE_CONTEXT_new.md §7.3.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+CrispStage = Literal[
+    "data_understanding",
+    "data_preparation",
+    "modeling",  # no tools in v1; the planner does not emit modeling tasks
+    "evaluation",
+]
+
+
+class ToolCall(BaseModel):
+    """One call to a registry tool with concrete, fully-specified arguments.
+
+    `tool` must exist in the ToolRegistry. `args` is validated against the
+    tool's input_schema; it may contain "${t<id>}" placeholders that the
+    TaskRunner resolves from an upstream task's output at execution time.
+    """
+
+    tool: str
+    args: dict[str, Any] = Field(default_factory=dict)
+
+
+class Task(BaseModel):
+    id: str  # "t1", "t2", ...
+    stage: CrispStage
+    objective: str  # plain-language intent for this step
+    tool_calls: list[ToolCall] = Field(..., min_length=1)  # ordered chain
+    expected_output: str  # named result this task produces
+    success_criteria: str  # REPORTING signal, not a control trigger
+    depends_on: list[str] = Field(default_factory=list)  # task ids
+    parallelizable_with: list[str] = Field(default_factory=list)
+    estimated_cost: Literal["low", "medium", "high"] = "low"
+
+
+class TaskList(BaseModel):
+    plan_id: str
+    goal_restated: str
+    assumptions: list[str] = Field(default_factory=list)
+    open_questions: list[str] = Field(default_factory=list)
+    tasks: list[Task] = Field(default_factory=list)