feat(chisel): Phase B — eval harness + canonical queries

pebble/chisel/__init__.py

@@ -1,39 +1,40 @@
 """Chisel — Pebble's tool/workflow authoring framework.
 
-A tool or workflow becomes a directory under ``pebble/chisel/{tools,workflows}/``
-containing a declarative ``manifest.yaml`` plus a Python ``handler.py``. At
-process start, ``autoload()`` walks those dirs and registers each one on the
-shared ``DEFAULT_REGISTRY`` so the planner / executor / renderer pick them up
+A tool or workflow is a directory under ``pebble/chisel/{tools,workflows}/``
+containing a declarative ``manifest.yaml`` plus a Python ``handler.py``.
+``autoload()`` walks those dirs and registers each unit on the shared
+``DEFAULT_REGISTRY`` so the planner / executor / renderer pick them up
 through the existing contract.
 
-Public surface (kept small on purpose):
-
-    autoload(registry=None, root=None) -> AutoloadReport
-    snapshot(registry) -> ToolRegistry
-    slash_command_map() -> dict[str, str]
-    dispatch_workflow(intent) -> str | None
-
-See ``tasks/pebble-chisel-plan.md`` for the locked phase-A spec.
+Autoload runs at module import so any pebble-package consumer (the
+streaming handler, the router, the CLI) sees a populated registry
+without explicit wiring. Tests pass ``registry=fresh_registry`` to
+``autoload()`` for isolation; the maps used by the router
+(``lookup_slash`` / ``lookup_intent``) reset on every call.
 """
 
 from __future__ import annotations
 
 from .autoload import (
     AutoloadReport,
+    WorkflowEntry,
     autoload,
-    build_workflow_plan,
-    dispatch_workflow,
-    slash_command_map,
-    slash_to_intent,
+    lookup_intent,
+    lookup_slash,
 )
 from .reload import snapshot
 
 __all__ = [
     "AutoloadReport",
+    "WorkflowEntry",
     "autoload",
-    "build_workflow_plan",
-    "dispatch_workflow",
-    "slash_command_map",
-    "slash_to_intent",
+    "lookup_intent",
+    "lookup_slash",
     "snapshot",
 ]
+
+
+# Run autoload at import time so any pebble path that imports chisel
+# sees a populated DEFAULT_REGISTRY. Errors flow through the report and
+# get logged; the process boots with whatever loaded successfully.
+_BOOT_REPORT = autoload()

pebble/chisel/autoload.py

@@ -1,27 +1,26 @@
-"""Walk ``pebble/chisel/{tools,workflows}/`` and register each unit
-on a ``ToolRegistry``.
+"""Walk ``pebble/chisel/{tools,workflows}/`` and register each unit on
+a ``ToolRegistry``.
 
-Failure policy (plan §9, Phase-A risks): a malformed manifest or an
-import error in one handler must NOT block the others. ``autoload``
-returns an ``AutoloadReport`` listing what loaded and which dirs errored
-so the app surfaces failures at ``/api/chisel/health`` (Phase C) without
-crashing the process.
+Failure policy: a malformed manifest or import error in one unit must
+NOT block the others. ``autoload`` returns an ``AutoloadReport`` listing
+what loaded and which dirs errored so the app surfaces failures at
+``/api/chisel/health`` (Phase C) without crashing the process.
 
-Public entry points:
+Public surface:
   * ``autoload(registry=None, root=None)`` — discover + register.
