[NOTICKET][AI] refactor(planner): R11 - single-source placeholder regex + name-checked data-access specs

Comments are not a sync mechanism. Remove the two duplicated definitions on the
agent side.

- PLACEHOLDER_RE: defined once in planner/schemas.py (it's part of the ToolCall
${t<id>} placeholder convention); planner/validator.py and slow_path/task_runner.py
now import it instead of each compiling their own copy "kept in sync" by comment.
- planner/registry.py still holds local data-access spec BODIES (stub pending
KM-465 #4) but derives/validates their NAMES against the canonical
src.tools.data_access.DATA_ACCESS_TOOLS: _data_access_slice() raises at
default_registry() if a tool is added/renamed/removed upstream, instead of
drifting silently. Registry output unchanged (same 12 tools, same order).

Tests: 51 passed (planner + slow_path); ruff clean; registry builds.

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

PROGRESS.md

@@ -50,7 +50,7 @@ Verified against code before logging. Severity: **critical** / important / nice-
 | R8 | **Catalog upsert race** — per-user advisory lock around read-merge-upsert (`store.py`); concurrent uploads can drop a source. | important | DB | `[ ]` |
 | R9 | **`extra="ignore"`** in `settings.py:15` (currently `allow` → typo'd env vars silently swallowed); require Azure keys in prod. | nice-to-have | B | `[ ]` |
 | R10 | **Read-only enforcement is session-state, not a server role.** `REPO_CONTEXT.md` counts "read-only DB credentials" as a defense layer but nothing requests/verifies a read-only role. Either request read-only creds at registration (verify via `SELECT current_setting(...)`) or drop the claim. | important | DB | `[ ]` |
-| R11 | **De-duplicate** `_PLACEHOLDER_RE` (`task_runner.py:31` vs validator) and `_DATA_ACCESS_TOOLS` (invoker vs planner registry) — import one from the other; comments aren't a sync mechanism. **TAB slice done (90e80f9):** canonical `DATA_ACCESS_TOOLS` now lives once in `tools/data_access.py`; `invoker.py` imports it (was a duplicated frozenset synced by comment). **Agent remainder:** `planner/registry.py` should derive its data-access spec names from that constant, and `_PLACEHOLDER_RE` (validator ↔ task_runner) still needs dedup — both agent-side. | nice-to-have | agent/tool | `[~]` |
+| R11 | **De-duplicate** `_PLACEHOLDER_RE` (`task_runner.py:31` vs validator) and `_DATA_ACCESS_TOOLS` (invoker vs planner registry) — import one from the other; comments aren't a sync mechanism. **TAB slice done (90e80f9):** canonical `DATA_ACCESS_TOOLS` now lives once in `tools/data_access.py`; `invoker.py` imports it (was a duplicated frozenset synced by comment). **Agent slice done (2026-06-10):** `PLACEHOLDER_RE` single-sourced in `planner/schemas.py` (part of the ToolCall placeholder convention); validator + task_runner import it. `planner/registry.py` keeps local spec *bodies* (stub pending KM-465 #4) but name-checks them against `DATA_ACCESS_TOOLS` in `_data_access_slice()` — upstream rename/add now raises at `default_registry()` instead of drifting silently. Registry output unchanged (same 12 tools, same order). | nice-to-have | agent/tool | `[x]` |
 | R12 | **Doc/process hygiene** — some code docstrings cite internal design specs that are not committed to the repo (design docs are kept out of version control), so the references dangle for anyone but the author; `CLAUDE.md` lists deleted modules (enricher, `pipeline/orchestrator.py`); `main` is 38 commits behind on a dead architecture. | nice-to-have | agent | `[ ]` |
 | R13 | **Pre-existing test failure** (found during R2, NOT caused by it): `tests/query/planner/test_prompt.py::test_render_catalog_with_sources` fails — `query/planner/prompt.py::render_catalog` now renders stable IDs (`src_test_db`) the test asserts are absent. Old query-planner path; confirmed failing on a clean tree. | nice-to-have | DB | `[ ]` |
 | T1 | **`input_schema` is presence-only, not type-checked** — `ToolSpec.input_schema` comment said "validates ToolCall.args", but `TaskRunner._validate_args` only enforces `required` presence; the `properties` types are documentation, never validated at runtime. Clarified the contract in `tools/contracts.py` so nobody assumes type-safety (a wrong-typed arg passes validation, surfaces only inside the compute fn). Doc-only, no behavior change (90e80f9). | nice-to-have | TAB | `[x]` |

src/agents/planner/registry.py

@@ -8,10 +8,12 @@ outside it).
   `ToolSpec` (`src/tools/contracts.py`, KM-465/KM-627) and the prompt-style tool
   descriptions (KM-625). No longer a stub on our side — it tracks the real registry.
 - **Data access (`query_structured` / `retrieve_documents` / `list_sources` /
-  `describe_source`) — STILL A LOCAL STUB.** The tool team owns these too, but their
-  wrappers + `ToolSpec`s haven't landed yet (KM-465 #4). We keep best-guess specs
-  here so the Planner can plan end-to-end; when the real ones ship, delete this slice
-  and swap `default_registry()` for the tool team's full composition.
+  `describe_source`) — spec BODIES still a local stub.** The tool team owns these too,
+  but their wrappers + `ToolSpec`s haven't landed yet (KM-465 #4). We keep best-guess
+  spec bodies here so the Planner can plan end-to-end — but the NAMES derive from
+  `src.tools.data_access.DATA_ACCESS_TOOLS` (R11), so a tool rename/addition upstream
+  fails loudly here instead of drifting silently. When the real specs ship, delete
+  this slice and swap `default_registry()` for the tool team's full composition.
 
 **Confirmed conventions (KM-465):** Pattern A — `analyze_*` tools take a `data`
 `"${t<id>}"` placeholder pointing at an upstream `query_structured` output (no
@@ -25,14 +27,16 @@ See AGENT_ARCHITECTURE_CONTEXT_new.md §9.2 / §9.3.
 
 from __future__ import annotations
 
+from src.tools.data_access import DATA_ACCESS_TOOLS
 from src.tools.registry import analytics_registry
 
 from .contracts import ToolRegistry, ToolSpec
 
 # --------------------------------------------------------------------------- #
-# Data-access slice — LOCAL STUB pending the tool team's real specs (KM-465 #4).
+# Data-access slice — spec bodies are a LOCAL STUB pending the tool team's real
+# specs (KM-465 #4); the canonical NAME SET is `DATA_ACCESS_TOOLS` (tool-owned).
 # --------------------------------------------------------------------------- #
-_DATA_ACCESS_TOOLS: list[ToolSpec] = [
+_DATA_ACCESS_SPEC_BODIES: tuple[ToolSpec, ...] = (
     ToolSpec(
         name="query_structured",
         category="analytics.query",
@@ -95,14 +99,34 @@ _DATA_ACCESS_TOOLS: list[ToolSpec] = [
             "before querying it. Do NOT use it to fetch data rows (use query_structured)."
         ),
     ),
-]
+)
+
+_DATA_ACCESS_SPECS: dict[str, ToolSpec] = {s.name: s for s in _DATA_ACCESS_SPEC_BODIES}
+
+
+def _data_access_slice() -> list[ToolSpec]:
+    """Data-access specs in body order, with names checked against the tool layer.
+
+    `DATA_ACCESS_TOOLS` (src.tools.data_access) is the canonical name set; the
+    spec bodies above are still our local stub. Any mismatch (a tool added,
+    renamed, or removed upstream) raises here instead of drifting silently.
+    """
+    if set(_DATA_ACCESS_SPECS) != DATA_ACCESS_TOOLS:
+        missing = sorted(DATA_ACCESS_TOOLS - _DATA_ACCESS_SPECS.keys())
+        stale = sorted(_DATA_ACCESS_SPECS.keys() - DATA_ACCESS_TOOLS)
+        raise RuntimeError(
+            "planner data-access specs out of sync with "
+            f"src.tools.data_access.DATA_ACCESS_TOOLS: missing spec for {missing}, "
+            f"stale spec for {stale}"
+        )
+    return list(_DATA_ACCESS_SPECS.values())
 
 
 def default_registry() -> ToolRegistry:
     """The v1 registry: stub data-access slice + the real analytics slice.
 
     The analytics tools come live from `src.tools.registry` (the tool team's real
-    registry); only the data-access slice is still a local stub. A fresh instance
-    per call.
+    registry); the data-access spec bodies are still a local stub, name-checked
+    against `DATA_ACCESS_TOOLS`. A fresh instance per call.
     """
-    return ToolRegistry(tools=[*_DATA_ACCESS_TOOLS, *analytics_registry().tools])
+    return ToolRegistry(tools=[*_data_access_slice(), *analytics_registry().tools])

src/agents/planner/schemas.py

@@ -10,10 +10,17 @@ See AGENT_ARCHITECTURE_CONTEXT_new.md §7.3.
 
 from __future__ import annotations
 
+import re
 from typing import Any, Literal
 
 from pydantic import BaseModel, Field
 
+# The "${t<id>}" placeholder convention (Pattern A): a ToolCall arg whose value
+# matches this pattern refers to an upstream task's output, resolved by the
+# TaskRunner at execution time. Single definition — the planner validator and
+# the TaskRunner both import it (R11).
+PLACEHOLDER_RE = re.compile(r"\$\{(t[^}]+)\}")
+
 CrispStage = Literal[
     "data_understanding",
     "data_preparation",

src/agents/planner/validator.py

@@ -11,8 +11,6 @@ rejects structurally-invalid plans (duplicate ids, dangling edges, cycles).
 
 from __future__ import annotations
 
-import re
-
 from pydantic import ValidationError
 
 from ...catalog.models import Catalog
@@ -21,11 +19,10 @@ from ...query.ir.validator import IRValidationError, IRValidator
 from .contracts import ToolRegistry
 from .errors import PlannerValidationError
 from .inputs import Constraints
-from .schemas import TaskList
+from .schemas import PLACEHOLDER_RE, TaskList
 
 # Heuristic: a checkable success_criteria mentions a measurable signal.
 _CHECKABLE_TOKENS = ("rate", "count", "match", "produced", "above", "below", "equal")
-_PLACEHOLDER_RE = re.compile(r"\$\{(t[^}]+)\}")
 
 # DFS colors for cycle detection.
 _WHITE, _GREY, _BLACK = 0, 1, 2
@@ -169,7 +166,7 @@ class PlannerValidator:
 
 
 def _is_placeholder(value: str) -> bool:
-    return bool(_PLACEHOLDER_RE.fullmatch(value.strip()))
+    return bool(PLACEHOLDER_RE.fullmatch(value.strip()))
 
 
 def _placeholder_refs(task) -> set[str]:
@@ -177,7 +174,7 @@ def _placeholder_refs(task) -> set[str]:
     for call in task.tool_calls:
         for value in call.args.values():
             if isinstance(value, str):
-                refs.update(_PLACEHOLDER_RE.findall(value))
+                refs.update(PLACEHOLDER_RE.findall(value))
     return refs
 
 

src/agents/slow_path/task_runner.py

@@ -16,22 +16,18 @@ See AGENT_ARCHITECTURE_CONTEXT_new.md §7.4.
 from __future__ import annotations
 
 import asyncio
-import re
 from typing import Any
 
 from src.middlewares.logging import get_logger
 
 from ..planner.contracts import ToolOutput, ToolRegistry
-from ..planner.schemas import Task
+from ..planner.schemas import PLACEHOLDER_RE, Task
 from ..planner.schemas import TaskList as PlanTaskList
 from .invoker import ToolInvoker
 from .schemas import RunState, TaskResult, TaskStatus
 
 logger = get_logger("task_runner")
 
-# Mirrors planner/validator.py:28 `_PLACEHOLDER_RE` — keep the two in sync.
-_PLACEHOLDER_RE = re.compile(r"\$\{(t[^}]+)\}")
-
 
 class TaskRunner:
     """Runs a `TaskList` against a `ToolInvoker`, producing a `RunState`."""
@@ -131,7 +127,7 @@ class TaskRunner:
         # the referenced task's representative output (its last ToolOutput).
         # Materializing that envelope into a DataFrame is the invoker's job.
         if isinstance(value, str):
-            match = _PLACEHOLDER_RE.fullmatch(value.strip())
+            match = PLACEHOLDER_RE.fullmatch(value.strip())
             if match:
                 upstream = results.get(match.group(1))
                 if upstream is None or not upstream.outputs: