diff --git a/pebble/chisel/__init__.py b/pebble/chisel/__init__.py index 887f378f..cd21bcec 100644 --- a/pebble/chisel/__init__.py +++ b/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() diff --git a/pebble/chisel/autoload.py b/pebble/chisel/autoload.py index 8523138e..ecd71b2f 100644 --- a/pebble/chisel/autoload.py +++ b/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,38 +79,26 @@ 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}")) @@ -147,7 +106,7 @@ def autoload( 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) @@ -155,6 +114,9 @@ def autoload( 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 @@ -162,7 +124,11 @@ def autoload( # 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) diff --git a/pebble/chisel/cli.py b/pebble/chisel/cli.py index d857ecec..67aa539f 100644 --- a/pebble/chisel/cli.py +++ b/pebble/chisel/cli.py @@ -1,14 +1,11 @@ """``chisel`` CLI — author/inspect/validate tools and workflows. -Phase A surface (argparse, plan §11.4): +Surface (argparse): - * ``chisel list`` — print loaded tools + workflows. - * ``chisel validate [path]`` — schema + lint check one unit or the tree. - * ``chisel scaffold `` — write a fresh tool dir with manifest + - handler stub + handler_test.py scaffold (plan §P10). - -Phase B will add ``chisel eval``. Phase C wires the same operations -behind ``/api/chisel/*``. + * ``chisel list`` — print loaded tools + workflows. + * ``chisel validate [path]`` — schema + lint check. + * ``chisel scaffold `` — create a new tool dir from a stub. + * ``chisel eval [--unit] [--tag]`` — run canonical_queries through the planner. """ from __future__ import annotations @@ -48,10 +45,7 @@ async def run(args: Input, ctx: HandlerContext) -> dict[str, Any]: version: 1.0.0 tags: [] requires_human: false -cost_estimate: - fixed: 0.0 -output_kind: prose -scope: global +cost_estimate_usd: 0.0 """ TEST_STUB = '''"""Tests for `{name}`.""" @@ -86,6 +80,10 @@ def main(argv: list[str] | None = None) -> int: s = sub.add_parser("scaffold", help="create a new tool dir") s.add_argument("name", help="tool name, snake_case") + e = sub.add_parser("eval", help="run canonical_queries through the planner") + e.add_argument("--unit", help="restrict to one tool/workflow by name", default=None) + e.add_argument("--tag", help="restrict to queries tagged with this label", default=None) + args = parser.parse_args(argv) if args.cmd == "list": @@ -94,6 +92,8 @@ def main(argv: list[str] | None = None) -> int: return _cmd_validate(args.path) if args.cmd == "scaffold": return _cmd_scaffold(args.name) + if args.cmd == "eval": + return _cmd_eval(unit=args.unit, tag=args.tag) parser.error(f"unknown command {args.cmd!r}") return 2 @@ -114,13 +114,22 @@ def _cmd_validate(path: str | None) -> int: report = autoload() failed = False + # Lint scope: a specific dir if given, else every loaded handler. + chisel_root = Path(__file__).parent if path: - handler_path = Path(path) / "handler.py" - if handler_path.is_file(): - errs = lint_handler_module(handler_path) - for err in errs: - print(f"lint:{err.rule}:{err.lineno}: {err.message}") - failed = True + targets = [Path(path) / "handler.py"] + else: + targets = [ + chisel_root / "tools" / name / "handler.py" + for name in report.loaded_tools + ] + + for handler_path in targets: + if not handler_path.is_file(): + continue + for err in lint_handler_module(handler_path): + print(f"lint {handler_path.parent.name}: {err.rule}:{err.lineno}: {err.message}") + failed = True for unit_path, reason in report.errors: print(f"autoload: {unit_path}: {reason}") @@ -147,5 +156,50 @@ def _cmd_scaffold(name: str) -> int: return 0 +def _cmd_eval(*, unit: str | None, tag: str | None) -> int: + import asyncio + import os + + from pebble.chisel.eval import ( + format_results, + load_canonical_queries, + run_plan_eval, + ) + from pebble.orchestrator.planner import Planner + from pebble.orchestrator.tools import DEFAULT_REGISTRY, ToolContext + + autoload() + + queries = load_canonical_queries() + if unit: + queries = [q for q in queries if q.unit == unit] + if tag: + queries = [q for q in queries if tag in q.query.tags] + + if not queries: + print("chisel eval: no canonical queries matched filters.") + return 0 + + if not os.environ.get("ANTHROPIC_API_KEY"): + print( + "chisel eval: ANTHROPIC_API_KEY not set. Skipping live planner " + "calls — schema validation passed.", + ) + return 0 + + from pebble.llm.anthropic_client import get_default_client + + client = get_default_client() + planner = Planner(client=client, registry=DEFAULT_REGISTRY) + ctx = ToolContext(user_email="eval@pursuit.org", conversation_id="chisel-eval") + + async def _run_all(): + return [await run_plan_eval(lq, planner=planner, ctx=ctx) for lq in queries] + + results = asyncio.run(_run_all()) + print(format_results(results)) + return 0 if all(r.passed for r in results) else 1 + + if __name__ == "__main__": raise SystemExit(main()) diff --git a/pebble/chisel/eval.py b/pebble/chisel/eval.py new file mode 100644 index 00000000..d4024bab --- /dev/null +++ b/pebble/chisel/eval.py @@ -0,0 +1,256 @@ +"""Chisel eval harness — canonical queries → expected plans → expected prose. + +Each tool / workflow may ship a ``canonical_queries.yaml`` next to its +manifest. The eval runner loads all of them, runs each query through +the planner (and optionally the full pipeline), and reports per-query +pass/fail against the expectations. + +Plan-level assertions are cheap and deterministic-ish: + * The planner is called with ``temperature=0`` (verified at + ``planner.py``); plan-shape expectations are stable. + * ``expected_plan[]`` is an ordered list of ``ExpectedStep`` entries + matched against ``plan.steps`` in order. + +Prose-level assertions need the full pipeline (executor + renderer) +and therefore live behind a gated CI job — ``chisel eval`` only runs +them when ``--with-prose`` is passed AND a real DB / HTTP client is +wired in. Substring includes/excludes per plan §11.6 (deliberately no +regex). +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional + +import yaml +from pydantic import BaseModel, ConfigDict, Field, ValidationError + +from pebble.orchestrator.planner import Planner, PlannerError +from pebble.orchestrator.schemas import Plan +from pebble.orchestrator.tools import ToolContext + + +# --------------------------------------------------------------------------- +# canonical_queries.yaml schema +# --------------------------------------------------------------------------- + +class ExpectedStep(BaseModel): + """One expected step in the planner's output. ``args_includes`` + asserts each (key, value) pair is present in the actual step's + args (subset match). ``args_excludes`` asserts the listed keys are + NOT in the actual args.""" + model_config = ConfigDict(extra="forbid") + + tool: str = Field(min_length=1) + args_includes: dict[str, Any] = Field(default_factory=dict) + args_excludes: tuple[str, ...] = () + + +class ExpectedProse(BaseModel): + """Substring assertions on the rendered final-response text.""" + model_config = ConfigDict(extra="forbid") + + includes: tuple[str, ...] = () + excludes: tuple[str, ...] = () + + +class CanonicalQuery(BaseModel): + """One canonical query — the user prompt + expectations.""" + model_config = ConfigDict(extra="forbid") + + id: str = Field(min_length=1, pattern=r"^[a-z][a-z0-9_]*$") + user_query: str = Field(min_length=1) + expected_plan: tuple[ExpectedStep, ...] = () + expected_prose: Optional[ExpectedProse] = None + tags: tuple[str, ...] = () + skip_reason: Optional[str] = None + + +class CanonicalQueriesFile(BaseModel): + """Top-level shape of ``canonical_queries.yaml``.""" + model_config = ConfigDict(extra="forbid") + + queries: tuple[CanonicalQuery, ...] + + +# --------------------------------------------------------------------------- +# Loader +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class LoadedQuery: + """A canonical query plus the path it was loaded from (for error + messages and diff output).""" + source: Path + unit: str # tool or workflow name + query: CanonicalQuery + + +def load_canonical_queries(chisel_root: Optional[Path] = None) -> list[LoadedQuery]: + """Walk ``pebble/chisel/{tools,workflows}/*/canonical_queries.yaml`` + and return every query with its source path.""" + if chisel_root is None: + chisel_root = Path(__file__).parent + + out: list[LoadedQuery] = [] + for sub in ("tools", "workflows"): + root = chisel_root / sub + if not root.is_dir(): + continue + for unit_dir in sorted(p for p in root.iterdir() if p.is_dir()): + f = unit_dir / "canonical_queries.yaml" + if not f.is_file(): + continue + raw = yaml.safe_load(f.read_text(encoding="utf-8")) or {} + try: + parsed = CanonicalQueriesFile(**raw) + except ValidationError as e: + raise ValueError( + f"{f}: invalid canonical_queries.yaml: {e.errors()}", + ) from e + for q in parsed.queries: + out.append(LoadedQuery(source=f, unit=unit_dir.name, query=q)) + return out + + +# --------------------------------------------------------------------------- +# Assertions +# --------------------------------------------------------------------------- + +def assert_plan(actual: Plan, expected: tuple[ExpectedStep, ...]) -> list[str]: + """Compare ``actual.steps`` against ``expected``. Returns a list + of mismatch messages; empty list = pass. + + Step ordering is checked positionally. Extra steps after the + expected suffix are tolerated only if the expected list is empty + (no expectations = no plan-shape checks).""" + if not expected: + return [] + + failures: list[str] = [] + if len(actual.steps) < len(expected): + failures.append( + f"step_count: expected at least {len(expected)} steps, " + f"got {len(actual.steps)} ({[s.tool for s in actual.steps]})", + ) + return failures + + for idx, exp in enumerate(expected): + act = actual.steps[idx] + if act.tool != exp.tool: + failures.append( + f"step[{idx}].tool: expected {exp.tool!r}, got {act.tool!r}", + ) + continue + for k, v in exp.args_includes.items(): + if k not in act.args: + failures.append( + f"step[{idx}].args: missing key {k!r}", + ) + elif act.args[k] != v: + failures.append( + f"step[{idx}].args[{k!r}]: expected {v!r}, got {act.args[k]!r}", + ) + for k in exp.args_excludes: + if k in act.args: + failures.append( + f"step[{idx}].args: forbidden key {k!r} present " + f"(value={act.args[k]!r})", + ) + return failures + + +def assert_prose(text: str, expected: ExpectedProse) -> list[str]: + """Substring checks (case-sensitive). Plan §11.6: deliberately NOT + regex — too easy to write fragile assertions.""" + failures: list[str] = [] + for needle in expected.includes: + if needle not in text: + failures.append(f"prose.includes: missing substring {needle!r}") + for needle in expected.excludes: + if needle in text: + failures.append(f"prose.excludes: forbidden substring {needle!r} present") + return failures + + +# --------------------------------------------------------------------------- +# Eval runner +# --------------------------------------------------------------------------- + +@dataclass +class EvalResult: + query_id: str + unit: str + source: Path + passed: bool + plan_failures: list[str] = field(default_factory=list) + prose_failures: list[str] = field(default_factory=list) + planner_error: Optional[str] = None + duration_ms: int = 0 + skipped: bool = False + skip_reason: Optional[str] = None + + +async def run_plan_eval( + loaded: LoadedQuery, + *, + planner: Planner, + ctx: ToolContext, +) -> EvalResult: + """Run one canonical query through the planner only. Cheap path — + no executor / renderer. Verifies the plan shape matches expectation. + """ + q = loaded.query + started = time.perf_counter() + + if q.skip_reason: + return EvalResult( + query_id=q.id, unit=loaded.unit, source=loaded.source, + passed=True, skipped=True, skip_reason=q.skip_reason, + ) + + plan_or_err = await planner.plan(user_query=q.user_query, ctx=ctx) + duration_ms = int((time.perf_counter() - started) * 1000) + + if isinstance(plan_or_err, PlannerError): + return EvalResult( + query_id=q.id, unit=loaded.unit, source=loaded.source, + passed=False, planner_error=f"{plan_or_err.reason}: {plan_or_err.detail}", + duration_ms=duration_ms, + ) + + failures = assert_plan(plan_or_err, q.expected_plan) + return EvalResult( + query_id=q.id, unit=loaded.unit, source=loaded.source, + passed=not failures, plan_failures=failures, duration_ms=duration_ms, + ) + + +# --------------------------------------------------------------------------- +# Reporting +# --------------------------------------------------------------------------- + +def format_results(results: list[EvalResult]) -> str: + """Human-readable summary for the CLI / CI log.""" + passed = sum(1 for r in results if r.passed and not r.skipped) + failed = sum(1 for r in results if not r.passed) + skipped = sum(1 for r in results if r.skipped) + total = len(results) + lines = [f"chisel eval — {passed}/{total} passed ({failed} failed, {skipped} skipped)"] + for r in results: + if r.skipped: + lines.append(f" SKIP {r.unit}/{r.query_id} — {r.skip_reason}") + elif r.passed: + lines.append(f" PASS {r.unit}/{r.query_id} ({r.duration_ms}ms)") + else: + lines.append(f" FAIL {r.unit}/{r.query_id} ({r.duration_ms}ms)") + if r.planner_error: + lines.append(f" planner_error: {r.planner_error}") + for f in r.plan_failures: + lines.append(f" plan: {f}") + for f in r.prose_failures: + lines.append(f" prose: {f}") + return "\n".join(lines) diff --git a/pebble/chisel/handler_adapter.py b/pebble/chisel/handler_adapter.py index 729a32c2..0d7e4433 100644 --- a/pebble/chisel/handler_adapter.py +++ b/pebble/chisel/handler_adapter.py @@ -1,28 +1,19 @@ """Wrap a Chisel-authored handler into the existing ``ToolHandler`` contract (``async def(args: dict, ctx: ToolContext) -> ToolResult``). -The Chisel author writes: +The author writes: class Input(BaseModel): query: str - limit: int = 8 async def run(args: Input, ctx: HandlerContext) -> dict: - ... return {"items": [...]} # plain dict # OR: raise SomeError(...) on failure -This module: - 1. parses ``args`` dict → ``Input`` (returns ``ok=False`` on validation error); - 2. wraps ``ctx`` (``ToolContext``) in a ``HandlerContext`` that exposes - ``cite()`` plus the underlying ctx fields; - 3. calls ``run``, times it, catches exceptions; - 4. converts the returned dict into a ``ToolResult`` with the manifest's - ``version`` recorded in ``tool_version`` (P11 forward-compat with - Sprint-11 scratchpad). - -Handlers stop hand-rolling ToolResult — that's the per-tool boilerplate -elimination win (P2). +The adapter parses dict → Pydantic, times the call, catches exceptions, +records ``tool_version`` (P11), and returns a ToolResult. Handlers +never hand-roll ToolResult — that's the per-tool boilerplate elimination +win (P2). """ from __future__ import annotations @@ -38,10 +29,9 @@ async def run(args: Input, ctx: HandlerContext) -> dict: class HandlerContext: - """Thin read-only wrapper over ``ToolContext`` exposing a ``cite()`` - helper. v1 keeps ``http_client`` as the raw httpx client (no extra - wrapping; see plan §11.7). - """ + """Per-call wrapper around ``ToolContext`` that adds citation + collection. Attribute access (``ctx.user_email``, ``ctx.http_client``) + transparently forwards to the underlying ToolContext.""" __slots__ = ("_ctx", "_citations") @@ -49,38 +39,19 @@ def __init__(self, ctx: ToolContext) -> None: self._ctx = ctx self._citations: list[str] = [] - @property - def user_email(self) -> str: - return self._ctx.user_email - - @property - def conversation_id(self) -> str: - return self._ctx.conversation_id - - @property - def org_id(self) -> str: - return self._ctx.org_id + def __getattr__(self, name: str) -> Any: + return getattr(self._ctx, name) - @property - def db_pool(self) -> Any: - return self._ctx.db_pool - - @property - def http_client(self) -> Any: - return self._ctx.http_client - - def cite(self, entity_type: str, entity_id: str) -> str: - """Append a citation in the canonical ``entity_type:entity_id`` - shape and return it so handlers can also include it inline.""" + def cite(self, entity_type: Any, entity_id: Any) -> str: cite_id = f"{entity_type}:{entity_id}" self._citations.append(cite_id) return cite_id - def collected_citations(self) -> tuple[str, ...]: + @property + def citations(self) -> tuple[str, ...]: return tuple(self._citations) -# Author's run() signature: (parsed_args, handler_ctx) -> awaitable dict UserRun = Callable[[BaseModel, HandlerContext], Awaitable[dict[str, Any]]] @@ -94,45 +65,35 @@ def build_handler_wrapper( """Return an async function with the legacy ToolHandler signature that the existing registry / executor / planner consume unchanged.""" + def _fail(error: str, started: float) -> ToolResult: + return ToolResult( + step_id=uuid4(), + tool=tool_name, + ok=False, + error=error, + duration_ms=int((time.perf_counter() - started) * 1000), + tool_version=tool_version, + ) + async def adapter(args: dict[str, Any], ctx: ToolContext) -> ToolResult: started = time.perf_counter() try: parsed = input_model(**args) except ValidationError as e: - return ToolResult( - step_id=uuid4(), - tool=tool_name, - ok=False, - error=f"input_validation: {e.errors()}", - duration_ms=int((time.perf_counter() - started) * 1000), - tool_version=tool_version, - ) + return _fail(f"input_validation: {e.errors()}", started) handler_ctx = HandlerContext(ctx) try: data = await user_run(parsed, handler_ctx) except Exception as e: # noqa: BLE001 — wrap any handler error - return ToolResult( - step_id=uuid4(), - tool=tool_name, - ok=False, - error=f"{type(e).__name__}: {e}", - duration_ms=int((time.perf_counter() - started) * 1000), - tool_version=tool_version, - ) + return _fail(f"{type(e).__name__}: {e}", started) if not isinstance(data, dict): - return ToolResult( - step_id=uuid4(), - tool=tool_name, - ok=False, - error=( - f"handler_contract: run() must return dict, " - f"got {type(data).__name__}" - ), - duration_ms=int((time.perf_counter() - started) * 1000), - tool_version=tool_version, + return _fail( + f"handler_contract: run() must return dict, " + f"got {type(data).__name__}", + started, ) return ToolResult( @@ -140,7 +101,7 @@ async def adapter(args: dict[str, Any], ctx: ToolContext) -> ToolResult: tool=tool_name, ok=True, data=data, - citations=handler_ctx.collected_citations(), + citations=handler_ctx.citations, duration_ms=int((time.perf_counter() - started) * 1000), tool_version=tool_version, ) diff --git a/pebble/chisel/lints.py b/pebble/chisel/lints.py index f6b0e449..a4643f56 100644 --- a/pebble/chisel/lints.py +++ b/pebble/chisel/lints.py @@ -1,13 +1,13 @@ -"""Static lints over a Chisel handler module. +"""Static lints over a Chisel handler module. Advisory at autoload +(emitted as warnings); ``chisel validate`` exits non-zero on lint hits. -Per plan §3 / §7 lint surface: - - * ``import httpx`` (or ``from httpx import ...``) at module top — handlers - must use ``ctx.http_client`` so audit + timeout policy is centralized. - * ``os.environ`` reads inside ``run()`` — config must flow through ctx. +Checks: + * ``import httpx`` (or ``from httpx import ...``) — handlers must use + ``ctx.http_client`` so audit + timeout policy stay centralized. * ``async def run`` is required — sync ``run`` breaks the adapter. -Run via ``chisel validate`` or at autoload time (errors → AutoloadReport). +Dropped ``no_env_in_run`` (Phase B cleanup): trivially evadable via +``from os import environ``; load-bearing only if enforced robustly. """ from __future__ import annotations @@ -24,78 +24,50 @@ class LintError: lineno: int = 0 -def lint_handler_module( - path: Path | str, - *, - overrides: tuple[str, ...] = (), -) -> list[LintError]: +def lint_handler_module(path: Path | str) -> list[LintError]: p = Path(path) source = p.read_text(encoding="utf-8") tree = ast.parse(source, filename=str(p)) errors: list[LintError] = [] - - overrides_set = set(overrides) - - def emit(rule: str, message: str, lineno: int) -> None: - if rule in overrides_set: - return - errors.append(LintError(rule=rule, message=message, lineno=lineno)) - - _check_no_bare_httpx(tree, emit) - _check_run_signature(tree, emit) - _check_no_env_in_run(tree, emit) + _check_no_bare_httpx(tree, errors) + _check_async_run(tree, errors) return errors -def _check_no_bare_httpx(tree: ast.AST, emit) -> None: +def _check_no_bare_httpx(tree: ast.AST, errors: list[LintError]) -> None: for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: if alias.name == "httpx" or alias.name.startswith("httpx."): - emit( - "no_bare_httpx", - "handler imports httpx directly; use ctx.http_client", - node.lineno, - ) + errors.append(LintError( + rule="no_bare_httpx", + message="handler imports httpx directly; use ctx.http_client", + lineno=node.lineno, + )) elif isinstance(node, ast.ImportFrom): if node.module == "httpx" or ( node.module and node.module.startswith("httpx.") ): - emit( - "no_bare_httpx", - "handler imports from httpx directly; use ctx.http_client", - node.lineno, - ) + errors.append(LintError( + rule="no_bare_httpx", + message="handler imports from httpx directly; use ctx.http_client", + lineno=node.lineno, + )) -def _check_run_signature(tree: ast.AST, emit) -> None: +def _check_async_run(tree: ast.AST, errors: list[LintError]) -> None: found = False for node in ast.iter_child_nodes(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == "run": found = True if isinstance(node, ast.FunctionDef): - emit( - "async_run_required", - "`run` must be defined with `async def`", - node.lineno, - ) + errors.append(LintError( + rule="async_run_required", + message="`run` must be defined with `async def`", + lineno=node.lineno, + )) if not found: - emit("async_run_required", "module is missing `async def run`", 0) - - -def _check_no_env_in_run(tree: ast.AST, emit) -> None: - for node in ast.iter_child_nodes(tree): - if not isinstance(node, ast.AsyncFunctionDef) or node.name != "run": - continue - for sub in ast.walk(node): - if isinstance(sub, ast.Attribute): - if ( - isinstance(sub.value, ast.Name) - and sub.value.id == "os" - and sub.attr == "environ" - ): - emit( - "no_env_in_run", - "run() reads os.environ; flow config through ctx", - sub.lineno, - ) + errors.append(LintError( + rule="async_run_required", + message="module is missing `async def run`", + )) diff --git a/pebble/chisel/manifest.py b/pebble/chisel/manifest.py index 9eb67c3f..87f0c546 100644 --- a/pebble/chisel/manifest.py +++ b/pebble/chisel/manifest.py @@ -2,92 +2,35 @@ (workflow). One schema, one source of truth; load + validate happens in ``autoload``. -Shape locked by ``tasks/pebble-chisel-plan.md §3``. Highlights: - - * ``cost_estimate`` is either fixed (``{fixed: 0.001}``) or variable - (``{variable: {max: 0.50}}``); v1 collapses to a single float internally - but the YAML shape leaves room (P8). - * ``output_kind`` discriminates renderer behaviour so tools without a - per-tool renderer (``generate_chart``, ``request_human_review``) don't - need to ship a render template (P7). - * ``scope`` defaults to ``global``; org-scoped manifests are reserved for - v1.1 but the field is in the schema today (P9). +Trimmed to fields with live consumers. Phase-A speculation (eval_fixtures, +lint_overrides, scope, requires_permission, output_kind, VariableCost) +removed; add back when there's a real reader. """ from __future__ import annotations -from typing import Literal, Optional +from typing import Optional from pydantic import BaseModel, ConfigDict, Field, model_validator -# --------------------------------------------------------------------------- -# Cost — fixed | variable -# --------------------------------------------------------------------------- - -class FixedCost(BaseModel): - """``cost_estimate: {fixed: 0.001}`` — single per-call number.""" - model_config = ConfigDict(extra="forbid") - fixed: float = Field(ge=0.0) - - -class VariableCost(BaseModel): - """``cost_estimate: {variable: {max: 0.50}}`` — LLM-driven tools.""" - model_config = ConfigDict(extra="forbid") - - class _Variable(BaseModel): - model_config = ConfigDict(extra="forbid") - max: float = Field(ge=0.0) - - variable: _Variable - - -CostEstimate = FixedCost | VariableCost - - -def cost_estimate_to_float(cost: CostEstimate) -> float: - """Collapse to the single-float shape the planner's budget pre-flight - consumes. v1 uses the cap for the variable case.""" - if isinstance(cost, FixedCost): - return cost.fixed - return cost.variable.max - - # --------------------------------------------------------------------------- # Tool manifest # --------------------------------------------------------------------------- -OutputKind = Literal["prose", "chart", "checkpoint", "none"] -Scope = Literal["global", "org"] - - class ToolManifest(BaseModel): - """The declarative half of a Chisel tool. The Python half is + """Declarative half of a Chisel tool. The Python half is ``handler.py``'s Pydantic input model + ``async def run``.""" model_config = ConfigDict(extra="forbid") - # Identity name: str = Field(min_length=1, pattern=r"^[a-z][a-z0-9_]*$") description: str = Field(min_length=1) version: str = Field(default="1.0.0", pattern=r"^\d+\.\d+\.\d+$") tags: tuple[str, ...] = () - # Behaviour requires_human: bool = False - requires_permission: Optional[str] = Field( - default=None, - description="Sprint-12 RBAC permission name (snake_case, e.g. 'chisel_write').", - ) - cost_estimate: CostEstimate = FixedCost(fixed=0.0) - output_kind: OutputKind = "prose" - scope: Scope = "global" - - # Phase-B forward-compat: path inside the tool dir to canonical queries. - eval_fixtures: Optional[str] = None - - # Lint overrides — per-tool escape hatches if a check fires false-positive. - lint_overrides: tuple[str, ...] = () + cost_estimate_usd: float = Field(default=0.0, ge=0.0) # --------------------------------------------------------------------------- @@ -95,7 +38,8 @@ class ToolManifest(BaseModel): # --------------------------------------------------------------------------- class WorkflowStep(BaseModel): - """One step in a declaratively-authored workflow.""" + """One step in a declarative workflow. Used only when + ``has_custom_plan=False``.""" model_config = ConfigDict(extra="forbid") tool: str = Field(min_length=1) args: dict = Field(default_factory=dict) @@ -103,9 +47,8 @@ class WorkflowStep(BaseModel): class WorkflowManifest(BaseModel): - """``workflow.yaml`` — declarative form. Engineers can drop a - ``build_plan.py`` alongside for advanced cases that don't fit the - declarative shape; ``has_custom_plan`` flags those for the GUI.""" + """``workflow.yaml``. ``dispatch_intent`` is auto-derived as + ``workflow_`` if omitted — convention every workflow has used.""" model_config = ConfigDict(extra="forbid") @@ -113,27 +56,24 @@ class WorkflowManifest(BaseModel): description: str = Field(min_length=1) version: str = Field(default="1.0.0", pattern=r"^\d+\.\d+\.\d+$") - # Slash command + planner intent the dispatcher matches on. slash_command: Optional[str] = Field( default=None, pattern=r"^/[a-z][a-z0-9_-]*$", ) dispatch_intent: Optional[str] = None - # Declarative form (used if has_custom_plan is False). steps: tuple[WorkflowStep, ...] = () - - # Engineer escape hatch: build_plan.py beside workflow.yaml. has_custom_plan: bool = False - requires_permission: Optional[str] = None - cost_estimate: CostEstimate = FixedCost(fixed=0.0) - scope: Scope = "global" + cost_estimate_usd: float = Field(default=0.0, ge=0.0) @model_validator(mode="after") - def _shape_required(self) -> "WorkflowManifest": + def _normalize(self) -> "WorkflowManifest": if not self.has_custom_plan and not self.steps: raise ValueError( "WorkflowManifest must declare steps[] or set has_custom_plan=true", ) + if not self.dispatch_intent: + # Auto-fill: convention every workflow uses. + object.__setattr__(self, "dispatch_intent", f"workflow_{self.name}") return self diff --git a/pebble/chisel/rbac.py b/pebble/chisel/rbac.py deleted file mode 100644 index c8c9d558..00000000 --- a/pebble/chisel/rbac.py +++ /dev/null @@ -1,60 +0,0 @@ -"""RBAC stub for Chisel — interim, until Sprint-12 lands. - -Plan §11.9: bypass list driven by ``PEBBLE_CHISEL_RBAC_BYPASS_USERS`` -(comma-separated emails), defaulting to ``PEBBLE_CHAT_ALLOWED_EMAILS`` -if that env exists. When Sprint-12 ships the real permission resolver, -this module becomes a one-call shim onto that resolver. - -Manifest fields consumed: - * ``requires_permission`` — snake_case name (e.g. ``chisel_write``) - matching the Sprint-12 convention sample ``use_pebble_research``. -""" - -from __future__ import annotations - -import os -from dataclasses import dataclass - - -@dataclass(frozen=True) -class PermissionResult: - ok: bool - reason: str = "" - - -def _split_emails(raw: str) -> set[str]: - return {e.strip().lower() for e in raw.split(",") if e.strip()} - - -def _bypass_set() -> set[str]: - raw = os.environ.get("PEBBLE_CHISEL_RBAC_BYPASS_USERS", "").strip() - if raw: - return _split_emails(raw) - raw = os.environ.get("PEBBLE_CHAT_ALLOWED_EMAILS", "").strip() - return _split_emails(raw) - - -def check_permission( - *, - user_email: str, - required_permission: str | None, -) -> PermissionResult: - """Return PermissionResult(ok=True) when: - - * ``required_permission`` is None (tool/workflow doesn't gate); - * the user is in the bypass set; - - otherwise (Sprint-12 hook absent) PermissionResult(ok=False). - Once Sprint-12 lands, the else-branch defers to its resolver. - """ - if not required_permission: - return PermissionResult(ok=True) - - bypass = _bypass_set() - if user_email and user_email.lower() in bypass: - return PermissionResult(ok=True, reason="bypass_list") - - return PermissionResult( - ok=False, - reason=f"missing_permission: {required_permission}", - ) diff --git a/pebble/chisel/schema.py b/pebble/chisel/schema.py index 3a27a105..b0819025 100644 --- a/pebble/chisel/schema.py +++ b/pebble/chisel/schema.py @@ -1,17 +1,10 @@ """Pydantic → strict JSON Schema. -Pydantic's default ``model_json_schema()`` emits permissive object schemas -(no ``additionalProperties: false``). The Pebble planner historically uses -``make_input_schema(additional_properties=False)`` so the planner cannot -smuggle unknown args past tool handlers — see ``tasks/pebble-chisel-plan.md -§P1`` for the failure mode. - -This module post-processes the Pydantic-emitted schema: - - 1. Inlines ``$ref`` / ``$defs`` so the result is self-contained. - 2. Injects ``additionalProperties: false`` at every ``type: object`` node. - -Tests assert the invariant on every registered tool's schema. +Pydantic's default ``model_json_schema()`` emits permissive object +schemas (no ``additionalProperties: false``). The Pebble planner needs +strict schemas so it cannot smuggle unknown args past tool handlers +(plan §P1). This module inlines ``$ref`` / ``$defs`` and forces +``additionalProperties: false`` on every object node. """ from __future__ import annotations @@ -23,29 +16,14 @@ def pydantic_to_strict_schema(model: type[BaseModel]) -> dict[str, Any]: - """Emit a JSON Schema in the shape the Anthropic tool-use API consumes, - with ``additionalProperties: false`` enforced at every object node.""" + """Emit a JSON Schema in the shape the Anthropic tool-use API + consumes, with ``additionalProperties: false`` enforced everywhere.""" raw = model.model_json_schema(ref_template="#/$defs/{model}") inlined = _inline_refs(raw) - return _enforce_strict_objects(inlined) - - -def assert_strict(schema: dict[str, Any]) -> None: - """Invariant check: every ``type: object`` node has - ``additionalProperties: false``. Used in tests.""" - for node in _walk_object_nodes(schema): - ap = node.get("additionalProperties") - if ap is not False: - raise AssertionError( - f"object node missing additionalProperties:false (got {ap!r}): " - f"keys={sorted(node.keys())}", - ) + _enforce_strict_objects(inlined) + return inlined -# --------------------------------------------------------------------------- -# internal -# --------------------------------------------------------------------------- - def _inline_refs(schema: dict[str, Any]) -> dict[str, Any]: defs = schema.get("$defs", {}) out = deepcopy(schema) @@ -69,24 +47,14 @@ def resolve(node: Any) -> Any: return resolve(out) -def _enforce_strict_objects(schema: dict[str, Any]) -> dict[str, Any]: - out = deepcopy(schema) - for node in _walk_object_nodes(out): - node.setdefault("additionalProperties", False) - if node["additionalProperties"] is not False: - node["additionalProperties"] = False - return out - - -def _walk_object_nodes(schema: dict[str, Any]) -> list[dict[str, Any]]: - """Yield every dict node where ``type == 'object'``. Walks into - properties, items, anyOf/oneOf/allOf branches, and $defs.""" - found: list[dict[str, Any]] = [] +def _enforce_strict_objects(schema: dict[str, Any]) -> None: + """Mutate ``schema`` in place so every ``type: object`` node has + ``additionalProperties: false``.""" def visit(node: Any) -> None: if isinstance(node, dict): if node.get("type") == "object": - found.append(node) + node["additionalProperties"] = False for v in node.values(): visit(v) elif isinstance(node, list): @@ -94,4 +62,3 @@ def visit(node: Any) -> None: visit(v) visit(schema) - return found diff --git a/pebble/chisel/tools/aggregate_pipeline_views/canonical_queries.yaml b/pebble/chisel/tools/aggregate_pipeline_views/canonical_queries.yaml new file mode 100644 index 00000000..8a8e296d --- /dev/null +++ b/pebble/chisel/tools/aggregate_pipeline_views/canonical_queries.yaml @@ -0,0 +1,18 @@ +queries: + - id: smoke_nl_weekly_review + user_query: "Show me the weekly pipeline review" + expected_plan: + - tool: aggregate_pipeline_views + tags: [smoke] + + - id: smoke_nl_at_risk + user_query: "What deals are at risk this month?" + expected_plan: + - tool: aggregate_pipeline_views + tags: [smoke] + + - id: smoke_nl_pipeline_coverage + user_query: "Who's carrying the most pipeline?" + expected_plan: + - tool: aggregate_pipeline_views + tags: [smoke] diff --git a/pebble/chisel/tools/aggregate_pipeline_views/manifest.yaml b/pebble/chisel/tools/aggregate_pipeline_views/manifest.yaml index db9e121f..c5b4d04c 100644 --- a/pebble/chisel/tools/aggregate_pipeline_views/manifest.yaml +++ b/pebble/chisel/tools/aggregate_pipeline_views/manifest.yaml @@ -11,7 +11,4 @@ tags: - workflow - pipeline requires_human: false -cost_estimate: - fixed: 0.0 -output_kind: prose -scope: global +cost_estimate_usd: 0.0 diff --git a/pebble/chisel/tools/generate_chart/canonical_queries.yaml b/pebble/chisel/tools/generate_chart/canonical_queries.yaml new file mode 100644 index 00000000..0d36fe45 --- /dev/null +++ b/pebble/chisel/tools/generate_chart/canonical_queries.yaml @@ -0,0 +1,10 @@ +queries: + # generate_chart is downstream of an aggregation step; planner-only + # eval can't fully verify its use without an upstream tool result. + # Skipped at plan-eval time; gated CI runs the full pipeline. + - id: skipped_generate_chart_smoke + user_query: "Plot pipeline by owner as a bar chart" + skip_reason: "downstream of search/aggregate — requires full pipeline (gated CI)" + expected_plan: + - tool: generate_chart + tags: [smoke, deferred] diff --git a/pebble/chisel/tools/generate_chart/manifest.yaml b/pebble/chisel/tools/generate_chart/manifest.yaml index 83b333c3..63ae46c0 100644 --- a/pebble/chisel/tools/generate_chart/manifest.yaml +++ b/pebble/chisel/tools/generate_chart/manifest.yaml @@ -12,7 +12,4 @@ version: 1.0.0 tags: - rendering requires_human: false -cost_estimate: - fixed: 0.0 -output_kind: chart -scope: global +cost_estimate_usd: 0.0 diff --git a/pebble/chisel/tools/get_record/canonical_queries.yaml b/pebble/chisel/tools/get_record/canonical_queries.yaml new file mode 100644 index 00000000..dcfa49c6 --- /dev/null +++ b/pebble/chisel/tools/get_record/canonical_queries.yaml @@ -0,0 +1,9 @@ +queries: + - id: smoke_lookup_after_search + user_query: "Get the full Salesforce account record for ID 001ABC" + expected_plan: + - tool: get_record + args_includes: + entity_type: sf_account + entity_id: "001ABC" + tags: [smoke] diff --git a/pebble/chisel/tools/get_record/handler_test.py b/pebble/chisel/tools/get_record/handler_test.py index 7fd81900..65007600 100644 --- a/pebble/chisel/tools/get_record/handler_test.py +++ b/pebble/chisel/tools/get_record/handler_test.py @@ -50,7 +50,7 @@ async def test_happy_path_returns_record_with_citation() -> None: assert out["entity_type"] == "sf_account" assert out["entity_id"] == "001ABC" assert out["record"]["Name"] == "Acme" - assert hctx.collected_citations() == ("sf_account:001ABC",) + assert hctx.citations == ("sf_account:001ABC",) client.get.assert_awaited_once_with("/api/salesforce/accounts/001ABC") diff --git a/pebble/chisel/tools/get_record/manifest.yaml b/pebble/chisel/tools/get_record/manifest.yaml index d949e696..237838b2 100644 --- a/pebble/chisel/tools/get_record/manifest.yaml +++ b/pebble/chisel/tools/get_record/manifest.yaml @@ -7,7 +7,4 @@ version: 1.0.0 tags: - crm_read requires_human: false -cost_estimate: - fixed: 0.0 -output_kind: prose -scope: global +cost_estimate_usd: 0.0 diff --git a/pebble/chisel/tools/request_human_review/canonical_queries.yaml b/pebble/chisel/tools/request_human_review/canonical_queries.yaml new file mode 100644 index 00000000..cd17198d --- /dev/null +++ b/pebble/chisel/tools/request_human_review/canonical_queries.yaml @@ -0,0 +1,10 @@ +queries: + # request_human_review fires when the planner is uncertain — hard to + # provoke deterministically with a single query. Defer assertion to + # the gated CI run with the full pipeline. + - id: skipped_request_human_review_smoke + user_query: "Update Acme's owner — but there are two Acmes in the system" + skip_reason: "ambiguity-driven; deterministic provocation needs full pipeline" + expected_plan: + - tool: request_human_review + tags: [smoke, deferred] diff --git a/pebble/chisel/tools/request_human_review/handler_test.py b/pebble/chisel/tools/request_human_review/handler_test.py index 6c4d9384..a664e093 100644 --- a/pebble/chisel/tools/request_human_review/handler_test.py +++ b/pebble/chisel/tools/request_human_review/handler_test.py @@ -46,15 +46,12 @@ def test_handler_rejects_empty_reason() -> None: def test_autoload_registers_tool() -> None: """Behavioural parity: autoload discovers this manifest + handler, registers a working ToolSpec with strict schema and requires_human=True.""" - from pebble.chisel.schema import assert_strict - reg = ToolRegistry() root = Path(__file__).resolve().parents[3] / "chisel" - # Load only this tool by passing the chisel root and filtering after. report = autoload(registry=reg, root=root) assert "request_human_review" not in (e[0] for e in report.errors) spec = reg.get("request_human_review") assert spec is not None assert spec.requires_human is True - assert_strict(spec.input_schema) + assert spec.input_schema["additionalProperties"] is False assert "reason" in spec.input_schema["properties"] diff --git a/pebble/chisel/tools/request_human_review/manifest.yaml b/pebble/chisel/tools/request_human_review/manifest.yaml index a6eca80b..0e12ebf3 100644 --- a/pebble/chisel/tools/request_human_review/manifest.yaml +++ b/pebble/chisel/tools/request_human_review/manifest.yaml @@ -8,7 +8,4 @@ version: 1.0.0 tags: - control_flow requires_human: true -cost_estimate: - fixed: 0.0 -output_kind: checkpoint -scope: global +cost_estimate_usd: 0.0 diff --git a/pebble/chisel/tools/search_crm/canonical_queries.yaml b/pebble/chisel/tools/search_crm/canonical_queries.yaml new file mode 100644 index 00000000..0c7d0b80 --- /dev/null +++ b/pebble/chisel/tools/search_crm/canonical_queries.yaml @@ -0,0 +1,14 @@ +queries: + - id: smoke_find_by_name + user_query: "Find Acme Corp in our records" + expected_plan: + - tool: search_crm + args_includes: + query: "Acme Corp" + tags: [smoke] + + - id: smoke_find_by_email + user_query: "Search for jane@example.com" + expected_plan: + - tool: search_crm + tags: [smoke] diff --git a/pebble/chisel/tools/search_crm/handler_test.py b/pebble/chisel/tools/search_crm/handler_test.py index 7891da60..52004c47 100644 --- a/pebble/chisel/tools/search_crm/handler_test.py +++ b/pebble/chisel/tools/search_crm/handler_test.py @@ -57,7 +57,7 @@ async def test_happy_path_collects_citations_and_passes_params() -> None: assert out["total_count"] == 2 assert out["backend_used"] == "postgres_fts" assert out["query"] == "acme" - assert hctx.collected_citations() == ("sf_account:001A", "sf_account:001B") + assert hctx.citations == ("sf_account:001A", "sf_account:001B") client.get.assert_awaited_once_with( "/api/search", params={"q": "acme", "limit": 10, "types": "sf_account"}, ) diff --git a/pebble/chisel/tools/search_crm/manifest.yaml b/pebble/chisel/tools/search_crm/manifest.yaml index f6b02456..4e226603 100644 --- a/pebble/chisel/tools/search_crm/manifest.yaml +++ b/pebble/chisel/tools/search_crm/manifest.yaml @@ -10,7 +10,4 @@ version: 1.0.0 tags: - crm_read requires_human: false -cost_estimate: - fixed: 0.0 -output_kind: prose -scope: global +cost_estimate_usd: 0.0 diff --git a/pebble/chisel/workflows/weekly_pipeline_review/canonical_queries.yaml b/pebble/chisel/workflows/weekly_pipeline_review/canonical_queries.yaml new file mode 100644 index 00000000..5295051a --- /dev/null +++ b/pebble/chisel/workflows/weekly_pipeline_review/canonical_queries.yaml @@ -0,0 +1,14 @@ +queries: + # Slash-command path bypasses the planner — build_plan.py is + # deterministic and unit-tested. These queries cover the NL path + # where the planner picks aggregate_pipeline_views directly. + - id: smoke_slash_pipeline + user_query: "/pipeline" + skip_reason: "slash commands bypass the planner; deterministic plan is unit-tested separately" + tags: [smoke, slash, deferred] + + - id: smoke_nl_pipeline_review + user_query: "Run the weekly pipeline review for me" + expected_plan: + - tool: aggregate_pipeline_views + tags: [smoke] diff --git a/pebble/chisel/workflows/weekly_pipeline_review/workflow.yaml b/pebble/chisel/workflows/weekly_pipeline_review/workflow.yaml index 18f65ab2..9644de13 100644 --- a/pebble/chisel/workflows/weekly_pipeline_review/workflow.yaml +++ b/pebble/chisel/workflows/weekly_pipeline_review/workflow.yaml @@ -4,8 +4,5 @@ description: | command or the planner picking aggregate_pipeline_views directly. version: 1.0.0 slash_command: /pipeline -dispatch_intent: workflow_weekly_pipeline_review has_custom_plan: true -cost_estimate: - fixed: 0.0 -scope: global +cost_estimate_usd: 0.0 diff --git a/pebble/handlers/streaming.py b/pebble/handlers/streaming.py index 1c722fba..49254936 100644 --- a/pebble/handlers/streaming.py +++ b/pebble/handlers/streaming.py @@ -59,17 +59,11 @@ from ..orchestrator.planner import Planner from ..orchestrator.scratchpad import ScratchpadWriter from ..orchestrator.tools import DEFAULT_REGISTRY, ToolContext -# Chisel autoload — discovers ``pebble/chisel/{tools,workflows}/`` and -# registers each unit on DEFAULT_REGISTRY. Replaces the legacy -# side-effect imports of orchestrator.builtin_tools + workflows that -# used to auto-register at module load. Errors surface in the report -# rather than crashing — the app starts with whatever loaded. +# Importing ``pebble.chisel`` runs ``autoload()`` once at module load, +# registering every chisel tool/workflow on DEFAULT_REGISTRY. Errors +# flow through the boot report and get logged; the process boots with +# whatever loaded. from .. import chisel as _chisel -_chisel_autoload_report = _chisel.autoload() -if not _chisel_autoload_report.ok(): - logging.getLogger(__name__).warning( - "chisel.autoload had errors: %s", _chisel_autoload_report.errors, - ) from ..router import RouteResult logger = logging.getLogger(__name__) @@ -254,4 +248,7 @@ def _build_workflow_plan_for_intent(intent: str, user_query: str): """Look up the chisel workflow registered for ``intent`` and call its build_plan. Returns None if no workflow matches — caller surfaces a clean error to the user.""" - return _chisel.build_workflow_plan(intent, user_query=user_query) + entry = _chisel.lookup_intent(intent) + if entry is None: + return None + return entry.build_plan(user_query=user_query) diff --git a/pebble/router.py b/pebble/router.py index 65d9a64e..8919e1b8 100644 --- a/pebble/router.py +++ b/pebble/router.py @@ -87,11 +87,10 @@ def _check_redirect(query: str) -> RouteResult | None: # Slash commands resolve to Chisel-registered workflows. Adding a # workflow = drop a ``workflow.yaml`` under ``pebble/chisel/workflows/`` -# declaring ``slash_command:`` and ``dispatch_intent:``. ``autoload()`` -# populates ``chisel.slash_to_intent()`` so this router picks it up at -# request time without per-workflow router changes. +# declaring ``slash_command:``. ``chisel.lookup_slash()`` picks it up +# at request time without per-workflow router changes. -from . import chisel as _chisel # populated by streaming.py's autoload() +from . import chisel as _chisel # autoload runs on import def _check_slash_command(query: str) -> RouteResult | None: @@ -109,11 +108,11 @@ def _check_slash_command(query: str) -> RouteResult | None: return None head, _, rest = stripped.partition(" ") head_lower = head.lower() - intent = _chisel.slash_to_intent(head_lower) - if intent is None: + entry = _chisel.lookup_slash(head_lower) + if entry is None: return None return RouteResult( - level=2, intent=intent, confidence=1.0, + level=2, intent=entry.dispatch_intent, confidence=1.0, entities={"slash_command": head_lower, "args": rest.strip()}, ) diff --git a/pebble/tests/conftest.py b/pebble/tests/conftest.py index f8b7b57c..e7228ddc 100644 --- a/pebble/tests/conftest.py +++ b/pebble/tests/conftest.py @@ -12,18 +12,12 @@ @pytest.fixture(autouse=True) def _chisel_real_autoload(): - """Ensure the chisel module-level slash / intent / plan-builder maps - point at the real ``pebble/chisel/`` tree at the start of every test. - - Framework tests in ``test_chisel_framework.py`` call ``autoload(root= - tmp_path)`` which resets those maps to the temp tree's contents; that - leaves later tests reading from stale state. Re-running autoload - before each test (and after) keeps router / streaming tests reading - the real ``/pipeline`` workflow.""" + """Restore the chisel workflow maps to the real ``pebble/chisel/`` + tree before each test. Framework tests in ``test_chisel_framework.py`` + call ``autoload(root=tmp_path)`` which resets the maps to a temp + tree — without this fixture, the next test reads stale state.""" from pebble import chisel chisel.autoload() - yield - chisel.autoload() @pytest.fixture diff --git a/pebble/tests/test_chisel_eval.py b/pebble/tests/test_chisel_eval.py new file mode 100644 index 00000000..6bb24bd2 --- /dev/null +++ b/pebble/tests/test_chisel_eval.py @@ -0,0 +1,321 @@ +"""Tests for the chisel eval harness. + +Covers Phase B.4 surface: + * canonical_queries.yaml schema (Pydantic validation) + * assert_plan: happy + step-count + tool-mismatch + arg-includes/excludes + * assert_prose: substring includes / excludes + * loader: discovery across tools + workflows, malformed file errors out + * run_plan_eval: stub planner happy path, planner-error path, skip path + * format_results: pass / fail / skip rendering +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path +from textwrap import dedent +from uuid import uuid4 + +import pytest +from pydantic import ValidationError + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) + +from pebble.chisel.eval import ( + CanonicalQueriesFile, + CanonicalQuery, + EvalResult, + ExpectedProse, + ExpectedStep, + LoadedQuery, + assert_plan, + assert_prose, + format_results, + load_canonical_queries, + run_plan_eval, +) +from pebble.orchestrator.planner import Planner, PlannerLLMResponse +from pebble.orchestrator.schemas import Plan, PlanStep +from pebble.orchestrator.tools import ( + ToolContext, + ToolRegistry, + ToolSpec, + make_input_schema, +) + + +# --------------------------------------------------------------------------- +# Schema +# --------------------------------------------------------------------------- + +def test_canonical_query_minimal_valid() -> None: + q = CanonicalQuery(id="smoke_x", user_query="Find Acme") + assert q.expected_plan == () + assert q.expected_prose is None + + +def test_canonical_query_rejects_bad_id() -> None: + with pytest.raises(ValidationError): + CanonicalQuery(id="BadID", user_query="x") + with pytest.raises(ValidationError): + CanonicalQuery(id="1bad", user_query="x") + + +def test_canonical_query_rejects_unknown_field() -> None: + with pytest.raises(ValidationError): + CanonicalQuery(id="ok_id", user_query="x", typo=1) + + +def test_expected_step_rejects_empty_tool() -> None: + with pytest.raises(ValidationError): + ExpectedStep(tool="") + + +# --------------------------------------------------------------------------- +# assert_plan +# --------------------------------------------------------------------------- + +def _plan(*steps: PlanStep) -> Plan: + return Plan(user_query="q", steps=tuple(steps)) + + +def test_assert_plan_empty_expected_always_passes() -> None: + plan = _plan(PlanStep(tool="search_crm", args={"query": "x"})) + assert assert_plan(plan, ()) == [] + + +def test_assert_plan_happy_path_with_args_includes() -> None: + plan = _plan(PlanStep(tool="search_crm", args={"query": "Acme", "limit": 8})) + expected = (ExpectedStep(tool="search_crm", args_includes={"query": "Acme"}),) + assert assert_plan(plan, expected) == [] + + +def test_assert_plan_flags_step_count_mismatch() -> None: + plan = _plan(PlanStep(tool="search_crm", args={"query": "x"})) + expected = ( + ExpectedStep(tool="search_crm"), + ExpectedStep(tool="get_record"), + ) + failures = assert_plan(plan, expected) + assert any("step_count" in f for f in failures) + + +def test_assert_plan_flags_wrong_tool() -> None: + plan = _plan(PlanStep(tool="get_record", args={"entity_id": "x"})) + expected = (ExpectedStep(tool="search_crm"),) + failures = assert_plan(plan, expected) + assert any("step[0].tool" in f for f in failures) + + +def test_assert_plan_flags_missing_arg() -> None: + plan = _plan(PlanStep(tool="search_crm", args={"limit": 10})) + expected = (ExpectedStep(tool="search_crm", args_includes={"query": "Acme"}),) + failures = assert_plan(plan, expected) + assert any("missing key 'query'" in f for f in failures) + + +def test_assert_plan_flags_arg_value_mismatch() -> None: + plan = _plan(PlanStep(tool="search_crm", args={"query": "Beta"})) + expected = (ExpectedStep(tool="search_crm", args_includes={"query": "Acme"}),) + failures = assert_plan(plan, expected) + assert any("expected 'Acme', got 'Beta'" in f for f in failures) + + +def test_assert_plan_flags_forbidden_arg() -> None: + plan = _plan(PlanStep(tool="search_crm", args={"query": "x", "secret": "leak"})) + expected = (ExpectedStep(tool="search_crm", args_excludes=("secret",)),) + failures = assert_plan(plan, expected) + assert any("forbidden key 'secret'" in f for f in failures) + + +def test_assert_plan_tolerates_extra_steps() -> None: + plan = _plan( + PlanStep(tool="search_crm", args={"query": "x"}), + PlanStep(tool="get_record", args={"entity_type": "sf_account", "entity_id": "1"}), + ) + expected = (ExpectedStep(tool="search_crm"),) + assert assert_plan(plan, expected) == [] + + +# --------------------------------------------------------------------------- +# assert_prose +# --------------------------------------------------------------------------- + +def test_assert_prose_includes_pass_and_fail() -> None: + text = "Pipeline coverage: Alice 100, Bob 200." + assert assert_prose(text, ExpectedProse(includes=("Alice", "Bob"))) == [] + failures = assert_prose(text, ExpectedProse(includes=("Carol",))) + assert any("missing substring 'Carol'" in f for f in failures) + + +def test_assert_prose_excludes_pass_and_fail() -> None: + text = "Pipeline coverage clean." + assert assert_prose(text, ExpectedProse(excludes=("error",))) == [] + failures = assert_prose("System error encountered.", ExpectedProse(excludes=("error",))) + assert any("forbidden substring 'error'" in f for f in failures) + + +# --------------------------------------------------------------------------- +# Loader +# --------------------------------------------------------------------------- + +def test_load_canonical_queries_real_chisel_root() -> None: + """Discovery walks pebble/chisel/ — must find the fixtures we + shipped alongside the migrated tools.""" + loaded = load_canonical_queries() + units = {lq.unit for lq in loaded} + assert "search_crm" in units + assert "aggregate_pipeline_views" in units + assert "weekly_pipeline_review" in units + + +def test_load_canonical_queries_rejects_malformed(tmp_path: Path) -> None: + """Malformed canonical_queries.yaml raises with the file path in + the error message — easier than chasing nested Pydantic errors.""" + (tmp_path / "tools" / "broken_tool").mkdir(parents=True) + (tmp_path / "tools" / "broken_tool" / "canonical_queries.yaml").write_text( + dedent( + """ + queries: + - id: BadID + user_query: x + """ + ).strip(), + encoding="utf-8", + ) + with pytest.raises(ValueError, match="canonical_queries.yaml"): + load_canonical_queries(chisel_root=tmp_path) + + +def test_canonical_queries_file_round_trip() -> None: + raw = { + "queries": [ + { + "id": "smoke_x", + "user_query": "Find Acme", + "expected_plan": [ + {"tool": "search_crm", "args_includes": {"query": "Acme"}}, + ], + "expected_prose": {"includes": ["Acme"], "excludes": ["error"]}, + "tags": ["smoke"], + }, + ], + } + parsed = CanonicalQueriesFile(**raw) + assert parsed.queries[0].expected_plan[0].args_includes == {"query": "Acme"} + assert parsed.queries[0].expected_prose.includes == ("Acme",) + + +# --------------------------------------------------------------------------- +# run_plan_eval with stub planner +# --------------------------------------------------------------------------- + +class _StubLLMClient: + """Returns a hand-canned plan-JSON string.""" + def __init__(self, plan_json: str): + self._plan_json = plan_json + + async def emit_plan(self, *, system, user, tools, max_tokens=2048): + return PlannerLLMResponse(text=self._plan_json) + + +def _registry_with_search() -> ToolRegistry: + reg = ToolRegistry() + async def handler(args, ctx): + from pebble.orchestrator.schemas import ToolResult + return ToolResult(step_id=uuid4(), tool="search_crm", ok=True) + reg.register( + ToolSpec( + name="search_crm", + description="x", + input_schema=make_input_schema( + properties={"query": {"type": "string"}}, + required_keys=["query"], + ), + handler=handler, + ) + ) + return reg + + +def _loaded(query: CanonicalQuery, tmp_path: Path) -> LoadedQuery: + return LoadedQuery(source=tmp_path / "x.yaml", unit="search_crm", query=query) + + +@pytest.mark.asyncio +async def test_run_plan_eval_pass(tmp_path: Path) -> None: + plan_json = ( + '{"rationale":"r","estimated_cost_usd":0,"estimated_tool_calls":1,' + '"steps":[{"tool":"search_crm","args":{"query":"Acme"},' + '"expected_shape":"","success_criteria":""}]}' + ) + reg = _registry_with_search() + planner = Planner(client=_StubLLMClient(plan_json), registry=reg) + q = CanonicalQuery( + id="smoke_x", user_query="Find Acme", + expected_plan=(ExpectedStep(tool="search_crm", args_includes={"query": "Acme"}),), + ) + res = await run_plan_eval( + _loaded(q, tmp_path), + planner=planner, + ctx=ToolContext(user_email="t@x", conversation_id="c1"), + ) + assert res.passed and not res.plan_failures + + +@pytest.mark.asyncio +async def test_run_plan_eval_planner_error(tmp_path: Path) -> None: + reg = _registry_with_search() + planner = Planner(client=_StubLLMClient("not json"), registry=reg) + q = CanonicalQuery(id="smoke_x", user_query="Find Acme", + expected_plan=(ExpectedStep(tool="search_crm"),)) + res = await run_plan_eval( + _loaded(q, tmp_path), + planner=planner, + ctx=ToolContext(user_email="t@x", conversation_id="c1"), + ) + assert not res.passed + assert res.planner_error is not None + + +@pytest.mark.asyncio +async def test_run_plan_eval_skip(tmp_path: Path) -> None: + reg = _registry_with_search() + planner = Planner(client=_StubLLMClient(""), registry=reg) + q = CanonicalQuery( + id="smoke_x", user_query="Find Acme", + skip_reason="needs full pipeline", + ) + res = await run_plan_eval( + _loaded(q, tmp_path), + planner=planner, + ctx=ToolContext(user_email="t@x", conversation_id="c1"), + ) + assert res.skipped and res.passed + assert res.skip_reason == "needs full pipeline" + + +# --------------------------------------------------------------------------- +# format_results +# --------------------------------------------------------------------------- + +def test_format_results_summary(tmp_path: Path) -> None: + results = [ + EvalResult(query_id="a", unit="u", source=tmp_path / "x", passed=True, duration_ms=10), + EvalResult( + query_id="b", unit="u", source=tmp_path / "x", + passed=False, plan_failures=["step[0].tool: expected 'x', got 'y'"], + duration_ms=12, + ), + EvalResult( + query_id="c", unit="u", source=tmp_path / "x", + passed=True, skipped=True, skip_reason="deferred", + ), + ] + out = format_results(results) + assert "1/3 passed" in out + assert "PASS u/a" in out + assert "FAIL u/b" in out + assert "SKIP u/c — deferred" in out + assert "step[0].tool" in out diff --git a/pebble/tests/test_chisel_framework.py b/pebble/tests/test_chisel_framework.py index 773c8fd3..0fd653be 100644 --- a/pebble/tests/test_chisel_framework.py +++ b/pebble/tests/test_chisel_framework.py @@ -1,15 +1,14 @@ -"""Phase A.1 unit tests for the Chisel framework. +"""Unit tests for the Chisel framework. -Covers, per plan §8: - - * manifest schema validation - * Pydantic→strict JSON Schema (P1 — additionalProperties:false at every object node) +Covers: + * manifest schema validation (ToolManifest + WorkflowManifest) + * Pydantic→strict JSON Schema (additionalProperties:false everywhere) * handler adapter happy / validation-error / exception / wrong-return-type * autoload: empty dirs, malformed manifest doesn't poison siblings, - registry= argument honored (P4), idempotent across calls - * RBAC stub: bypass env, missing perm, no requirement - * lints: bare httpx, sync run, os.environ in run, overrides - * snapshot: survives in-flight mutation of source registry (P5) + registry= argument honored, lint warnings surface in the report + * workflow dispatch: slash + intent lookup, declarative + custom builders + * lints: bare httpx, sync/missing run + * snapshot: survives in-flight mutation of source registry """ from __future__ import annotations @@ -24,19 +23,12 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) -from pebble.chisel.autoload import autoload +from pebble.chisel.autoload import autoload, lookup_intent, lookup_slash from pebble.chisel.handler_adapter import HandlerContext, build_handler_wrapper from pebble.chisel.lints import lint_handler_module -from pebble.chisel.manifest import ( - FixedCost, - ToolManifest, - VariableCost, - WorkflowManifest, - cost_estimate_to_float, -) -from pebble.chisel.rbac import check_permission +from pebble.chisel.manifest import ToolManifest, WorkflowManifest from pebble.chisel.reload import snapshot -from pebble.chisel.schema import assert_strict, pydantic_to_strict_schema +from pebble.chisel.schema import pydantic_to_strict_schema from pebble.orchestrator.tools import ToolContext, ToolRegistry, ToolSpec @@ -47,9 +39,8 @@ def test_tool_manifest_minimal_valid() -> None: m = ToolManifest(name="search_crm", description="x") assert m.version == "1.0.0" - assert isinstance(m.cost_estimate, FixedCost) - assert m.output_kind == "prose" - assert m.scope == "global" + assert m.cost_estimate_usd == 0.0 + assert m.requires_human is False def test_tool_manifest_rejects_bad_name() -> None: @@ -64,15 +55,6 @@ def test_tool_manifest_rejects_unknown_field() -> None: ToolManifest(name="ok_name", description="x", typo_field=1) -def test_cost_variable_collapses_to_max() -> None: - m = ToolManifest( - name="llm_tool", - description="x", - cost_estimate=VariableCost(variable={"max": 0.5}), - ) - assert cost_estimate_to_float(m.cost_estimate) == pytest.approx(0.5) - - def test_workflow_requires_steps_or_custom_plan() -> None: with pytest.raises(ValidationError): WorkflowManifest(name="wf", description="x") @@ -90,8 +72,20 @@ def test_workflow_slash_command_format() -> None: WorkflowManifest(name="wf", description="x", has_custom_plan=True, slash_command="/pipeline") +def test_workflow_dispatch_intent_auto_filled() -> None: + """Convention: dispatch_intent defaults to workflow_.""" + wf = WorkflowManifest(name="foo", description="x", has_custom_plan=True) + assert wf.dispatch_intent == "workflow_foo" + + explicit = WorkflowManifest( + name="bar", description="x", has_custom_plan=True, + dispatch_intent="custom_intent", + ) + assert explicit.dispatch_intent == "custom_intent" + + # --------------------------------------------------------------------------- -# schema strictness (P1) +# schema strictness # --------------------------------------------------------------------------- class _Inner(BaseModel): @@ -104,9 +98,25 @@ class _NestedInput(BaseModel): tags: list[str] = [] +def _walk_object_nodes(schema): + out = [] + def visit(node): + if isinstance(node, dict): + if node.get("type") == "object": + out.append(node) + for v in node.values(): + visit(v) + elif isinstance(node, list): + for v in node: + visit(v) + visit(schema) + return out + + def test_strict_schema_has_no_permissive_objects() -> None: schema = pydantic_to_strict_schema(_NestedInput) - assert_strict(schema) + for node in _walk_object_nodes(schema): + assert node["additionalProperties"] is False def test_strict_schema_inlines_refs() -> None: @@ -117,14 +127,8 @@ def test_strict_schema_inlines_refs() -> None: assert inner_prop["additionalProperties"] is False -def test_assert_strict_flags_permissive_object() -> None: - bad = {"type": "object", "properties": {"x": {"type": "integer"}}} - with pytest.raises(AssertionError): - assert_strict(bad) - - # --------------------------------------------------------------------------- -# handler adapter (P2, P11) +# handler adapter # --------------------------------------------------------------------------- class _AdapterInput(BaseModel): @@ -138,7 +142,7 @@ def _make_ctx() -> ToolContext: @pytest.mark.asyncio async def test_adapter_happy_path_records_version_and_citations() -> None: - async def run(args: _AdapterInput, ctx: HandlerContext) -> dict: + async def run(args, ctx): ctx.cite("sf_account", "001ABC") return {"hits": 1, "query": args.query} @@ -156,7 +160,7 @@ async def run(args: _AdapterInput, ctx: HandlerContext) -> dict: @pytest.mark.asyncio async def test_adapter_input_validation_returns_ok_false() -> None: - async def run(args: _AdapterInput, ctx: HandlerContext) -> dict: + async def run(args, ctx): return {} wrapped = build_handler_wrapper( @@ -171,7 +175,7 @@ async def run(args: _AdapterInput, ctx: HandlerContext) -> dict: @pytest.mark.asyncio async def test_adapter_handler_exception_wrapped() -> None: - async def run(args: _AdapterInput, ctx: HandlerContext) -> dict: + async def run(args, ctx): raise RuntimeError("boom") wrapped = build_handler_wrapper( @@ -185,8 +189,8 @@ async def run(args: _AdapterInput, ctx: HandlerContext) -> dict: @pytest.mark.asyncio async def test_adapter_non_dict_return_rejected() -> None: - async def run(args: _AdapterInput, ctx: HandlerContext) -> dict: - return ["wrong shape"] # type: ignore[return-value] + async def run(args, ctx): + return ["wrong shape"] wrapped = build_handler_wrapper( tool_name="t", tool_version="1.0.0", @@ -197,8 +201,19 @@ async def run(args: _AdapterInput, ctx: HandlerContext) -> dict: assert "handler_contract" in (res.error or "") +@pytest.mark.asyncio +async def test_handler_context_forwards_to_toolcontext() -> None: + """HandlerContext exposes ToolContext attributes via __getattr__.""" + ctx = ToolContext(user_email="rm@pursuit.org", conversation_id="c1", org_id="acme") + hctx = HandlerContext(ctx) + assert hctx.user_email == "rm@pursuit.org" + assert hctx.conversation_id == "c1" + assert hctx.org_id == "acme" + assert hctx.http_client is None + + # --------------------------------------------------------------------------- -# autoload (P4 — isolation, robustness) +# autoload # --------------------------------------------------------------------------- def _write_tool(root: Path, name: str, *, manifest_yaml: str, handler_py: str) -> None: @@ -247,16 +262,14 @@ def test_autoload_loads_tool_into_isolated_registry(tmp_path: Path) -> None: reg = ToolRegistry() report = autoload(registry=reg, root=tmp_path) assert report.loaded_tools == ["alpha"] - assert "alpha" in reg spec = reg.get("alpha") assert isinstance(spec, ToolSpec) - # input_schema must be strict - assert_strict(spec.input_schema) + for node in _walk_object_nodes(spec.input_schema): + assert node["additionalProperties"] is False def test_autoload_isolates_malformed_from_siblings(tmp_path: Path) -> None: _write_tool(tmp_path, "good", manifest_yaml=_ok_manifest("good"), handler_py=_ok_handler("good")) - # Malformed: invalid name _write_tool( tmp_path, "bad", manifest_yaml="name: BadName\ndescription: x", @@ -271,7 +284,7 @@ def test_autoload_isolates_malformed_from_siblings(tmp_path: Path) -> None: assert "bad" not in reg -def test_autoload_workflow_declarative_populates_slash_intent_and_builder(tmp_path: Path) -> None: +def test_autoload_workflow_declarative_populates_lookup_and_builder(tmp_path: Path) -> None: wf = tmp_path / "workflows" / "demo_wf" wf.mkdir(parents=True) (wf / "workflow.yaml").write_text( @@ -280,7 +293,6 @@ def test_autoload_workflow_declarative_populates_slash_intent_and_builder(tmp_pa name: demo_wf description: declarative demo slash_command: /demo - dispatch_intent: workflow_demo steps: - tool: aggregate_pipeline_views args: {days_to_close: 30} @@ -291,20 +303,18 @@ def test_autoload_workflow_declarative_populates_slash_intent_and_builder(tmp_pa ) reg = ToolRegistry() report = autoload(registry=reg, root=tmp_path) - from pebble.chisel.autoload import ( - build_workflow_plan, - dispatch_workflow, - slash_command_map, - ) - assert "demo_wf" in report.loaded_workflows - assert slash_command_map().get("/demo") == "demo_wf" - assert dispatch_workflow("workflow_demo") == "demo_wf" - plan = build_workflow_plan("workflow_demo", user_query="hi") - assert plan is not None + entry = lookup_slash("/demo") + assert entry is not None + assert entry.name == "demo_wf" + assert entry.dispatch_intent == "workflow_demo_wf" # auto-filled + + via_intent = lookup_intent("workflow_demo_wf") + assert via_intent is entry + + plan = entry.build_plan(user_query="hi") assert plan.user_query == "hi" - assert len(plan.steps) == 1 assert plan.steps[0].tool == "aggregate_pipeline_views" @@ -318,7 +328,6 @@ def test_autoload_workflow_custom_build_plan(tmp_path: Path) -> None: name: custom_wf description: custom-plan demo slash_command: /custom - dispatch_intent: workflow_custom has_custom_plan: true """ ).strip(), @@ -340,11 +349,11 @@ def build_plan(*, user_query="custom", multiplier=1, **_): ) reg = ToolRegistry() report = autoload(registry=reg, root=tmp_path) - from pebble.chisel.autoload import build_workflow_plan - assert "custom_wf" in report.loaded_workflows, report.errors - plan = build_workflow_plan("workflow_custom", user_query="hi", multiplier=3) - assert plan is not None + + entry = lookup_intent("workflow_custom_wf") + assert entry is not None + plan = entry.build_plan(user_query="hi", multiplier=3) assert plan.steps[0].args == {"q": "xxx"} @@ -367,6 +376,26 @@ def test_autoload_workflow_missing_build_plan_reports_error(tmp_path: Path) -> N assert any("build_plan" in reason for _, reason in report.errors) +def test_autoload_lint_warnings_surface_in_report(tmp_path: Path) -> None: + """A handler with a bare httpx import still registers, but the + lint warning shows up in the report so CI/log readers see it.""" + handler_with_lint = dedent( + """ + import httpx + from pydantic import BaseModel + class Input(BaseModel): + q: str = "default" + async def run(args, ctx): + return {"q": args.q} + """ + ).strip() + _write_tool(tmp_path, "lintme", manifest_yaml=_ok_manifest("lintme"), handler_py=handler_with_lint) + reg = ToolRegistry() + report = autoload(registry=reg, root=tmp_path) + assert "lintme" in report.loaded_tools # registered despite lint hit + assert any("no_bare_httpx" in msg for _, msg in report.lint_warnings) + + def test_autoload_isolation_does_not_touch_default_registry(tmp_path: Path) -> None: _write_tool(tmp_path, "iso_only", manifest_yaml=_ok_manifest("iso_only"), handler_py=_ok_handler("iso_only")) reg = ToolRegistry() @@ -375,37 +404,6 @@ def test_autoload_isolation_does_not_touch_default_registry(tmp_path: Path) -> N assert "iso_only" not in DEFAULT_REGISTRY -# --------------------------------------------------------------------------- -# RBAC stub (§11.9) -# --------------------------------------------------------------------------- - -def test_rbac_no_requirement_always_ok() -> None: - res = check_permission(user_email="anyone@x", required_permission=None) - assert res.ok - - -def test_rbac_bypass_env(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setenv("PEBBLE_CHISEL_RBAC_BYPASS_USERS", "rm@pursuit.org , staff@pursuit.org") - res = check_permission(user_email="RM@pursuit.org", required_permission="chisel_write") - assert res.ok - assert res.reason == "bypass_list" - - -def test_rbac_missing_permission(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.delenv("PEBBLE_CHISEL_RBAC_BYPASS_USERS", raising=False) - monkeypatch.delenv("PEBBLE_CHAT_ALLOWED_EMAILS", raising=False) - res = check_permission(user_email="x@y", required_permission="chisel_write") - assert not res.ok - assert "chisel_write" in res.reason - - -def test_rbac_falls_back_to_chat_allowed_emails(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.delenv("PEBBLE_CHISEL_RBAC_BYPASS_USERS", raising=False) - monkeypatch.setenv("PEBBLE_CHAT_ALLOWED_EMAILS", "fallback@pursuit.org") - res = check_permission(user_email="fallback@pursuit.org", required_permission="chisel_write") - assert res.ok - - # --------------------------------------------------------------------------- # lints # --------------------------------------------------------------------------- @@ -441,40 +439,15 @@ def run(args, ctx): assert "async_run_required" in rules -def test_lints_flag_env_in_run(tmp_path: Path) -> None: +def test_lints_flag_missing_run(tmp_path: Path) -> None: p = tmp_path / "handler.py" - p.write_text( - dedent( - """ - import os - async def run(args, ctx): - return {"v": os.environ.get("X")} - """ - ), - encoding="utf-8", - ) + p.write_text("x = 1\n", encoding="utf-8") rules = {e.rule for e in lint_handler_module(p)} - assert "no_env_in_run" in rules - - -def test_lints_overrides_suppress(tmp_path: Path) -> None: - p = tmp_path / "handler.py" - p.write_text( - dedent( - """ - import httpx - async def run(args, ctx): - return {} - """ - ), - encoding="utf-8", - ) - rules = {e.rule for e in lint_handler_module(p, overrides=("no_bare_httpx",))} - assert "no_bare_httpx" not in rules + assert "async_run_required" in rules # --------------------------------------------------------------------------- -# snapshot (P5) +# snapshot # --------------------------------------------------------------------------- @pytest.mark.asyncio @@ -496,7 +469,6 @@ async def handler(args, ctx): ) snap = snapshot(source) - # Mutate source — simulate reload mid-request. source.unregister("x") source.register( ToolSpec( diff --git a/pebble/tests/test_router.py b/pebble/tests/test_router.py index f5494378..502e4b2f 100644 --- a/pebble/tests/test_router.py +++ b/pebble/tests/test_router.py @@ -190,5 +190,7 @@ async def test_slash_command_overrides_mode_classifier(self): def test_slash_commands_table_has_pipeline(self): """Smoke test: chisel autoload registers the /pipeline slash.""" - assert _chisel.slash_command_map().get("/pipeline") == "weekly_pipeline_review" - assert _chisel.slash_to_intent("/pipeline") == "workflow_weekly_pipeline_review" + entry = _chisel.lookup_slash("/pipeline") + assert entry is not None + assert entry.name == "weekly_pipeline_review" + assert entry.dispatch_intent == "workflow_weekly_pipeline_review"