From ef86a9a8a9d7f74d7290f0e7c66c0c632fd9aacc Mon Sep 17 00:00:00 2001 From: Ruslan Strazhnyk Date: Fri, 10 Jul 2026 08:30:37 +0200 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20expansion=20scope=20=E2=80=94=20cro?= =?UTF-8?q?ss-framework=20heal,=20MCP=20server,=20watch=20mode,=20heal=20h?= =?UTF-8?q?istory=20(v0.2.0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four expansion features that widen 9lives beyond a Playwright-only CI tool: * Cross-framework heal (QUA-1272): framework adapter layer under src/ninelives/frameworks/. Cypress specs run via 'npx cypress run --reporter json --quiet' in the user's own project; Selenium (Python) specs run via the user's own pytest. Auto-detected per spec (.cy.* / .py / package.json deps), overridable with --framework. Cypress locator misses (wrapped in AssertionError) are classified as selector drift so the behavior-vs-drift guard doesn't eat them; Selenium's NoSuchElementException JSON payload feeds selector extraction. Tier 1 never injects Playwright waits into non-Playwright code; Tier 2 prompts are framework-labelled. * MCP server (QUA-1271): '9l mcp' — zero-dependency stdio JSON-RPC implementing initialize / tools/list / tools/call with heal_test and run_test, so Claude Code / Cursor / Codex heal red tests in-loop. Heal-loop output is rerouted to stderr to keep the protocol stream clean; non-interactive mode saves a .healed copy instead of prompting. * 9l watch + pre-commit hooks: poll-based heal-on-save watcher (no OS watcher deps) and .pre-commit-hooks.yaml with 9lives-heal / 9lives-run. * Heal history + 9l report: every heal attempt appends to .9lives/history.jsonl (local only); '9l report' aggregates it into a brittle-selector report with anchor stats and pin-a-testid recommendations, terminal or --md. Also: sync __version__ (was stuck at 0.1.0 since first release — '9l --version' lied) and bump to 0.2.0. 48 tests green (20 new), ruff clean. --- .pre-commit-hooks.yaml | 25 +++ README.md | 60 ++++++ pyproject.toml | 4 +- src/ninelives/__init__.py | 2 +- src/ninelives/cli.py | 209 +++++++++++++++---- src/ninelives/frameworks/__init__.py | 61 ++++++ src/ninelives/frameworks/cypress.py | 107 ++++++++++ src/ninelives/frameworks/playwright.py | 23 +++ src/ninelives/frameworks/selenium.py | 106 ++++++++++ src/ninelives/healing/parse.py | 29 ++- src/ninelives/healing/strategy.py | 18 ++ src/ninelives/healing/tier1.py | 7 +- src/ninelives/healing/tier2.py | 20 +- src/ninelives/history.py | 138 +++++++++++++ src/ninelives/mcp_server.py | 202 ++++++++++++++++++ src/ninelives/watch.py | 92 +++++++++ tests/test_expansion.py | 276 +++++++++++++++++++++++++ tests/test_healing.py | 8 +- 18 files changed, 1336 insertions(+), 51 deletions(-) create mode 100644 .pre-commit-hooks.yaml create mode 100644 src/ninelives/frameworks/__init__.py create mode 100644 src/ninelives/frameworks/cypress.py create mode 100644 src/ninelives/frameworks/playwright.py create mode 100644 src/ninelives/frameworks/selenium.py create mode 100644 src/ninelives/history.py create mode 100644 src/ninelives/mcp_server.py create mode 100644 src/ninelives/watch.py create mode 100644 tests/test_expansion.py diff --git a/.pre-commit-hooks.yaml b/.pre-commit-hooks.yaml new file mode 100644 index 0000000..fb75b24 --- /dev/null +++ b/.pre-commit-hooks.yaml @@ -0,0 +1,25 @@ +# 9lives as a pre-commit hook — heal failing specs before they reach CI. +# +# repo: https://github.com/Quality-Max/9lives +# rev: v0.2.0 +# hooks: +# - id: 9lives-heal +# +# Both hooks need the runtime the spec's framework needs (Node for +# Playwright/Cypress, pytest for Selenium) available on PATH. +- id: 9lives-heal + name: 9lives — heal failing specs + description: Run changed specs through `9l heal --yes`; drifted selectors are fixed in place, assertion failures still block the commit. + entry: 9l heal --yes + language: python + files: '(\.(spec|test)\.[cm]?[jt]sx?|\.cy\.[cm]?[jt]sx?)$' + pass_filenames: true + verbose: true + +- id: 9lives-run + name: 9lives — run specs (no healing) + description: Run changed specs through `9l run` and fail the commit on red, without modifying anything. + entry: 9l run + language: python + files: '(\.(spec|test)\.[cm]?[jt]sx?|\.cy\.[cm]?[jt]sx?)$' + pass_filenames: true diff --git a/README.md b/README.md index 5c50384..00b430b 100644 --- a/README.md +++ b/README.md @@ -35,11 +35,71 @@ Requires Node.js ≥ 18 (Playwright itself runs on Node). Check your setup with 9l run # run a spec locally; screenshots/videos/traces in .9lives/ 9l heal # run → heal → re-run → diff → apply on confirm 9l heal --yes # CI mode: apply automatically, exit code tells the story +9l watch [dir] # heal on save — polls specs, heals whatever changes +9l report # brittle-selector report from your local heal history +9l mcp # serve heal_test/run_test as MCP tools for coding agents 9l doctor # environment check ``` Works inside an existing Playwright project (uses your `package.json`) or on a bare `.spec.ts` file (scaffolds an ephemeral project automatically). All commands accept multiple specs/globs. +## Cypress & Selenium too + +The heal verb is framework-agnostic. `9l heal` auto-detects the framework per spec — `.cy.js`/`.cy.ts` (or a package.json depending on cypress) runs through **Cypress**, `.py` specs run through **Selenium via your own pytest**, everything else is Playwright. Force it with `--framework cypress|selenium|playwright`. + +```bash +9l heal cypress/e2e/login.cy.js # runs `npx cypress run` in your project +9l heal tests/test_checkout.py # runs your pytest + selenium +``` + +Same loop everywhere: classify → Tier 1 offline selector repair → Tier 2 via your subscription → re-run → diff → approve. Cypress's `Expected to find element: \`#x\`` and Selenium's `NoSuchElementException` payloads both carry the failing selector, and both are correctly treated as *selector drift* — never as assertion failures (Cypress wraps locator misses in `AssertionError`; 9lives sees through that so the behavior-vs-drift guard doesn't misfire). 9lives never scaffolds a Cypress project or a Python env — those specs run against your own install. + +## For coding agents: `9l mcp` + +Your agent wrote code, a test went red — let it heal the test **in-loop** instead of waiting for CI. `9l mcp` speaks MCP over stdio (zero extra dependencies) and exposes two tools: `heal_test` (run → heal → verify → return the diff; `apply: true` writes it in place, otherwise a `.healed` copy is saved for review) and `run_test`. + +```bash +# Claude Code +claude mcp add 9lives -- 9l mcp +# or without installing first: +claude mcp add 9lives -- uvx --from 9lives 9l mcp +``` + +```json +// Cursor (.cursor/mcp.json) / Codex — any MCP host with stdio servers +{ "mcpServers": { "9lives": { "command": "9l", "args": ["mcp"] } } } +``` + +The behavior-vs-drift guard applies to agents too: a failing assertion comes back as `needs-human`, with an explicit note that forcing it green would mask a real bug. + +## Heal on save & pre-commit + +`9l watch` makes healing part of the edit-save loop: it polls your specs (no OS-specific watchers, works everywhere) and runs the heal loop on whatever changed. `--yes` applies automatically. + +```bash +9l watch tests/ --yes +``` + +As a [pre-commit](https://pre-commit.com) hook — heal (or just run) changed specs before they ever reach CI: + +```yaml +repos: + - repo: https://github.com/Quality-Max/9lives + rev: v0.2.0 + hooks: + - id: 9lives-heal # heals drifted selectors in place; assertion failures still block + # - id: 9lives-run # strict variant: run only, never modify +``` + +## Which selectors are rotting? `9l report` + +Every heal appends a line to `.9lives/history.jsonl` next to the spec — locally, never uploaded. `9l report` aggregates that history into a brittle-selector report: which selectors keep breaking, which anchor (testid/id/text/class) keeps re-finding them, and what to pin instead. + +```bash +9l report # terminal table, worst selectors first +9l report --md brittle-selectors.md +``` + ## In CI: the GitHub Action ```yaml diff --git a/pyproject.toml b/pyproject.toml index 6355ca0..9d02b7e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,13 +4,13 @@ build-backend = "hatchling.build" [project] name = "9lives" -version = "0.1.3" +version = "0.2.0" description = "Self-healing QA for the coding-agent era. Your tests have nine lives." readme = "README.md" license = { text = "MIT" } requires-python = ">=3.10" authors = [{ name = "QualityMax" }] -keywords = ["playwright", "testing", "self-healing", "qa", "ai", "agents"] +keywords = ["playwright", "cypress", "selenium", "testing", "self-healing", "qa", "ai", "agents", "mcp"] classifiers = [ "Development Status :: 3 - Alpha", "Intended Audience :: Developers", diff --git a/src/ninelives/__init__.py b/src/ninelives/__init__.py index 5b04856..587b6ca 100644 --- a/src/ninelives/__init__.py +++ b/src/ninelives/__init__.py @@ -1,3 +1,3 @@ """9lives — self-healing QA for the coding-agent era.""" -__version__ = "0.1.0" +__version__ = "0.2.0" diff --git a/src/ninelives/cli.py b/src/ninelives/cli.py index ddc55e4..757b646 100644 --- a/src/ninelives/cli.py +++ b/src/ninelives/cli.py @@ -1,10 +1,16 @@ """9lives CLI — `9l` / `9lives`. Commands: - 9l run Run a Playwright spec locally, artifacts in .9lives/ + 9l run Run a spec locally, artifacts in .9lives/ 9l heal Run → classify → Tier 1 (offline) → Tier 2 (your key) → re-run → show diff → apply on confirm + 9l watch [paths] Heal on save — poll specs and heal whatever changes + 9l report [path] Brittle-selector report from local heal history + 9l mcp Serve heal_test/run_test as MCP tools for coding agents 9l doctor Check node/npx/playwright/API keys + +Frameworks: Playwright (native), Cypress and Selenium/pytest via adapters — +auto-detected per spec, or forced with --framework. """ import argparse @@ -18,6 +24,7 @@ from pathlib import Path from . import __version__ +from .frameworks import FRAMEWORKS, get_adapter from .healing.parse import extract_failed_selector from .healing.patch import diff_stats, generate_unified_diff from .healing.strategy import FailureType, HealingTier, TestFailure, healing_strategy_selector @@ -26,8 +33,7 @@ from .llm.agent_cli import detect_agent_clis from .llm.client import LLMClient from .report.github import SpecOutcome, write_github_reports -from .runner.artifacts import collect_artifacts -from .runner.execute import RunnerError, ensure_project_ready, run_spec +from .runner.execute import RunnerError from .runner.project import find_user_project logger = logging.getLogger(__name__) @@ -36,20 +42,42 @@ PAW = "\U0001f43e" +def _add_framework_flag(parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--framework", + choices=("auto", *FRAMEWORKS), + default="auto", + help="test framework (default: auto-detect from the spec name and project)", + ) + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="9l", description="9lives — your tests have nine lives. https://9lives.run") parser.add_argument("--version", action="version", version=f"9lives {__version__}") parser.add_argument("-v", "--verbose", action="store_true", help="debug logging") sub = parser.add_subparsers(dest="command", required=True) - run_parser = sub.add_parser("run", help="run Playwright specs locally") + run_parser = sub.add_parser("run", help="run specs locally") run_parser.add_argument("specs", type=Path, nargs="+") + _add_framework_flag(run_parser) heal_parser = sub.add_parser("heal", help="run specs and heal them if they fail") heal_parser.add_argument("specs", type=Path, nargs="+") heal_parser.add_argument("-y", "--yes", action="store_true", help="apply healed code without confirmation") heal_parser.add_argument("--max-iterations", type=int, default=MAX_HEALING_ITERATIONS) + _add_framework_flag(heal_parser) + watch_parser = sub.add_parser("watch", help="watch specs and heal on save") + watch_parser.add_argument("paths", type=Path, nargs="*", help="spec files or directories (default: cwd)") + watch_parser.add_argument("-y", "--yes", action="store_true", help="apply healed code without confirmation") + watch_parser.add_argument("--interval", type=float, default=1.0, help="poll interval in seconds") + _add_framework_flag(watch_parser) + + report_parser = sub.add_parser("report", help="brittle-selector report from heal history") + report_parser.add_argument("path", type=Path, nargs="?", default=Path("."), help="project root to scan") + report_parser.add_argument("--md", type=Path, default=None, help="also write the report as markdown to this file") + + sub.add_parser("mcp", help="run as an MCP server (stdio) for coding agents") sub.add_parser("doctor", help="check environment prerequisites") args = parser.parse_args(argv) @@ -60,9 +88,17 @@ def main(argv: list[str] | None = None) -> int: try: if args.command == "run": - return cmd_run(args.specs) + return cmd_run(args.specs, framework=args.framework) if args.command == "heal": - return cmd_heal(args.specs, auto_apply=args.yes, max_iterations=args.max_iterations) + return cmd_heal(args.specs, auto_apply=args.yes, max_iterations=args.max_iterations, framework=args.framework) + if args.command == "watch": + return cmd_watch(args.paths, auto_apply=args.yes, framework=args.framework, interval=args.interval) + if args.command == "report": + return cmd_report(args.path, md_path=args.md) + if args.command == "mcp": + from .mcp_server import serve + + return serve() if args.command == "doctor": return cmd_doctor() except RunnerError as e: @@ -75,11 +111,11 @@ def _dest_root(spec: Path) -> Path: return spec.resolve().parent / ".9lives" -def cmd_run(specs: list[Path]) -> int: +def cmd_run(specs: list[Path], framework: str = "auto") -> int: outcomes = [] exit_code = 0 for spec in specs: - outcome = run_one(spec) + outcome = run_one(spec, framework=framework) outcomes.append(outcome) if outcome.status != "passed": exit_code = 1 @@ -87,11 +123,12 @@ def cmd_run(specs: list[Path]) -> int: return exit_code -def run_one(spec: Path) -> SpecOutcome: - ensure_project_ready(spec) - print(f"{PAW} running {spec} …") - result = run_spec(spec) - artifacts = collect_artifacts(result.project_dir, _dest_root(spec)) if result.project_dir else None +def run_one(spec: Path, framework: str = "auto") -> SpecOutcome: + adapter = get_adapter(spec, framework) + adapter.preflight(spec) + print(f"{PAW} running {spec} ({adapter.name}) …") + result = adapter.run(spec) + artifacts = adapter.collect(result, _dest_root(spec)) if result.passed: print(f"{PAW} PASSED in {result.duration_ms / 1000:.1f}s") @@ -107,11 +144,16 @@ def run_one(spec: Path) -> SpecOutcome: return SpecOutcome(spec=spec.name, status="failed", detail=first_line) -def cmd_heal(specs: list[Path], auto_apply: bool = False, max_iterations: int = MAX_HEALING_ITERATIONS) -> int: +def cmd_heal( + specs: list[Path], + auto_apply: bool = False, + max_iterations: int = MAX_HEALING_ITERATIONS, + framework: str = "auto", +) -> int: outcomes = [] exit_code = 0 for spec in specs: - outcome = heal_one(spec, auto_apply=auto_apply, max_iterations=max_iterations) + outcome = heal_one(spec, auto_apply=auto_apply, max_iterations=max_iterations, framework=framework) outcomes.append(outcome) if outcome.status in ("failed", "needs-human"): exit_code = 1 @@ -119,29 +161,72 @@ def cmd_heal(specs: list[Path], auto_apply: bool = False, max_iterations: int = return exit_code -def heal_one(spec: Path, auto_apply: bool = False, max_iterations: int = MAX_HEALING_ITERATIONS) -> SpecOutcome: +def heal_one( + spec: Path, + auto_apply: bool = False, + max_iterations: int = MAX_HEALING_ITERATIONS, + framework: str = "auto", + interactive: bool = True, +) -> SpecOutcome: spec = spec.resolve() - ensure_project_ready(spec) # fail fast on a project missing @playwright/test, before scaffolding/healing + adapter = get_adapter(spec, framework) + adapter.preflight(spec) # fail fast on a broken project, before scaffolding/healing # Heal against a working copy so the user's own file is only touched after - # approval. When the spec lives inside a real Playwright project, that copy - # MUST stay a sibling of the original — otherwise its playwright.config - # (baseURL, projects, globalSetup), relative fixture imports, and testMatch - # don't resolve and every heal re-run would misfire in a bare scaffold. - # Only truly project-less specs fall back to an ephemeral temp dir. - if find_user_project(spec.parent): - working_spec = spec.with_name(f"_9lives_heal_{os.getpid()}_{spec.name}") - else: + # approval. When the spec lives inside a real project, that copy MUST stay + # a sibling of the original — otherwise its config (baseURL, projects, + # globalSetup), relative fixture imports, and testMatch don't resolve and + # every heal re-run would misfire in a bare scaffold. Only truly + # project-less Playwright specs fall back to an ephemeral temp dir. + if adapter.name == "playwright" and not find_user_project(spec.parent): working_spec = Path(tempfile.mkdtemp(prefix="ninelives-heal-")) / spec.name + else: + working_spec = spec.with_name(f"_9lives_heal_{os.getpid()}_{spec.name}") try: - return _heal_loop(spec, working_spec, auto_apply=auto_apply, max_iterations=max_iterations) + return _heal_loop( + spec, working_spec, adapter, auto_apply=auto_apply, max_iterations=max_iterations, interactive=interactive + ) finally: # The sibling copy lives in the user's tree; never leave it behind. if working_spec.parent == spec.parent and working_spec.exists(): working_spec.unlink() -def _heal_loop(spec: Path, working_spec: Path, *, auto_apply: bool, max_iterations: int) -> SpecOutcome: +def _record_history( + spec: Path, + adapter, + status: str, + failure: TestFailure | None = None, + healing=None, + applied: bool | None = None, + iterations: int | None = None, + detail: str = "", +) -> None: + """Append to .9lives/history.jsonl — best-effort, never breaks healing.""" + try: + from .history import record_heal + + record_heal( + spec, + framework=adapter.name, + status=status, + failure_type=failure.failure_type.value if failure else None, + failed_selector=failure.failed_selector if failure else None, + tier=healing.tier.value if healing else None, + confidence=healing.confidence if healing else None, + anchor=healing.metadata.get("anchor") if healing else None, + healed_selector=healing.metadata.get("new_selector") if healing else None, + applied=applied, + iterations=iterations, + detail=detail[:200], + ) + except Exception: + logger.debug("heal-history recording failed", exc_info=True) + + +def _heal_loop( + spec: Path, working_spec: Path, adapter, *, auto_apply: bool, max_iterations: int, interactive: bool = True +) -> SpecOutcome: original_code = spec.read_text() current_code = original_code working_spec.write_text(current_code) @@ -150,10 +235,13 @@ def _heal_loop(spec: Path, working_spec: Path, *, auto_apply: bool, max_iteratio healed = False tier1_used = False # Tier 1 gets one shot per session; a still-failing test escalates last_changes: list[str] = [] + failure: TestFailure | None = None + healing = None + iteration = 0 for iteration in range(1, max_iterations + 1): print(f"{PAW} run {iteration}/{max_iterations}: {spec.name} …") - result = run_spec(working_spec) + result = adapter.run(working_spec) if result.passed: if iteration == 1: @@ -163,7 +251,7 @@ def _heal_loop(spec: Path, working_spec: Path, *, auto_apply: bool, max_iteratio break error = result.errors[0] - artifacts = collect_artifacts(result.project_dir, _dest_root(spec)) if result.project_dir else None + artifacts = adapter.collect(result, _dest_root(spec)) # error-context.md carries the call log (the failing selector) and the # page snapshot (where Tier 1 finds the element's new identity). call_log = artifacts.call_log if artifacts else "" @@ -176,6 +264,7 @@ def _heal_loop(spec: Path, working_spec: Path, *, auto_apply: bool, max_iteratio stack_trace=error.stack, test_code=current_code, page_html=page_snapshot, + framework=adapter.name, ) tier = healing_strategy_selector.select_strategy(failure) print(f" failure: {failure.failure_type.value} (selector: {failure.failed_selector or 'n/a'}) → {tier.value}") @@ -202,15 +291,14 @@ def _heal_loop(spec: Path, working_spec: Path, *, auto_apply: bool, max_iteratio print(" NINELIVES_HEAL_ASSERTIONS=1 to let Tier 2 propose an assertion update.") else: print(f"{PAW} this failure needs a human ({tier.value}) — no automatic fix attempted.") - return SpecOutcome( - spec=spec.name, - status="needs-human", - detail=f"{failure.failure_type.value}: {error.message.splitlines()[0] if error.message else ''}", - ) + detail = f"{failure.failure_type.value}: {error.message.splitlines()[0] if error.message else ''}" + _record_history(spec, adapter, "needs-human", failure=failure, iterations=iteration, detail=detail) + return SpecOutcome(spec=spec.name, status="needs-human", detail=detail) if not healing.success or not healing.healed_code: reason = healing.metadata.get("reason", "no fix produced") print(f"{PAW} could not heal: {reason}") + _record_history(spec, adapter, "failed", failure=failure, healing=healing, iterations=iteration, detail=reason) return SpecOutcome(spec=spec.name, status="failed", detail=f"unhealed — {reason}") last_changes = healing.changes_made @@ -226,12 +314,14 @@ def _heal_loop(spec: Path, working_spec: Path, *, auto_apply: bool, max_iteratio # success instead of a false "still failing after N attempts". if not healed and current_code != original_code: print(f"{PAW} verifying final heal …") - if run_spec(working_spec).passed: + if adapter.run(working_spec).passed: healed = True if not healed: print(f"{PAW} still failing after {max_iterations} heal attempts. Artifacts in {_dest_root(spec)}") - return SpecOutcome(spec=spec.name, status="failed", detail=f"still failing after {max_iterations} heal attempts") + detail = f"still failing after {max_iterations} heal attempts" + _record_history(spec, adapter, "failed", failure=failure, healing=healing, iterations=iteration, detail=detail) + return SpecOutcome(spec=spec.name, status="failed", detail=detail) diff = generate_unified_diff(original_code, current_code, script_name=spec.name) stats = diff_stats(diff) @@ -240,18 +330,54 @@ def _heal_loop(spec: Path, working_spec: Path, *, auto_apply: bool, max_iteratio detail = "; ".join(last_changes)[:200] if last_changes else "selector healed" if not auto_apply: - answer = input(f"apply to {spec}? [y/N] ").strip().lower() - if answer not in ("y", "yes"): + apply_it = False + if interactive: + answer = input(f"apply to {spec}? [y/N] ").strip().lower() + apply_it = answer in ("y", "yes") + if not apply_it: healed_copy = spec.with_suffix(spec.suffix + ".healed") healed_copy.write_text(current_code) print(f"{PAW} not applied — healed copy saved to {healed_copy}") + _record_history( + spec, adapter, "healed", failure=failure, healing=healing, applied=False, iterations=iteration, detail=detail + ) return SpecOutcome(spec=spec.name, status="healed", detail=f"not applied — saved to {healed_copy.name}", diff=diff) spec.write_text(current_code) print(f"{PAW} applied. Your test came back to life.") + _record_history( + spec, adapter, "healed", failure=failure, healing=healing, applied=True, iterations=iteration, detail=detail + ) return SpecOutcome(spec=spec.name, status="healed", detail=detail, diff=diff) +def cmd_watch(paths: list[Path], auto_apply: bool = False, framework: str = "auto", interval: float = 1.0) -> int: + from .watch import watch_loop + + targets = list(paths) or [Path(".")] + + def on_change(path: Path) -> None: + print(f"\n{PAW} {path} changed") + try: + heal_one(path, auto_apply=auto_apply, framework=framework) + except RunnerError as e: + print(f"{PAW} error: {e}", file=sys.stderr) + + return watch_loop(targets, on_change, interval=interval) + + +def cmd_report(path: Path, md_path: Path | None = None) -> int: + from .history import load_history, render_report, selector_report + + records = load_history(path) + rows = selector_report(records) + print(render_report(rows)) + if md_path: + md_path.write_text(render_report(rows, markdown=True) + "\n") + print(f"\n{PAW} markdown report → {md_path}") + return 0 + + def cmd_doctor() -> int: checks: list[tuple[str, bool, str]] = [] @@ -264,6 +390,13 @@ def cmd_doctor() -> int: node_version = "found but not responding" checks.append(("node", bool(node), node_version or "not found — install Node.js >= 18")) checks.append(("npx", bool(shutil.which("npx")), "" if shutil.which("npx") else "not found")) + checks.append( + ( + "pytest", + bool(shutil.which("pytest")), + "" if shutil.which("pytest") else "not found — only needed for Selenium (Python) specs", + ) + ) clis = detect_agent_clis() checks.append( diff --git a/src/ninelives/frameworks/__init__.py b/src/ninelives/frameworks/__init__.py new file mode 100644 index 0000000..8fb9abb --- /dev/null +++ b/src/ninelives/frameworks/__init__.py @@ -0,0 +1,61 @@ +"""Framework adapters — make the heal verb framework-agnostic. + +`9l heal` speaks Playwright natively; these adapters extend the same +run → classify → heal loop to Cypress and Selenium (pytest) specs. Each +adapter knows how to run one spec and return a `RunResult`; everything +downstream (classification, Tier 1 anchors, Tier 2 prompts, diffs, +reports, history) is shared. +""" + +import json +import re +from pathlib import Path + +from ..runner.project import find_enclosing_package_json +from .cypress import CypressAdapter +from .playwright import PlaywrightAdapter +from .selenium import SeleniumAdapter + +FRAMEWORKS = ("playwright", "cypress", "selenium") + +_ADAPTERS = { + "playwright": PlaywrightAdapter(), + "cypress": CypressAdapter(), + "selenium": SeleniumAdapter(), +} + +_CYPRESS_SUFFIX = re.compile(r"\.cy\.[cm]?[jt]sx?$", re.IGNORECASE) + + +def detect_framework(spec: Path) -> str: + """Infer the test framework for a spec file. + + Order: the `.cy.*` naming convention, then Python (Selenium runs via + pytest), then the enclosing package.json's dependencies — a project + that depends on cypress but not @playwright/test is a Cypress project + even for plain `.spec.js` names. Playwright is the default. + """ + name = spec.name.lower() + if _CYPRESS_SUFFIX.search(name): + return "cypress" + if name.endswith(".py"): + return "selenium" + + pkg_dir = find_enclosing_package_json(spec.resolve().parent) + if pkg_dir: + try: + data = json.loads((pkg_dir / "package.json").read_text()) + except (OSError, json.JSONDecodeError): + return "playwright" + deps = {**data.get("dependencies", {}), **data.get("devDependencies", {})} + if "cypress" in deps and "@playwright/test" not in deps: + return "cypress" + return "playwright" + + +def get_adapter(spec: Path, framework: str = "auto"): + """Resolve the adapter for a spec, honoring an explicit --framework.""" + name = framework if framework and framework != "auto" else detect_framework(spec) + if name not in _ADAPTERS: + raise ValueError(f"Unknown framework: {name} (expected one of {', '.join(FRAMEWORKS)})") + return _ADAPTERS[name] diff --git a/src/ninelives/frameworks/cypress.py b/src/ninelives/frameworks/cypress.py new file mode 100644 index 0000000..6b41a7b --- /dev/null +++ b/src/ninelives/frameworks/cypress.py @@ -0,0 +1,107 @@ +"""Cypress adapter — run one Cypress spec and parse the Mocha JSON report. + +Cypress specs run inside the user's own Cypress project (no ephemeral +scaffolding — a Cypress install is ~700 MB of binary, so 9lives never +creates one behind the user's back). `--quiet` keeps stdout to the +reporter's JSON; a brace-scan fallback survives any preamble noise. +""" + +import json +import logging +from pathlib import Path + +from ..runner.execute import RUN_TIMEOUT_SECONDS, RunnerError, RunResult, TestError, _require, _run + +logger = logging.getLogger(__name__) + + +class CypressAdapter: + name = "cypress" + language = "javascript" + + def preflight(self, spec: Path) -> None: + if self.project_dir(spec) is None: + raise RunnerError( + "No Cypress project found around this spec. 9lives runs Cypress specs " + "against your own project (it never scaffolds a Cypress install):\n" + " npm install -D cypress\n" + "then re-run from inside the project." + ) + + def project_dir(self, spec: Path) -> Path | None: + """Nearest ancestor whose package.json depends on cypress.""" + start = spec.resolve().parent + for directory in [start, *start.parents]: + package_json = directory / "package.json" + if package_json.is_file(): + try: + data = json.loads(package_json.read_text()) + except (OSError, json.JSONDecodeError): + continue + deps = {**data.get("dependencies", {}), **data.get("devDependencies", {})} + if "cypress" in deps: + return directory + return None + + def run(self, spec: Path) -> RunResult: + spec = spec.resolve() + if not spec.is_file(): + raise RunnerError(f"Spec not found: {spec}") + npx = _require("npx") + project = self.project_dir(spec) + if project is None: + raise RunnerError("No Cypress project found (package.json with a cypress dependency)") + + cmd = [npx, "cypress", "run", "--spec", str(spec), "--reporter", "json", "--quiet"] + logger.info("Running: %s (cwd=%s)", " ".join(cmd), project) + proc = _run(cmd, project, RUN_TIMEOUT_SECONDS) + + errors, duration_ms = parse_mocha_json(proc.stdout) + passed = proc.returncode == 0 + if not passed and not errors: + errors = [TestError(title=spec.name, message=(proc.stderr or proc.stdout)[-3000:])] + + return RunResult( + passed=passed, + exit_code=proc.returncode, + errors=errors, + stdout=proc.stdout, + stderr=proc.stderr, + project_dir=project, + duration_ms=duration_ms, + ) + + def collect(self, result: RunResult, dest_root: Path): + # Cypress writes no failure-time page snapshot, so there is nothing to + # feed Tier 1's anchor search — heals rely on selector transformations + # and Tier 2. Screenshots stay where Cypress puts them. + return None + + +def parse_mocha_json(stdout: str) -> tuple[list[TestError], int]: + """Extract failures from Mocha's JSON reporter output on stdout. + + Cypress may still print banners around the JSON, so scan for the + outermost braces instead of trusting the whole stream. + """ + start = stdout.find("{") + end = stdout.rfind("}") + if start == -1 or end <= start: + return [], 0 + try: + report = json.loads(stdout[start : end + 1]) + except json.JSONDecodeError: + return [], 0 + + errors: list[TestError] = [] + for failure in report.get("failures", []): + err = failure.get("err", {}) or {} + errors.append( + TestError( + title=failure.get("fullTitle") or failure.get("title", "unknown"), + message=err.get("message", ""), + stack=err.get("stack", ""), + ) + ) + duration_ms = int(report.get("stats", {}).get("duration", 0) or 0) + return errors, duration_ms diff --git a/src/ninelives/frameworks/playwright.py b/src/ninelives/frameworks/playwright.py new file mode 100644 index 0000000..da23299 --- /dev/null +++ b/src/ninelives/frameworks/playwright.py @@ -0,0 +1,23 @@ +"""Playwright adapter — thin wrapper over the original execution kernel.""" + +from pathlib import Path + +from ..runner import execute +from ..runner.artifacts import Artifacts, collect_artifacts +from ..runner.execute import RunResult, ensure_project_ready + + +class PlaywrightAdapter: + name = "playwright" + language = "javascript" + + def preflight(self, spec: Path) -> None: + ensure_project_ready(spec) + + def run(self, spec: Path) -> RunResult: + return execute.run_spec(spec) + + def collect(self, result: RunResult, dest_root: Path) -> Artifacts | None: + if result.project_dir is None: + return None + return collect_artifacts(result.project_dir, dest_root) diff --git a/src/ninelives/frameworks/selenium.py b/src/ninelives/frameworks/selenium.py new file mode 100644 index 0000000..b53d967 --- /dev/null +++ b/src/ninelives/frameworks/selenium.py @@ -0,0 +1,106 @@ +"""Selenium adapter — run one Python Selenium spec through pytest. + +Selenium has no single project convention, but in the Python world it +almost always runs under pytest — so that's the contract: a `.py` spec, +executed with the user's own pytest (their venv carries selenium and the +driver setup). Failures are parsed from pytest's text output; the +`NoSuchElementException` payload carries the failing selector. +""" + +import logging +import re +import shutil +from pathlib import Path + +from ..runner.execute import RUN_TIMEOUT_SECONDS, RunnerError, RunResult, TestError, _run + +logger = logging.getLogger(__name__) + +_SECTION_HEADER = re.compile(r"^_{3,}\s+(.+?)\s+_{3,}$") +_SUMMARY_LINE = re.compile(r"^(?:FAILED|ERROR)\s") + + +class SeleniumAdapter: + name = "selenium" + language = "python" + + def preflight(self, spec: Path) -> None: + if shutil.which("pytest") is None: + raise RunnerError( + "'pytest' not found on PATH — Selenium specs run through your own pytest:\n" + " pip install pytest selenium\n" + "then re-run 9lives from the environment your tests live in." + ) + + def run(self, spec: Path) -> RunResult: + spec = spec.resolve() + if not spec.is_file(): + raise RunnerError(f"Spec not found: {spec}") + pytest_bin = shutil.which("pytest") + if pytest_bin is None: + raise RunnerError("'pytest' not found on PATH (see `9l doctor`)") + + cmd = [pytest_bin, str(spec), "--tb=long", "-q", "--color=no", "-p", "no:cacheprovider"] + logger.info("Running: %s", " ".join(cmd)) + proc = _run(cmd, spec.parent, RUN_TIMEOUT_SECONDS) + + errors = parse_pytest_output(proc.stdout) + passed = proc.returncode == 0 + if not passed and not errors: + errors = [TestError(title=spec.name, message=(proc.stderr or proc.stdout)[-3000:])] + + return RunResult( + passed=passed, + exit_code=proc.returncode, + errors=errors, + stdout=proc.stdout, + stderr=proc.stderr, + project_dir=spec.parent, + ) + + def collect(self, result: RunResult, dest_root: Path): + # No failure-time page snapshot — Selenium heals lean on the selector + # embedded in the exception plus Tier 2. + return None + + +def parse_pytest_output(stdout: str) -> list[TestError]: + """Extract one TestError per failed test from pytest's -q text output. + + pytest separates failures with `____ test_name ____` headers; the + exception itself arrives on `E ...` lines. Everything else in the + section is kept as the stack (it contains the failing source line, + which selector extraction also scans). + """ + errors: list[TestError] = [] + title: str | None = None + message_lines: list[str] = [] + stack_lines: list[str] = [] + + def flush() -> None: + if title and message_lines: + errors.append( + TestError( + title=title, + message="\n".join(message_lines), + stack="\n".join(stack_lines)[-4000:], + ) + ) + + for line in stdout.splitlines(): + header = _SECTION_HEADER.match(line.strip()) + if header: + flush() + title = header.group(1) + message_lines = [] + stack_lines = [] + elif _SUMMARY_LINE.match(line): + flush() + title = None + elif title is not None: + if line.startswith("E "): + message_lines.append(line[1:].strip()) + else: + stack_lines.append(line) + flush() + return errors diff --git a/src/ninelives/healing/parse.py b/src/ninelives/healing/parse.py index f0b11de..a070410 100644 --- a/src/ninelives/healing/parse.py +++ b/src/ninelives/healing/parse.py @@ -21,23 +21,50 @@ ) _WAITING_FOR = re.compile(r"waiting for (?:locator|selector)\s*\(?" + _QUOTED + r"\)?", re.IGNORECASE) +# Cypress: "Timed out retrying after 4000ms: Expected to find element: `#login`, +# but never found it." — the selector travels between backticks. +_CYPRESS_EXPECTED = re.compile(r"Expected to find (?:element|content):\s*`([^`]+)`", re.IGNORECASE) +_CY_CALL = re.compile(r"cy\.(?:get|find|contains)\(" + _QUOTED, re.IGNORECASE) + +# Selenium: NoSuchElementException carries a JSON payload — +# {"method":"css selector","selector":"#login"} +# and the pytest stack shows the source call: By.CSS_SELECTOR, "#login". +_SELENIUM_JSON = re.compile(r'"method"\s*:\s*"[^"]+"\s*,\s*"selector"\s*:\s*"((?:\\.|[^"\\])*)"') +_SELENIUM_BY = re.compile(r"By\.\w+\s*,\s*" + _QUOTED) + def _unescape(selector: str) -> str: return selector.replace("\\'", "'").replace('\\"', '"').replace("\\\\", "\\") def extract_failed_selector(error_message: str, stack_trace: str = "") -> str | None: - """Extract the failing selector from a Playwright error message.""" + """Extract the failing selector from a Playwright/Cypress/Selenium error.""" combined = f"{error_message}\n{stack_trace}" match = _WAITING_FOR.search(combined) if match: return _unescape(match.group(2)) + match = _CYPRESS_EXPECTED.search(combined) + if match: + return match.group(1) + + match = _SELENIUM_JSON.search(combined) + if match: + return _unescape(match.group(1)) + match = _SELECTOR_CALL.search(combined) if match: return _unescape(match.group(2)) + match = _CY_CALL.search(combined) + if match: + return _unescape(match.group(2)) + + match = _SELENIUM_BY.search(combined) + if match: + return _unescape(match.group(2)) + return None diff --git a/src/ninelives/healing/strategy.py b/src/ninelives/healing/strategy.py index 51b23f3..6d1ad09 100644 --- a/src/ninelives/healing/strategy.py +++ b/src/ninelives/healing/strategy.py @@ -53,6 +53,7 @@ class TestFailure: test_code: str = "" page_url: str = "" page_html: str = "" + framework: str = "playwright" # playwright | cypress | selenium def to_dict(self) -> dict[str, Any]: return { @@ -99,6 +100,19 @@ class HealingStrategySelector: Tier 3 (Human): flow changed, redesign needed — analysis only. """ + # Checked before everything else: Cypress wraps locator misses in + # AssertionError ("Timed out retrying …: Expected to find element: `#x`") + # and Selenium raises NoSuchElementException — both are selector drift, + # not behavior change, and must never trip the assertion guard. + LOCATOR_OVERRIDE_PATTERNS = [ + r"expected to find (?:element|content)", + r"never found it", + r"unable to locate element", + r"no such element", + r"NoSuchElementException", + r"StaleElementReferenceException", + ] + LOCATOR_PATTERNS = [ r"locator.*not found", r"element.*not found", @@ -154,6 +168,10 @@ def classify_failure(self, error_message: str, stack_trace: str = "") -> Failure """Classify the type of failure from error message.""" combined = f"{error_message} {stack_trace}".lower() + for pattern in self.LOCATOR_OVERRIDE_PATTERNS: + if re.search(pattern, combined, re.IGNORECASE): + return FailureType.LOCATOR_NOT_FOUND + for pattern in self.FLOW_CHANGE_PATTERNS: if re.search(pattern, combined, re.IGNORECASE): return FailureType.FLOW_CHANGED diff --git a/src/ninelives/healing/tier1.py b/src/ninelives/healing/tier1.py index a397b57..8527ba2 100644 --- a/src/ninelives/healing/tier1.py +++ b/src/ninelives/healing/tier1.py @@ -43,7 +43,7 @@ async def heal(self, failure: TestFailure) -> HealingResult: healed_code=healed_code, changes_made=[f"Re-found via {anchor}: replaced '{failure.failed_selector}' with '{alternative}'"], confidence=0.85, - metadata={"anchor": anchor}, + metadata={"anchor": anchor, "old_selector": failure.failed_selector, "new_selector": alternative}, ) transformed = self._try_transformations(failure.failed_selector) @@ -55,9 +55,12 @@ async def heal(self, failure: TestFailure) -> HealingResult: healed_code=healed_code, changes_made=[f"Transformed selector '{failure.failed_selector}' to '{transformed}'"], confidence=0.7, + metadata={"old_selector": failure.failed_selector, "new_selector": transformed}, ) - if self._is_timing_issue(failure): + # The wait/scroll injection writes Playwright API calls; other + # frameworks escalate to Tier 2 instead of gaining `await page.…` lines. + if failure.framework == "playwright" and self._is_timing_issue(failure): healed_code = self._add_wait_handling(failure.test_code, failure.failed_selector) return HealingResult( tier=HealingTier.TIER1_AUTO, diff --git a/src/ninelives/healing/tier2.py b/src/ninelives/healing/tier2.py index cb5454f..83de7e7 100644 --- a/src/ninelives/healing/tier2.py +++ b/src/ninelives/healing/tier2.py @@ -47,10 +47,18 @@ async def suggest(self, failure: TestFailure) -> HealingResult: metadata={"ai_reasoning": suggestion, "original_error": failure.error_message}, ) + # Framework → (human label, code-fence language) for the prompt. + FRAMEWORK_LABELS = { + "playwright": ("Playwright", "javascript"), + "cypress": ("Cypress", "javascript"), + "selenium": ("Selenium (Python + pytest)", "python"), + } + def _build_prompt(self, failure: TestFailure) -> str: """Build prompt for AI suggestion.""" + label, fence = self.FRAMEWORK_LABELS.get(failure.framework, ("Playwright", "javascript")) lines = [ - "A Playwright test is failing. Please suggest a fix.", + f"A {label} test is failing. Please suggest a fix.", "", "## Error Details", f"Error Type: {failure.failure_type.value}", @@ -61,7 +69,7 @@ def _build_prompt(self, failure: TestFailure) -> str: if failure.failed_selector: lines.extend([f"Failed Selector: {failure.failed_selector}", ""]) - lines.extend(["## Test Code", "```javascript", failure.test_code, "```", ""]) + lines.extend(["## Test Code", f"```{fence}", failure.test_code, "```", ""]) if failure.page_html: lines.extend(["## Page state at failure (snippet)", "```", failure.page_html[:3000], "```", ""]) @@ -81,7 +89,7 @@ def _build_prompt(self, failure: TestFailure) -> str: "REASONING: ", "CHANGES: ", "CODE:", - "```javascript", + f"```{fence}", "", "```", ] @@ -98,11 +106,13 @@ def _get_ai_suggestion(self, prompt: str) -> str | None: def _parse_suggestion(self, suggestion: str, original_code: str) -> str: """Parse the healed code from AI suggestion.""" - code_match = re.search(r"CODE:\s*```(?:javascript|typescript)?\s*\n(.*?)```", suggestion, re.DOTALL | re.IGNORECASE) + code_match = re.search( + r"CODE:\s*```(?:javascript|typescript|python|js|ts|py)?\s*\n(.*?)```", suggestion, re.DOTALL | re.IGNORECASE + ) if code_match: return code_match.group(1).strip() - code_block = re.search(r"```(?:javascript|typescript)?\s*\n(.*?)```", suggestion, re.DOTALL) + code_block = re.search(r"```(?:javascript|typescript|python|js|ts|py)?\s*\n(.*?)```", suggestion, re.DOTALL) if code_block: return code_block.group(1).strip() diff --git a/src/ninelives/history.py b/src/ninelives/history.py new file mode 100644 index 0000000..2cb15f2 --- /dev/null +++ b/src/ninelives/history.py @@ -0,0 +1,138 @@ +"""Heal history — the data only 9lives has. + +Every heal attempt appends one JSON line to `.9lives/history.jsonl` next to +the spec. `9l report` aggregates those lines into a brittle-selector report: +which selectors keep rotting, which anchor keeps re-finding them, and what +to pin instead. Local files only — nothing is uploaded anywhere. +""" + +import json +import logging +import time +from collections import Counter +from pathlib import Path + +logger = logging.getLogger(__name__) + +HISTORY_FILENAME = "history.jsonl" + + +def history_path(spec: Path) -> Path: + return spec.resolve().parent / ".9lives" / HISTORY_FILENAME + + +def record_heal(spec: Path, **fields) -> Path: + """Append one heal-attempt record for a spec. Never raises into the heal loop.""" + path = history_path(spec) + path.parent.mkdir(parents=True, exist_ok=True) + record = { + "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), + "spec": spec.name, + "spec_path": str(spec.resolve()), + **fields, + } + with path.open("a", encoding="utf-8") as f: + f.write(json.dumps(record, ensure_ascii=False) + "\n") + return path + + +def load_history(root: Path) -> list[dict]: + """Read every history file under `root` (skipping unparseable lines).""" + records: list[dict] = [] + for path in sorted(root.resolve().rglob(f".9lives/{HISTORY_FILENAME}")): + try: + lines = path.read_text(encoding="utf-8").splitlines() + except OSError: + continue + for line in lines: + try: + records.append(json.loads(line)) + except json.JSONDecodeError: + continue + return records + + +def selector_report(records: list[dict]) -> list[dict]: + """Aggregate heal records into per-(spec, selector) rows, worst first.""" + groups: dict[tuple[str, str], dict] = {} + for r in records: + if r.get("status") == "passed": + continue + key = (r.get("spec") or "?", r.get("failed_selector") or "(unknown)") + g = groups.setdefault( + key, + { + "spec": key[0], + "selector": key[1], + "heals": 0, + "unhealed": 0, + "needs_human": 0, + "anchors": Counter(), + "last_new_selector": "", + "first_ts": r.get("ts", ""), + "last_ts": r.get("ts", ""), + }, + ) + status = r.get("status") + if status == "healed": + g["heals"] += 1 + if r.get("healed_selector"): + g["last_new_selector"] = r["healed_selector"] + elif status == "needs-human": + g["needs_human"] += 1 + else: + g["unhealed"] += 1 + if r.get("anchor"): + g["anchors"][r["anchor"]] += 1 + g["last_ts"] = max(g["last_ts"], r.get("ts", "")) + g["first_ts"] = min(g["first_ts"], r.get("ts", "")) if r.get("ts") else g["first_ts"] + + rows = [] + for g in groups.values(): + g["top_anchor"] = g["anchors"].most_common(1)[0][0] if g["anchors"] else "" + g["events"] = g["heals"] + g["unhealed"] + g["needs_human"] + g["recommendation"] = _recommend(g) + rows.append(g) + rows.sort(key=lambda g: (g["events"], g["heals"]), reverse=True) + return rows + + +def _recommend(g: dict) -> str: + if g["needs_human"] and not g["heals"]: + return "failures look behavioral — review the app/test, not the selector" + if g["heals"] >= 2: + return f"rotting — healed {g['heals']}×; pin a data-testid" + if g["top_anchor"] == "text": + return "text-anchored — copy changes break it; add a data-testid" + if g["top_anchor"] == "class": + return "class-anchored — styling churn breaks it; add a data-testid or id" + if g["unhealed"] and not g["heals"]: + return "never auto-healed — likely needs a rewrite" + return "watch — one heal so far" + + +def render_report(rows: list[dict], markdown: bool = False) -> str: + """Render the brittle-selector report for the terminal or as markdown.""" + total_heals = sum(r["heals"] for r in rows) + title = f"🐾 brittle-selector report — {total_heals} heal(s) across {len(rows)} selector(s)" + + if not rows: + return f"{title}\n\nno heal history yet — every `9l heal` run adds to .9lives/history.jsonl" + + if markdown: + lines = [f"## {title}", "", "| selector | spec | heals | anchor | recommendation |", "|---|---|---|---|---|"] + for r in rows: + sel = r["selector"].replace("|", "\\|") + lines.append(f"| `{sel}` | `{r['spec']}` | {r['heals']} | {r['top_anchor'] or '—'} | {r['recommendation']} |") + lines.extend(["", "_generated locally by [9lives](https://9lives.run) `9l report` — data never leaves your machine_"]) + return "\n".join(lines) + + width = min(max((len(r["selector"]) for r in rows), default=8) + 2, 42) + lines = [title, ""] + lines.append(f" {'selector':<{width}} {'spec':<24} {'heals':>5} {'anchor':<10} recommendation") + for r in rows: + sel = r["selector"][: width - 1] + lines.append( + f" {sel:<{width}} {r['spec'][:23]:<24} {r['heals']:>5} {r['top_anchor'] or '—':<10} {r['recommendation']}" + ) + return "\n".join(lines) diff --git a/src/ninelives/mcp_server.py b/src/ninelives/mcp_server.py new file mode 100644 index 0000000..e3ea5ee --- /dev/null +++ b/src/ninelives/mcp_server.py @@ -0,0 +1,202 @@ +"""`9l mcp` — 9lives as an MCP server for coding agents. + +Claude Code / Cursor / Codex can call `heal_test` mid-session: the agent +ships a change, a spec goes red, and it heals in-loop instead of waiting +for CI. Zero dependencies: MCP's stdio transport is newline-delimited +JSON-RPC 2.0, small enough to speak directly. + + claude mcp add 9lives -- 9l mcp + +Protocol notes: stdout carries ONLY protocol frames; all heal-loop output +is redirected to stderr (which MCP hosts surface as server logs). +""" + +import contextlib +import json +import logging +import sys +from pathlib import Path + +from . import __version__ + +logger = logging.getLogger(__name__) + +PROTOCOL_VERSION = "2025-06-18" + +SERVER_INFO = {"name": "9lives", "title": "9lives — self-healing tests", "version": __version__} + +INSTRUCTIONS = ( + "Self-healing test runner. Call heal_test when a Playwright/Cypress/Selenium spec fails: " + "it re-runs the spec, repairs drifted selectors (offline Tier 1, LLM Tier 2), verifies green, " + "and returns the diff. With apply=false the fix is saved next to the spec as .healed " + "for you to review/apply; with apply=true it is written in place. " + "A needs-human status usually means a failing assertion — a possible real bug 9lives refuses to mask." +) + +TOOLS = [ + { + "name": "heal_test", + "title": "Heal a failing test", + "description": ( + "Run a test spec and self-heal it if it fails (selector drift, timing). Returns status " + "(passed | healed | failed | needs-human), the unified diff, and where the healed code went. " + "needs-human means the failure looks like a real behavior change — do not force the test green." + ), + "inputSchema": { + "type": "object", + "properties": { + "spec": {"type": "string", "description": "Path to the spec file (.spec.ts, .cy.js, test_*.py …)"}, + "apply": { + "type": "boolean", + "default": False, + "description": "Write the healed code to the spec in place (default: save a .healed copy)", + }, + "max_iterations": {"type": "integer", "default": 3, "minimum": 1, "maximum": 9}, + "framework": { + "type": "string", + "enum": ["auto", "playwright", "cypress", "selenium"], + "default": "auto", + }, + }, + "required": ["spec"], + }, + }, + { + "name": "run_test", + "title": "Run a test", + "description": "Run a test spec without healing. Returns pass/fail and the failure details.", + "inputSchema": { + "type": "object", + "properties": { + "spec": {"type": "string", "description": "Path to the spec file"}, + "framework": { + "type": "string", + "enum": ["auto", "playwright", "cypress", "selenium"], + "default": "auto", + }, + }, + "required": ["spec"], + }, + }, +] + + +def serve(stdin=None, stdout=None) -> int: + """Blocking newline-delimited JSON-RPC loop. Returns on EOF.""" + stdin = stdin if stdin is not None else sys.stdin + stdout = stdout if stdout is not None else sys.stdout + + for line in stdin: + line = line.strip() + if not line: + continue + try: + message = json.loads(line) + except json.JSONDecodeError: + _send(stdout, {"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": "Parse error"}}) + continue + response = _handle(message) + if response is not None: + _send(stdout, response) + return 0 + + +def _send(stdout, payload: dict) -> None: + stdout.write(json.dumps(payload, ensure_ascii=False) + "\n") + stdout.flush() + + +def _handle(message: dict): + """Dispatch one JSON-RPC message; returns a response dict or None for notifications.""" + method = message.get("method", "") + msg_id = message.get("id") + params = message.get("params") or {} + is_notification = "id" not in message + + if method == "initialize": + return _result( + msg_id, + { + "protocolVersion": params.get("protocolVersion") or PROTOCOL_VERSION, + "capabilities": {"tools": {}}, + "serverInfo": SERVER_INFO, + "instructions": INSTRUCTIONS, + }, + ) + if method.startswith("notifications/"): + return None + if method == "ping": + return _result(msg_id, {}) + if method == "tools/list": + return _result(msg_id, {"tools": TOOLS}) + if method == "tools/call": + return _call_tool(msg_id, params) + if is_notification: + return None + return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": f"Method not found: {method}"}} + + +def _result(msg_id, result: dict) -> dict: + return {"jsonrpc": "2.0", "id": msg_id, "result": result} + + +def _tool_text(msg_id, payload: dict, is_error: bool = False) -> dict: + text = json.dumps(payload, indent=2, ensure_ascii=False) + return _result(msg_id, {"content": [{"type": "text", "text": text}], "isError": is_error}) + + +def _call_tool(msg_id, params: dict) -> dict: + from .runner.execute import RunnerError + + name = params.get("name") + args = params.get("arguments") or {} + try: + if name == "heal_test": + return _tool_text(msg_id, _heal_test(args)) + if name == "run_test": + return _tool_text(msg_id, _run_test(args)) + return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32602, "message": f"Unknown tool: {name}"}} + except RunnerError as e: + return _tool_text(msg_id, {"error": str(e)}, is_error=True) + except Exception as e: # tool failures must come back as results, not crash the server + logger.exception("tool %s failed", name) + return _tool_text(msg_id, {"error": f"{type(e).__name__}: {e}"}, is_error=True) + + +def _heal_test(args: dict) -> dict: + from .cli import heal_one + + spec = Path(args["spec"]).expanduser() + apply = bool(args.get("apply", False)) + # The heal loop prints progress to stdout, which would corrupt the + # protocol stream — reroute it to stderr (host shows it as server logs). + with contextlib.redirect_stdout(sys.stderr): + outcome = heal_one( + spec, + auto_apply=apply, + max_iterations=int(args.get("max_iterations", 3)), + framework=args.get("framework", "auto"), + interactive=False, + ) + payload = { + "spec": str(spec), + "status": outcome.status, + "detail": outcome.detail, + "applied": apply and outcome.status == "healed", + } + if outcome.status == "healed" and not apply: + payload["healed_copy"] = str(spec.with_suffix(spec.suffix + ".healed")) + if outcome.diff: + payload["diff"] = outcome.diff + if outcome.status == "needs-human": + payload["note"] = "possible real bug — 9lives refuses to rewrite assertions to force a pass" + return payload + + +def _run_test(args: dict) -> dict: + from .cli import run_one + + spec = Path(args["spec"]).expanduser() + with contextlib.redirect_stdout(sys.stderr): + outcome = run_one(spec, framework=args.get("framework", "auto")) + return {"spec": str(spec), "status": outcome.status, "detail": outcome.detail} diff --git a/src/ninelives/watch.py b/src/ninelives/watch.py new file mode 100644 index 0000000..8a7112d --- /dev/null +++ b/src/ninelives/watch.py @@ -0,0 +1,92 @@ +"""`9l watch` — heal on save. + +Polls spec files for changes (zero dependencies, works everywhere) and runs +the heal loop on whatever changed — so healing becomes part of the local +edit-save loop, not just a CI afterthought. +""" + +import logging +import os +import re +import time +from pathlib import Path + +logger = logging.getLogger(__name__) + +DEFAULT_INTERVAL_SECONDS = 1.0 + +# What counts as a spec file when watching a directory. +SPEC_NAME = re.compile( + r"(\.(spec|test)\.[cm]?[jt]sx?$)|(\.cy\.[cm]?[jt]sx?$)|(^test_.*\.py$)|(_test\.py$)", + re.IGNORECASE, +) + +IGNORED_DIRS = { + "node_modules", + ".git", + ".9lives", + "__pycache__", + ".venv", + "venv", + "dist", + "build", + "test-results", + "playwright-report", + "cypress", # cypress/{videos,screenshots,downloads} churn on every run +} + + +def is_spec_file(path: Path) -> bool: + name = path.name + if name.startswith("_9lives_heal_") or name.endswith(".healed"): + return False # our own working copies / saved heals + return bool(SPEC_NAME.search(name)) + + +def scan(targets: list[Path]) -> dict[Path, float]: + """Snapshot mtimes of every watched spec file.""" + snapshot: dict[Path, float] = {} + for target in targets: + target = target.resolve() + if target.is_file(): + if is_spec_file(target): + snapshot[target] = target.stat().st_mtime + continue + for dirpath, dirnames, filenames in os.walk(target): + dirnames[:] = [d for d in dirnames if d not in IGNORED_DIRS and not d.startswith(".")] + for filename in filenames: + path = Path(dirpath) / filename + if is_spec_file(path): + try: + snapshot[path] = path.stat().st_mtime + except OSError: + continue + return snapshot + + +def changed_specs(before: dict[Path, float], after: dict[Path, float]) -> list[Path]: + """Files that are new or have a newer mtime, oldest change first.""" + changed = [path for path, mtime in after.items() if mtime > before.get(path, -1.0)] + return sorted(changed, key=lambda p: after[p]) + + +def watch_loop(targets: list[Path], on_change, interval: float = DEFAULT_INTERVAL_SECONDS) -> int: + """Poll until Ctrl-C, invoking `on_change(path)` for each changed spec. + + After the callback runs, the snapshot is re-taken so a heal that writes + the spec back doesn't re-trigger itself. + """ + snapshot = scan(targets) + print(f"🐾 watching {len(snapshot)} spec file(s) — save a file to heal it (Ctrl-C to stop)") + try: + while True: + time.sleep(interval) + current = scan(targets) + for path in changed_specs(snapshot, current): + # Let the editor finish writing before we run. + time.sleep(0.2) + on_change(path) + snapshot = scan(targets) + except KeyboardInterrupt: + print("\n🐾 watch stopped.") + return 0 diff --git a/tests/test_expansion.py b/tests/test_expansion.py new file mode 100644 index 0000000..25463d8 --- /dev/null +++ b/tests/test_expansion.py @@ -0,0 +1,276 @@ +"""Tests for the expansion features: framework adapters (Cypress/Selenium), +heal history + brittle-selector report, the MCP server, and watch mode.""" + +import asyncio +import io +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from ninelives.frameworks import detect_framework, get_adapter # noqa: E402 +from ninelives.frameworks.cypress import parse_mocha_json # noqa: E402 +from ninelives.frameworks.selenium import parse_pytest_output # noqa: E402 +from ninelives.healing.parse import extract_failed_selector # noqa: E402 +from ninelives.healing.strategy import FailureType, TestFailure, healing_strategy_selector # noqa: E402 +from ninelives.healing.tier1 import tier1_healer # noqa: E402 + + +# ---------- framework detection ---------- + + +def test_detect_framework_by_suffix(tmp_path): + assert detect_framework(tmp_path / "login.cy.ts") == "cypress" + assert detect_framework(tmp_path / "login.cy.js") == "cypress" + assert detect_framework(tmp_path / "test_login.py") == "selenium" + assert detect_framework(tmp_path / "login.spec.ts") == "playwright" # bare spec, no project + + +def test_detect_framework_from_package_json(tmp_path): + (tmp_path / "package.json").write_text('{"devDependencies": {"cypress": "13.0.0"}}') + spec = tmp_path / "login.spec.js" + spec.write_text("it('x', () => {});\n") + assert detect_framework(spec) == "cypress" + + # A project with BOTH stays playwright — the .cy. naming is the cypress signal there. + (tmp_path / "package.json").write_text('{"devDependencies": {"cypress": "13.0.0", "@playwright/test": "1.61.1"}}') + assert detect_framework(spec) == "playwright" + + +def test_get_adapter_explicit_override_beats_detection(tmp_path): + assert get_adapter(tmp_path / "login.spec.js", "selenium").name == "selenium" + assert get_adapter(tmp_path / "login.cy.js", "auto").name == "cypress" + + +# ---------- cypress ---------- + +CYPRESS_LOCATOR_MSG = ( + "AssertionError: Timed out retrying after 4000ms: Expected to find element: `#login-btn`, but never found it." +) + + +def test_cypress_locator_miss_is_not_an_assertion(): + # Cypress wraps locator misses in AssertionError — the behavior-vs-drift + # guard must NOT eat them; they are selector drift and healable. + assert healing_strategy_selector.classify_failure(CYPRESS_LOCATOR_MSG) == FailureType.LOCATOR_NOT_FOUND + + +def test_cypress_selector_extraction(): + assert extract_failed_selector(CYPRESS_LOCATOR_MSG) == "#login-btn" + assert extract_failed_selector("something failed", "cy.get('.submit-button').click()") == ".submit-button" + + +def test_parse_mocha_json_survives_banner_noise(): + stdout = ( + "It looks like this is your first time using Cypress\n" + + json.dumps( + { + "stats": {"duration": 1234}, + "failures": [ + { + "fullTitle": "login signs in", + "err": {"message": "Expected to find element: `#x`", "stack": "at ..."}, + } + ], + } + ) + + "\ntrailing noise" + ) + errors, duration = parse_mocha_json(stdout) + assert duration == 1234 + assert len(errors) == 1 + assert errors[0].title == "login signs in" + assert "#x" in errors[0].message + + +# ---------- selenium ---------- + +SELENIUM_MSG = ( + "selenium.common.exceptions.NoSuchElementException: Message: no such element: " + 'Unable to locate element: {"method":"css selector","selector":"#submit-btn"}' +) + + +def test_selenium_no_such_element_classifies_as_locator(): + assert healing_strategy_selector.classify_failure(SELENIUM_MSG) == FailureType.LOCATOR_NOT_FOUND + + +def test_selenium_selector_extraction_from_json_payload(): + assert extract_failed_selector(SELENIUM_MSG) == "#submit-btn" + + +def test_selenium_selector_extraction_from_by_call(): + stack = 'driver.find_element(By.CSS_SELECTOR, "#old-btn").click()' + assert extract_failed_selector("NoSuchElementException", stack) == "#old-btn" + + +def test_parse_pytest_output_extracts_failure(): + stdout = "\n".join( + [ + "____________________ test_login ____________________", + "", + " def test_login(driver):", + '> driver.find_element(By.CSS_SELECTOR, "#login-btn").click()', + "", + "E " + SELENIUM_MSG, + "", + "FAILED test_login.py::test_login - selenium.common.exceptions.NoSuchElementException", + "1 failed in 1.23s", + ] + ) + errors = parse_pytest_output(stdout) + assert len(errors) == 1 + assert errors[0].title == "test_login" + assert "NoSuchElementException" in errors[0].message + assert "find_element" in errors[0].stack + + +def test_tier1_never_injects_playwright_waits_into_other_frameworks(): + base = dict( + failure_type=FailureType.LOCATOR_TIMEOUT, + error_message="Timeout 30000ms exceeded waiting for selector", + failed_selector="button", # immune to transformations → reaches the timing branch + test_code="await page.locator('button').click();", + ) + cypress = asyncio.run(tier1_healer.heal(TestFailure(**base, framework="cypress"))) + assert not cypress.success # escalates instead of writing `await page.…` into Cypress code + + playwright = asyncio.run(tier1_healer.heal(TestFailure(**base, framework="playwright"))) + assert playwright.success + assert "waitFor" in playwright.healed_code + + +# ---------- heal history / report ---------- + + +def test_history_record_and_report(tmp_path): + from ninelives.history import load_history, record_heal, render_report, selector_report + + spec = tmp_path / "app" / "login.spec.js" + spec.parent.mkdir(parents=True) + spec.write_text("x") + + record_heal( + spec, framework="playwright", status="healed", failed_selector="#login", anchor="text", healed_selector="#signin" + ) + record_heal(spec, framework="playwright", status="healed", failed_selector="#login", anchor="text") + record_heal(spec, framework="playwright", status="needs-human", failed_selector=None, failure_type="assertion_failed") + + records = load_history(tmp_path) # rglob finds the nested .9lives/history.jsonl + assert len(records) == 3 + + rows = selector_report(records) + assert rows[0]["selector"] == "#login" + assert rows[0]["heals"] == 2 + assert "data-testid" in rows[0]["recommendation"] # healed 2× → pin something stable + + terminal = render_report(rows) + assert "#login" in terminal + markdown = render_report(rows, markdown=True) + assert "| `#login` |" in markdown + + +def test_history_recording_never_breaks_healing(tmp_path, monkeypatch): + # A read-only .9lives dir (or any history failure) must not crash the heal loop. + from ninelives import cli + from ninelives.frameworks import get_adapter + + spec = tmp_path / "login.spec.js" + spec.write_text("x") + monkeypatch.setattr("ninelives.history.record_heal", lambda *a, **k: (_ for _ in ()).throw(OSError("disk full"))) + cli._record_history(spec, get_adapter(spec), "healed") # must not raise + + +# ---------- MCP server ---------- + + +def _mcp_roundtrip(messages): + from ninelives import mcp_server + + stdin = io.StringIO("\n".join(json.dumps(m) for m in messages) + "\n") + stdout = io.StringIO() + mcp_server.serve(stdin=stdin, stdout=stdout) + return [json.loads(line) for line in stdout.getvalue().splitlines()] + + +def test_mcp_initialize_and_tools_list(): + responses = _mcp_roundtrip( + [ + {"jsonrpc": "2.0", "id": 0, "method": "initialize", "params": {"protocolVersion": "2025-03-26"}}, + {"jsonrpc": "2.0", "method": "notifications/initialized"}, + {"jsonrpc": "2.0", "id": 1, "method": "tools/list"}, + {"jsonrpc": "2.0", "id": 2, "method": "nope/nope"}, + ] + ) + assert responses[0]["result"]["protocolVersion"] == "2025-03-26" # echoes the client's version + assert responses[0]["result"]["serverInfo"]["name"] == "9lives" + tool_names = {t["name"] for t in responses[1]["result"]["tools"]} + assert tool_names == {"heal_test", "run_test"} + assert responses[2]["error"]["code"] == -32601 + + +def test_mcp_heal_test_returns_structured_result(tmp_path, monkeypatch): + from ninelives import cli + from ninelives.report.github import SpecOutcome + + seen = {} + + def fake_heal_one(spec, **kwargs): + seen.update(kwargs, spec=spec) + print("progress chatter that must NOT hit the protocol stream") + return SpecOutcome(spec=spec.name, status="healed", detail="not applied — saved", diff="--- a\n+++ b") + + monkeypatch.setattr(cli, "heal_one", fake_heal_one) + spec = tmp_path / "login.spec.js" + + responses = _mcp_roundtrip( + [ + { + "jsonrpc": "2.0", + "id": 7, + "method": "tools/call", + "params": {"name": "heal_test", "arguments": {"spec": str(spec)}}, + }, + {"jsonrpc": "2.0", "id": 8, "method": "tools/call", "params": {"name": "bogus", "arguments": {}}}, + ] + ) + assert seen["interactive"] is False # MCP must never block on input() + assert seen["auto_apply"] is False + + result = responses[0]["result"] + assert result["isError"] is False + payload = json.loads(result["content"][0]["text"]) + assert payload["status"] == "healed" + assert payload["healed_copy"].endswith(".spec.js.healed") + assert payload["applied"] is False + assert "diff" in payload + + assert responses[1]["error"]["code"] == -32602 + + +# ---------- watch ---------- + + +def test_watch_scan_finds_specs_and_ignores_noise(tmp_path): + from ninelives.watch import changed_specs, scan + + (tmp_path / "login.spec.js").write_text("x") + (tmp_path / "checkout.cy.ts").write_text("x") + (tmp_path / "test_flow.py").write_text("x") + (tmp_path / "_9lives_heal_1_login.spec.js").write_text("x") # our own working copy + (tmp_path / "login.spec.js.healed").write_text("x") + (tmp_path / "notes.txt").write_text("x") + (tmp_path / "node_modules").mkdir() + (tmp_path / "node_modules" / "dep.spec.js").write_text("x") + + snapshot = scan([tmp_path]) + names = {p.name for p in snapshot} + assert names == {"login.spec.js", "checkout.cy.ts", "test_flow.py"} + + import os + + target = tmp_path / "login.spec.js" + os.utime(target, (target.stat().st_atime, target.stat().st_mtime + 5)) + changed = changed_specs(snapshot, scan([tmp_path])) + assert [p.name for p in changed] == ["login.spec.js"] diff --git a/tests/test_healing.py b/tests/test_healing.py index 160b67b..f1b1920 100644 --- a/tests/test_healing.py +++ b/tests/test_healing.py @@ -291,8 +291,10 @@ def fake_run_spec(working_spec): seen["existed"] = Path(working_spec).exists() return RunResult(passed=True, exit_code=0) + from ninelives.runner import execute + monkeypatch.setattr(cli, "find_user_project", lambda p: tmp_path) - monkeypatch.setattr(cli, "run_spec", fake_run_spec) + monkeypatch.setattr(execute, "run_spec", fake_run_spec) # the playwright adapter routes through here outcome = cli.heal_one(spec, auto_apply=True) assert outcome.status == "passed" @@ -329,8 +331,10 @@ async def fake_heal(failure): confidence=0.85, ) + from ninelives.runner import execute + monkeypatch.setattr(cli, "find_user_project", lambda p: None) - monkeypatch.setattr(cli, "run_spec", fake_run_spec) + monkeypatch.setattr(execute, "run_spec", fake_run_spec) # the playwright adapter routes through here monkeypatch.setattr(cli.healing_strategy_selector, "classify_failure", lambda *a, **k: FailureType.LOCATOR_NOT_FOUND) monkeypatch.setattr(cli.healing_strategy_selector, "select_strategy", lambda f: HealingTier.TIER1_AUTO) monkeypatch.setattr(cli.tier1_healer, "heal", fake_heal) From 0e537ce31e40eea8231256a00eb92188d95a5f7e Mon Sep 17 00:00:00 2001 From: Ruslan Strazhnyk Date: Fri, 10 Jul 2026 08:45:35 +0200 Subject: [PATCH 2/2] fix(security): harden agent-callable surfaces per QualityMax SAST review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - MCP heal_test/run_test: validate the tool-supplied spec path — must be an existing file with a recognized test suffix, contained in the server's working directory (NINELIVES_MCP_UNRESTRICTED=1 to lift). The real risk isn't reading files: healing a spec ultimately EXECUTES it (pytest import / test run), so cwd containment is the meaningful boundary for a prompt-injected agent. - Selenium adapter: reject non-.py specs in preflight. The subprocess call was already injection-safe (argv list, no shell, resolved absolute path can't parse as a pytest flag) — this is defense-in-depth for the MCP path. 3 new tests (51 total). --- README.md | 2 +- src/ninelives/frameworks/selenium.py | 8 +++ src/ninelives/mcp_server.py | 35 ++++++++-- tests/test_expansion.py | 97 +++++++++++++++++++++++++++- 4 files changed, 135 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 00b430b..68f8e53 100644 --- a/README.md +++ b/README.md @@ -70,7 +70,7 @@ claude mcp add 9lives -- uvx --from 9lives 9l mcp { "mcpServers": { "9lives": { "command": "9l", "args": ["mcp"] } } } ``` -The behavior-vs-drift guard applies to agents too: a failing assertion comes back as `needs-human`, with an explicit note that forcing it green would mask a real bug. +The behavior-vs-drift guard applies to agents too: a failing assertion comes back as `needs-human`, with an explicit note that forcing it green would mask a real bug. `heal_test` only accepts existing test files under the directory `9l mcp` was started in (`NINELIVES_MCP_UNRESTRICTED=1` lifts this), since healing a spec ultimately executes it. ## Heal on save & pre-commit diff --git a/src/ninelives/frameworks/selenium.py b/src/ninelives/frameworks/selenium.py index b53d967..5f699ae 100644 --- a/src/ninelives/frameworks/selenium.py +++ b/src/ninelives/frameworks/selenium.py @@ -25,6 +25,14 @@ class SeleniumAdapter: language = "python" def preflight(self, spec: Path) -> None: + # Defense-in-depth for the agent-callable MCP path: run() below invokes + # pytest via an argv list (never a shell), and the spec path is resolved + # to an absolute path before being passed as an argument, so it can't be + # parsed as a pytest flag or injected into a shell. Still, only .py files + # make sense to run through pytest at all — reject anything else early, + # before the PATH check, so this fails even without pytest installed. + if spec.suffix.lower() != ".py": + raise RunnerError(f"Selenium specs run through pytest and must be .py files (got {spec.name})") if shutil.which("pytest") is None: raise RunnerError( "'pytest' not found on PATH — Selenium specs run through your own pytest:\n" diff --git a/src/ninelives/mcp_server.py b/src/ninelives/mcp_server.py index e3ea5ee..51828f8 100644 --- a/src/ninelives/mcp_server.py +++ b/src/ninelives/mcp_server.py @@ -14,10 +14,12 @@ import contextlib import json import logging +import os import sys from pathlib import Path from . import __version__ +from .runner.execute import RunnerError logger = logging.getLogger(__name__) @@ -30,9 +32,34 @@ "it re-runs the spec, repairs drifted selectors (offline Tier 1, LLM Tier 2), verifies green, " "and returns the diff. With apply=false the fix is saved next to the spec as .healed " "for you to review/apply; with apply=true it is written in place. " - "A needs-human status usually means a failing assertion — a possible real bug 9lives refuses to mask." + "A needs-human status usually means a failing assertion — a possible real bug 9lives refuses to mask. " + "Specs must be existing test files under the server's working directory." ) +_SPEC_SUFFIXES = {".js", ".mjs", ".cjs", ".ts", ".jsx", ".tsx", ".py"} + + +def _validated_spec(raw: str) -> Path: + """Resolve and validate a tool-supplied spec path. + + The MCP surface is agent-callable, and healing a spec ultimately + executes it (pytest import / test run) — so specs must be real test + files inside the server's working directory. Set + NINELIVES_MCP_UNRESTRICTED=1 to lift the cwd containment. + """ + spec = Path(raw).expanduser().resolve() + if not spec.is_file(): + raise RunnerError(f"spec not found: {spec}") + if spec.suffix.lower() not in _SPEC_SUFFIXES: + raise RunnerError(f"not a recognized test spec ({spec.suffix or 'no extension'}): {spec}") + if os.environ.get("NINELIVES_MCP_UNRESTRICTED") != "1" and not spec.is_relative_to(Path.cwd().resolve()): + raise RunnerError( + f"spec is outside the working directory: {spec} — run `9l mcp` from the project root, " + "or set NINELIVES_MCP_UNRESTRICTED=1 to allow it" + ) + return spec + + TOOLS = [ { "name": "heal_test", @@ -146,8 +173,6 @@ def _tool_text(msg_id, payload: dict, is_error: bool = False) -> dict: def _call_tool(msg_id, params: dict) -> dict: - from .runner.execute import RunnerError - name = params.get("name") args = params.get("arguments") or {} try: @@ -166,7 +191,7 @@ def _call_tool(msg_id, params: dict) -> dict: def _heal_test(args: dict) -> dict: from .cli import heal_one - spec = Path(args["spec"]).expanduser() + spec = _validated_spec(args["spec"]) apply = bool(args.get("apply", False)) # The heal loop prints progress to stdout, which would corrupt the # protocol stream — reroute it to stderr (host shows it as server logs). @@ -196,7 +221,7 @@ def _heal_test(args: dict) -> dict: def _run_test(args: dict) -> dict: from .cli import run_one - spec = Path(args["spec"]).expanduser() + spec = _validated_spec(args["spec"]) with contextlib.redirect_stdout(sys.stderr): outcome = run_one(spec, framework=args.get("framework", "auto")) return {"spec": str(spec), "status": outcome.status, "detail": outcome.detail} diff --git a/tests/test_expansion.py b/tests/test_expansion.py index 25463d8..49a26ff 100644 --- a/tests/test_expansion.py +++ b/tests/test_expansion.py @@ -7,14 +7,17 @@ import sys from pathlib import Path +import pytest + sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from ninelives.frameworks import detect_framework, get_adapter # noqa: E402 from ninelives.frameworks.cypress import parse_mocha_json # noqa: E402 -from ninelives.frameworks.selenium import parse_pytest_output # noqa: E402 +from ninelives.frameworks.selenium import SeleniumAdapter, parse_pytest_output # noqa: E402 from ninelives.healing.parse import extract_failed_selector # noqa: E402 from ninelives.healing.strategy import FailureType, TestFailure, healing_strategy_selector # noqa: E402 from ninelives.healing.tier1 import tier1_healer # noqa: E402 +from ninelives.runner.execute import RunnerError # noqa: E402 # ---------- framework detection ---------- @@ -222,7 +225,9 @@ def fake_heal_one(spec, **kwargs): return SpecOutcome(spec=spec.name, status="healed", detail="not applied — saved", diff="--- a\n+++ b") monkeypatch.setattr(cli, "heal_one", fake_heal_one) + monkeypatch.chdir(tmp_path) spec = tmp_path / "login.spec.js" + spec.write_text("x") responses = _mcp_roundtrip( [ @@ -274,3 +279,93 @@ def test_watch_scan_finds_specs_and_ignores_noise(tmp_path): os.utime(target, (target.stat().st_atime, target.stat().st_mtime + 5)) changed = changed_specs(snapshot, scan([tmp_path])) assert [p.name for p in changed] == ["login.spec.js"] + + +# ---------- security hardening ---------- + + +def test_selenium_preflight_rejects_non_py(tmp_path): + spec = tmp_path / "login.spec.js" + spec.write_text("x") + + with pytest.raises(RunnerError, match=r"\.py"): + SeleniumAdapter().preflight(spec) + + +def test_mcp_rejects_spec_outside_cwd(tmp_path, monkeypatch): + from ninelives import cli + from ninelives.report.github import SpecOutcome + + project = tmp_path / "proj" + project.mkdir() + outside_dir = tmp_path / "outside" + outside_dir.mkdir() + outside = outside_dir / "login.spec.js" + outside.write_text("x") + + monkeypatch.chdir(project) + + responses = _mcp_roundtrip( + [ + { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": {"name": "heal_test", "arguments": {"spec": str(outside)}}, + } + ] + ) + result = responses[0]["result"] + assert result["isError"] is True + text = result["content"][0]["text"] + assert "outside the working directory" in text + + def fake_heal_one(spec, **kwargs): + return SpecOutcome(spec=spec.name, status="healed", detail="not applied — saved", diff="--- a\n+++ b") + + monkeypatch.setattr(cli, "heal_one", fake_heal_one) + monkeypatch.setenv("NINELIVES_MCP_UNRESTRICTED", "1") + + responses = _mcp_roundtrip( + [ + { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": {"name": "heal_test", "arguments": {"spec": str(outside)}}, + } + ] + ) + result = responses[0]["result"] + assert result["isError"] is False + + +def test_mcp_rejects_non_spec_file(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + notes = tmp_path / "notes.txt" + notes.write_text("x") + + responses = _mcp_roundtrip( + [ + { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": {"name": "heal_test", "arguments": {"spec": str(notes)}}, + }, + { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": {"name": "heal_test", "arguments": {"spec": str(tmp_path / "missing.spec.js")}}, + }, + ] + ) + + result = responses[0]["result"] + assert result["isError"] is True + assert "not a recognized test spec" in result["content"][0]["text"] + + result = responses[1]["result"] + assert result["isError"] is True + assert "spec not found" in result["content"][0]["text"]