-  * ``slash_command_map()`` — ``{slash: workflow_name}`` for the router.
-  * ``dispatch_workflow(intent)`` — replaces the hard-coded
-    ``_build_workflow_plan_for_intent`` dispatch in ``handlers/streaming.py``.
+  * ``lookup_slash(slash)`` — slash → WorkflowEntry (router dispatch).
+  * ``lookup_intent(intent)`` — intent → WorkflowEntry (orchestrator dispatch).
 """
 
 from __future__ import annotations
 
 import importlib
 import importlib.util
+import logging
 import sys
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any, Callable, Optional
 
 import yaml
 from pydantic import BaseModel, ValidationError
@@ -33,71 +32,43 @@
 )
 
 from .handler_adapter import build_handler_wrapper
-from .manifest import (
-    ToolManifest,
-    WorkflowManifest,
-    cost_estimate_to_float,
-)
+from .lints import lint_handler_module
+from .manifest import ToolManifest, WorkflowManifest
 from .schema import pydantic_to_strict_schema
 
+logger = logging.getLogger(__name__)
+
 
 # ---------------------------------------------------------------------------
-# Report
+# Workflow registry (single source of truth — no dual maps)
 # ---------------------------------------------------------------------------
 
+@dataclass(frozen=True)
+class WorkflowEntry:
+    name: str
+    dispatch_intent: str
+    slash_command: Optional[str]
+    build_plan: Callable[..., Any]
+
+
 @dataclass
 class AutoloadReport:
     loaded_tools: list[str] = field(default_factory=list)
     loaded_workflows: list[str] = field(default_factory=list)
     errors: list[tuple[str, str]] = field(default_factory=list)  # (path, reason)
+    lint_warnings: list[tuple[str, str]] = field(default_factory=list)
 
-    def ok(self) -> bool:
-        return not self.errors
-
-
-# ---------------------------------------------------------------------------
-# Slash + intent dispatch — populated by autoload.
-# ---------------------------------------------------------------------------
-
-_SLASH_COMMANDS: dict[str, str] = {}
-_INTENT_DISPATCH: dict[str, str] = {}
-_PLAN_BUILDERS: dict[str, Any] = {}  # workflow_name → callable(**kwargs) -> Plan
-
-
-def slash_command_map() -> dict[str, str]:
-    """Return a copy of the slash → workflow_name map. Used by
-    ``pebble/router.py`` to replace its hard-coded ``_SLASH_COMMANDS``
-    dict."""
-    return dict(_SLASH_COMMANDS)
 
+_BY_SLASH: dict[str, WorkflowEntry] = {}
+_BY_INTENT: dict[str, WorkflowEntry] = {}
 
-def slash_to_intent(slash: str) -> Optional[str]:
-    """Return the dispatch_intent registered for a slash command, or
-    None. Used by ``pebble/router.py`` to populate RouteResult.intent."""
-    workflow_name = _SLASH_COMMANDS.get(slash)
-    if workflow_name is None:
-        return None
-    for intent, name in _INTENT_DISPATCH.items():
-        if name == workflow_name:
-            return intent
-    return None
 
+def lookup_slash(slash: str) -> Optional[WorkflowEntry]:
+    return _BY_SLASH.get(slash)
 
-def dispatch_workflow(intent: str) -> Optional[str]:
-    """Return the workflow name registered for ``intent`` (planner
-    output), or None. Replaces ``_build_workflow_plan_for_intent``."""
-    return _INTENT_DISPATCH.get(intent)
 
-
-def build_workflow_plan(intent_or_name: str, **kwargs: Any) -> Optional[Any]:
-    """Look up the workflow by intent (preferred) or by name, then call
-    its registered ``build_plan(**kwargs)``. Returns the Plan, or None
-    if no workflow is registered for that intent/name."""
-    name = _INTENT_DISPATCH.get(intent_or_name, intent_or_name)
-    builder = _PLAN_BUILDERS.get(name)
-    if builder is None:
-        return None
-    return builder(**kwargs)
+def lookup_intent(intent: str) -> Optional[WorkflowEntry]:
+    return _BY_INTENT.get(intent)
 
 
 # ---------------------------------------------------------------------------
@@ -108,61 +79,56 @@ def autoload(
     *,
     registry: Optional[ToolRegistry] = None,
     root: Optional[Path] = None,
-    reset: bool = True,
 ) -> AutoloadReport:
-    """Discover Chisel units and register them.
-
-    Args:
-        registry: registry to register specs on; defaults to
-            ``DEFAULT_REGISTRY`` so production paths Just Work. Tests pass a
-            fresh ``ToolRegistry()`` for isolation (plan §P4).
-        root: directory containing ``tools/`` and ``workflows/`` subdirs;
-            defaults to ``pebble/chisel/`` next to this file.
-        reset: clear the slash/intent maps before populating. Tests may
-            pass False to accumulate registrations across autoload calls.
-    """
+    """Discover Chisel units and register them. Resets the workflow
+    lookup maps before populating; pass an isolated ``registry`` to
+    avoid touching ``DEFAULT_REGISTRY`` from tests."""
     if registry is None:
         registry = DEFAULT_REGISTRY
     if root is None:
         root = Path(__file__).parent
 
     report = AutoloadReport()
-
-    if reset:
-        _SLASH_COMMANDS.clear()
-        _INTENT_DISPATCH.clear()
-        _PLAN_BUILDERS.clear()
+    _BY_SLASH.clear()
+    _BY_INTENT.clear()
 
     tools_root = root / "tools"
     if tools_root.is_dir():
         for tool_dir in sorted(p for p in tools_root.iterdir() if p.is_dir()):
-            if tool_dir.name.startswith("_") or tool_dir.name.startswith("."):
+            if tool_dir.name.startswith(("_", ".")):
                 continue
             try:
-                _load_tool(tool_dir, registry)
+                _load_tool(tool_dir, registry, report)
                 report.loaded_tools.append(tool_dir.name)
             except Exception as e:  # noqa: BLE001 — surface, don't crash
                 report.errors.append((str(tool_dir), f"{type(e).__name__}: {e}"))
 
     workflows_root = root / "workflows"
     if workflows_root.is_dir():
         for wf_dir in sorted(p for p in workflows_root.iterdir() if p.is_dir()):
-            if wf_dir.name.startswith("_") or wf_dir.name.startswith("."):
+            if wf_dir.name.startswith(("_", ".")):
                 continue
             try:
                 _load_workflow(wf_dir)
                 report.loaded_workflows.append(wf_dir.name)
             except Exception as e:  # noqa: BLE001
                 report.errors.append((str(wf_dir), f"{type(e).__name__}: {e}"))
 
+    for path, reason in report.lint_warnings:
+        logger.warning("chisel lint %s: %s", path, reason)
+
     return report
 
 
 # ---------------------------------------------------------------------------
 # Per-unit loaders
 # ---------------------------------------------------------------------------
 
-def _load_tool(tool_dir: Path, registry: ToolRegistry) -> None:
+def _load_tool(
+    tool_dir: Path,
+    registry: ToolRegistry,
+    report: AutoloadReport,
+) -> None:
     manifest_path = tool_dir / "manifest.yaml"
     if not manifest_path.is_file():
         raise FileNotFoundError(f"missing manifest.yaml in {tool_dir}")
@@ -177,6 +143,12 @@ def _load_tool(tool_dir: Path, registry: ToolRegistry) -> None:
     if not handler_path.is_file():
         raise FileNotFoundError(f"missing handler.py in {tool_dir}")
 
+    # Advisory lints — warnings only, don't block registration.
+    for err in lint_handler_module(handler_path):
+        report.lint_warnings.append(
+            (str(handler_path), f"{err.rule}:{err.lineno}: {err.message}"),
+        )
+
     module = _import_module(
         handler_path,
         package=f"pebble.chisel.tools.{tool_dir.name}.handler",
@@ -192,16 +164,15 @@ def _load_tool(tool_dir: Path, registry: ToolRegistry) -> None:
         user_run=user_run,
     )
 
-    spec = ToolSpec(
+    registry.register(ToolSpec(
         name=manifest.name,
         description=manifest.description,
         input_schema=pydantic_to_strict_schema(input_model),
         handler=wrapped,
-        cost_estimate_usd=cost_estimate_to_float(manifest.cost_estimate),
+        cost_estimate_usd=manifest.cost_estimate_usd,
         requires_human=manifest.requires_human,
         tags=manifest.tags,
-    )
-    registry.register(spec)
+    ))
 
 
 def _load_workflow(wf_dir: Path) -> None:
@@ -215,35 +186,34 @@ def _load_workflow(wf_dir: Path) -> None:
     except ValidationError as e:
         raise ValueError(f"workflow invalid: {e.errors()}") from e
 
-    if manifest.slash_command:
-        _SLASH_COMMANDS[manifest.slash_command] = manifest.name
-    if manifest.dispatch_intent:
-        _INTENT_DISPATCH[manifest.dispatch_intent] = manifest.name
-
     if manifest.has_custom_plan:
         build_plan_path = wf_dir / "build_plan.py"
         if not build_plan_path.is_file():
             raise FileNotFoundError(
                 f"workflow {manifest.name!r} sets has_custom_plan=true "
-                f"but {build_plan_path.name} is missing",
+                f"but build_plan.py is missing",
             )
         module = _import_module(
             build_plan_path,
             package=f"pebble.chisel.workflows.{wf_dir.name}.build_plan",
         )
-        # The build_plan import also has to work for tmp_path-rooted
-        # tests; spec_from_file_location path handles that branch.
         builder = _resolve_callable(module, "build_plan")
-        _PLAN_BUILDERS[manifest.name] = builder
     else:
-        # Declarative form — synthesize a build_plan from manifest.steps.
-        _PLAN_BUILDERS[manifest.name] = _make_declarative_builder(manifest)
+        builder = _make_declarative_builder(manifest)
+
+    entry = WorkflowEntry(
+        name=manifest.name,
+        dispatch_intent=manifest.dispatch_intent,  # type: ignore[arg-type]  # filled by validator
+        slash_command=manifest.slash_command,
+        build_plan=builder,
+    )
+    if entry.slash_command:
+        _BY_SLASH[entry.slash_command] = entry
+    _BY_INTENT[entry.dispatch_intent] = entry
 
 
-def _make_declarative_builder(manifest: WorkflowManifest) -> Any:
-    """Compile a workflow's declarative ``steps[]`` into a build_plan
-    callable so the orchestrator can run it through the same code path
-    as a custom build_plan."""
+def _make_declarative_builder(manifest: WorkflowManifest) -> Callable[..., Any]:
+    """Synthesize a build_plan callable from declarative ``steps[]``."""
     from pebble.orchestrator.schemas import Plan, PlanStep
 
     def builder(*, user_query: str = manifest.description, **_unused: Any) -> Plan:
@@ -261,17 +231,15 @@ def builder(*, user_query: str = manifest.description, **_unused: Any) -> Plan:
 
 
 # ---------------------------------------------------------------------------
-# helpers
+# Module-loading helpers
 # ---------------------------------------------------------------------------
 
 def _import_module(path: Path, *, package: str) -> Any:
-    """Load a chisel-resident Python module. Prefers the standard import
-    machinery when the module lives under the real ``pebble.chisel.*``
-    package tree (so relative imports like ``from .compute import x``
-    resolve). Falls back to spec_from_file_location for ad-hoc paths
-    used in tests (``tmp_path`` outside the source tree)."""
+    """Load a chisel module. Uses the standard import system when the
+    file lives under the real ``pebble.chisel.*`` tree (so relative
+    imports resolve); falls back to spec_from_file_location for
+    tmp_path-based tests."""
     try:
-        # Real source tree path → use the normal import system.
         path_resolved = path.resolve()
         chisel_root = Path(__file__).parent.resolve()
         path_resolved.relative_to(chisel_root)