cleveragents-core/noxfile.py

import json
import os
import sys
from pathlib import Path

import nox

# Global configuration
DEFAULT_PYTHON = "3.13"
SUPPORTED_PYTHONS = ["3.13"]
nox.options.reuse_existing_virtualenvs = True
nox.options.error_on_external_run = True

BEHAVE_PARALLEL_VERSION = "2.0.0"


def _default_processes() -> int:
    env_override = os.environ.get("TEST_PROCESSES")
    if env_override:
        return int(env_override)
    try:
        cpus = len(os.sched_getaffinity(0)) or 1
    except AttributeError:
        cpus = os.cpu_count() or 1
    # Keep default parallelism conservative to avoid timeout/OOM flakes
    # under heavy Robot/pabot subprocess fan-out in CI and shared runners.
    # Callers can still override with TEST_PROCESSES / --processes.
    return min(cpus, 2)


def _behave_parallel_args(posargs: list[str]) -> list[str]:
    has_custom_processes = any(
        arg in {"--processes", "-j"} or arg.startswith(("--processes=", "-j"))
        for arg in posargs
    )
    if has_custom_processes:
        return []
    return ["--processes", str(_default_processes())]


def _pabot_parallel_args(posargs: list[str]) -> list[str]:
    has_custom_processes = any(
        arg in {"--processes"} or arg.startswith("--processes=") for arg in posargs
    )
    if has_custom_processes:
        return []
    return ["--processes", str(_default_processes())]


def _split_pabot_args(posargs: list[str]) -> tuple[list[str], list[str]]:
    pabot_args: list[str] = []
    robot_args: list[str] = []
    i = 0
    while i < len(posargs):
        arg = posargs[i]
        if arg == "--processes" and i + 1 < len(posargs):
            pabot_args.extend([arg, posargs[i + 1]])
            i += 2
            continue
        if arg.startswith("--processes="):
            pabot_args.append(arg)
            i += 1
            continue
        robot_args.append(arg)
        i += 1
    return pabot_args, robot_args


def _create_template_db(session: nox.Session) -> str:
    """Build the pre-migrated SQLite template and return its path.

    Runs ``scripts/create_template_db.py`` to produce a SQLite file with all
    tables already created via ``Base.metadata.create_all()`` and the
    alembic_version table stamped at HEAD.  Each test scenario can then
    ``shutil.copy`` this file instead of running 25 Alembic migrations.
    """
    template_path = str(Path("build/.template-migrated.db").resolve())
    session.run(
        "python",
        "scripts/create_template_db.py",
        template_path,
        silent=True,
    )
    return template_path


def _install_behave_parallel(session: nox.Session) -> None:
    """Install behave-parallel with an in-process parallel runner.

    Instead of spawning one subprocess per feature file (the old model),
    the new CLI runs features in-process via behave's Python API.  Step
    definitions and environment hooks are loaded once; each feature file
    is then parsed and executed without interpreter startup overhead.

    Parallel execution uses ``multiprocessing.Pool`` with the ``fork``
    start method so that already-imported modules are shared (copy-on-write)
    across workers.
    """
    tmp_dir = Path(session.create_tmp())
    source_dir = tmp_dir / "behave-parallel-inprocess"
    source_dir.mkdir(parents=True, exist_ok=True)

    pkg_dir = source_dir / "behave_parallel"
    pkg_dir.mkdir(parents=True, exist_ok=True)
    (pkg_dir / "__init__.py").write_text("\n")
    (pkg_dir / "cli.py").write_text(_BEHAVE_PARALLEL_CLI_SOURCE)

    setup_path = source_dir / "setup.py"
    setup_path.write_text(
        "from setuptools import find_packages, setup\n"
        "setup(\n"
        '    name="behave-parallel",\n'
        f'    version="{BEHAVE_PARALLEL_VERSION}",\n'
        "    packages=find_packages(),\n"
        '    install_requires=["behave>=1.2.6"],\n'
        "    entry_points={\n"
        '        "console_scripts": '
        '["behave-parallel=behave_parallel.cli:main"],\n'
        "    },\n"
        ")\n"
    )

    session.install("setuptools", "wheel")
    session.install(str(source_dir))


# The in-process behave-parallel CLI source is kept as a module-level
# constant so the noxfile itself stays readable and ``ruff`` can still
# lint the surrounding Python without tripping over a giant raw string
# embedded inside a function body.
_BEHAVE_PARALLEL_CLI_SOURCE = r'''
"""In-process parallel behave runner.

Replaces the old subprocess-per-feature model with direct use of
behave's ``Runner`` API.  Step definitions and environment hooks are
loaded once per process; feature files are parsed and executed without
Python interpreter startup overhead.

Parallelism modes
-----------------
* **Sequential** (``--processes 1`` or ``BEHAVE_PARALLEL_COVERAGE=1``):
  All features run in a single ``Runner.run()`` call.
* **Parallel** (``--processes N``, N > 1, no coverage):
  Features are split into *N* equal-size chunks.  A
  ``multiprocessing.Pool`` with the ``fork`` start method dispatches
  each chunk to a worker that creates its own ``Runner`` (hooks and
  step definitions are re-loaded cheaply because all heavy modules are
  already in memory from the parent).
"""

from __future__ import annotations

import argparse
import io
import multiprocessing
import os
import sys
import time
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path

DEFAULT_FEATURE_ROOT = "features/"


# ---------------------------------------------------------------------------
# Summary helpers
# ---------------------------------------------------------------------------

def _empty_summary():
    return {
        "features": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
        "scenarios": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
        "steps": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
        "duration": 0.0,
    }


def _extract_summary(runner):
    """Build a summary dict from a completed behave Runner."""
    summary = _empty_summary()
    for feature in runner.features:
        status_name = feature.status.name
        if status_name == "passed":
            summary["features"]["passed"] += 1
        elif status_name == "skipped":
            summary["features"]["skipped"] += 1
        else:
            summary["features"]["failed"] += 1

        summary["duration"] += feature.duration or 0.0

        for scenario in feature.walk_scenarios():
            sname = scenario.status.name
            if sname == "passed":
                summary["scenarios"]["passed"] += 1
            elif sname == "skipped":
                summary["scenarios"]["skipped"] += 1
            elif sname == "failed":
                summary["scenarios"]["failed"] += 1
            else:
                summary["scenarios"]["errors"] += 1

            for step in scenario.steps:
                stname = step.status.name
                if stname == "passed":
                    summary["steps"]["passed"] += 1
                elif stname == "skipped":
                    summary["steps"]["skipped"] += 1
                elif stname == "failed":
                    summary["steps"]["failed"] += 1
                else:
                    summary["steps"]["errors"] += 1
    return summary


def _merge_summaries(summaries):
    total = _empty_summary()
    for s in summaries:
        for bucket in ("features", "scenarios", "steps"):
            for field in ("passed", "failed", "errors", "skipped"):
                total[bucket][field] += s.get(bucket, {}).get(field, 0)
        total["duration"] += float(s.get("duration", 0.0))
    return total


def _format_duration(seconds):
    minutes = int(seconds // 60)
    remainder = seconds % 60
    if minutes:
        return f"{minutes}m {remainder:.3f}s"
    return f"{remainder:.3f}s"


def _print_overall_summary(total, wall_seconds=None):
    print("\nOverall summary:")
    for bucket in ("features", "scenarios", "steps"):
        b = total[bucket]
        print(
            f"{b['passed']} {bucket} passed, {b['failed']} failed, "
            f"{b['errors']} errored, {b['skipped']} skipped"
        )
    if total["duration"]:
        print(f"Took {_format_duration(float(total['duration']))}")
    if wall_seconds is not None:
        print(f"Wall time: {_format_duration(wall_seconds)}")


def _has_failures(total):
    return (
        total["features"]["failed"] > 0
        or total["features"]["errors"] > 0
        or total["scenarios"]["failed"] > 0
        or total["scenarios"]["errors"] > 0
    )


def _no_scenarios_ran(total):
    """Return True when no scenario reached a terminal state.

    This catches runner-level crashes (e.g. ``before_all`` failure) that
    prevent any scenario from executing.  Without this guard the summary
    would contain all-zero counters, ``_has_failures()`` would return
    ``False``, and CI would silently pass a broken suite.
    """
    s = total["scenarios"]
    return s["passed"] + s["failed"] + s["errors"] + s["skipped"] == 0


# ---------------------------------------------------------------------------
# Feature discovery
# ---------------------------------------------------------------------------

def _iter_features(paths):
    collected = []
    for path in paths:
        p = Path(path)
        if p.is_dir():
            collected.extend(sorted(str(fp) for fp in p.rglob("*.feature")))
        else:
            collected.append(str(p))
    return collected


def _extract_features_and_args(argv):
    positional = []
    options = []
    for arg in argv:
        if arg.startswith("-"):
            options.append(arg)
        else:
            positional.append(arg)
    if positional:
        return positional, options
    return [DEFAULT_FEATURE_ROOT], options


# ---------------------------------------------------------------------------
# In-process behave execution
# ---------------------------------------------------------------------------

def _make_runner(feature_paths, behave_args):
    """Create a behave Runner with proper configuration defaults.

    Mirrors the format-defaulting logic from ``behave.__main__.run_behave``
    so that ``-q`` and bare invocations get a sensible formatter instead
    of crashing on ``config.format is None``.
    """
    from behave.configuration import Configuration
    from behave.runner import Runner
    from behave.runner_util import reset_runtime

    reset_runtime()
    args = list(behave_args) + [str(p) for p in feature_paths]
    config = Configuration(command_args=args)
    if not config.format:
        config.format = [config.default_format]
    return Runner(config)


def _run_features_inprocess(feature_paths, behave_args):
    """Run *all* feature_paths in a single behave Runner invocation.

    Returns ``(failed: bool, summary: dict)``.
    """
    runner = _make_runner(feature_paths, behave_args)
    failed = runner.run()
    summary = _extract_summary(runner)
    return failed, summary


def _worker_run_features(payload):
    """Entry point for multiprocessing workers.

    Runs a chunk of feature files in a forked child process.  Heavy
    Python modules (cleveragents, behave, SQLAlchemy, ...) are already
    loaded in the parent and shared via copy-on-write after ``fork``.

    ``load_step_definitions()`` and ``load_hooks()`` still execute
    inside each worker (they ``exec()`` the step .py files and
    environment.py), but every ``import`` they trigger is a cache hit.
    """
    feature_paths, behave_args = payload

    stdout_buf = io.StringIO()
    stderr_buf = io.StringIO()

    runner = _make_runner(feature_paths, behave_args)

    with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
        failed = runner.run()

    summary = _extract_summary(runner)
    return failed, stdout_buf.getvalue(), stderr_buf.getvalue(), summary


# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------

def main(argv=None):
    argv = list(sys.argv[1:] if argv is None else argv)

    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument("--processes", "-j", type=int, default=None)
    known, remaining = parser.parse_known_args(argv)
    processes = known.processes or os.cpu_count() or 1

    feature_args, other_args = _extract_features_and_args(remaining)

    # Pass-through --help / --version to behave
    if any(flag in remaining for flag in ("-h", "--help", "--version")):
        import subprocess
        code = subprocess.run(
            [sys.executable, "-m", "behave", *other_args, *feature_args]
        ).returncode
        sys.exit(code)

    feature_paths = _iter_features(feature_args)
    if not feature_paths:
        print("No feature files found", file=sys.stderr)
        sys.exit(0)

    coverage_mode = bool(os.environ.get("BEHAVE_PARALLEL_COVERAGE"))

    start = time.monotonic()

    if processes <= 1 or coverage_mode or len(feature_paths) == 1:
        # ---- sequential in-process mode ----
        _, total = _run_features_inprocess(feature_paths, other_args)
    else:
        # ---- parallel in-process mode (multiprocessing fork) ----
        # Pre-import heavy modules so forked children get them for free.
        try:
            import cleveragents  # noqa: F401
        except ImportError:
            pass
        try:
            import behave  # noqa: F401
        except ImportError:
            pass

        # Split features into roughly equal chunks.
        chunk_size = max(1, (len(feature_paths) + processes - 1) // processes)
        chunks = [
            feature_paths[i : i + chunk_size]
            for i in range(0, len(feature_paths), chunk_size)
        ]

        ctx = multiprocessing.get_context("fork")
        with ctx.Pool(processes=min(processes, len(chunks))) as pool:
            results = pool.map(
                _worker_run_features,
                [(chunk, other_args) for chunk in chunks],
            )

        summaries = []
        for _worker_failed, stdout, stderr, summary in results:
            if stdout:
                print(stdout, end="")
            if stderr:
                print(stderr, end="", file=sys.stderr)
            summaries.append(summary)
        total = _merge_summaries(summaries)

    wall = time.monotonic() - start
    _print_overall_summary(total, wall_seconds=wall)

    # Use the summary-based check rather than the raw runner ``failed``
    # boolean.  The ``@tdd_expected_fail`` handler in environment.py
    # inverts scenario statuses for TDD issue-capture tests, but behave's
    # ``runner.run()`` tracks step failures in a local variable that
    # cannot be updated by after_scenario hooks.  Relying solely on the
    # summary (which reflects the corrected scenario statuses) ensures
    # that TDD-inverted scenarios do not cause a spurious exit-code 1.
    if _has_failures(total):
        sys.exit(1)

    # Safety net: if features were requested but zero scenarios ran, the
    # runner crashed before executing any scenario (e.g. ``before_all``
    # failure).  Treat this as a failure so CI does not silently pass.
    if feature_paths and _no_scenarios_ran(total):
        print(
            "ERROR: features were requested but no scenarios ran — "
            "possible runner-level crash.",
            file=sys.stderr,
        )
        sys.exit(1)


if __name__ == "__main__":
    main()
'''


# =============================================================================


# QUICK DEVELOPMENT SESSIONS (Fast - Run daily)
# =============================================================================


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def lint(session: nox.Session):
    """Check code formatting and linting."""
    session.install("ruff>=0.15,<0.16")
    session.run("ruff", "check", "src/", "scripts/", "examples/", "features/", "robot/")


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def format(session: nox.Session):
    """Format code with ruff. Pass --check to verify without modifying."""
    session.install("ruff>=0.15,<0.16")
    session.run("ruff", "format", *session.posargs, ".")


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def typecheck(session: nox.Session):
    """Check types with pyright."""
    session.install("pyright")
    session.install("-e", ".")
    session.run("pyright", stderr=sys.stdout)


@nox.session(python=SUPPORTED_PYTHONS, reuse_venv=True, venv_backend="uv")
def unit_tests(session: nox.Session):
    """Run BDD tests with Behave."""
    session.install("-e", ".[tests]")
    _install_behave_parallel(session)

    # Build a pre-migrated template DB so each scenario can copy it
    # instead of running 25 Alembic migrations from scratch.
    template_path = _create_template_db(session)
    session.env["CLEVERAGENTS_TEMPLATE_DB"] = template_path

    behave_cmd = session.bin + "/behave-parallel"
    parallel_args = _behave_parallel_args(session.posargs)

    # If a specific feature file is passed, run only that file
    if session.posargs and session.posargs[0].endswith(".feature"):
        args = [
            behave_cmd,
            "-q",
            *parallel_args,
            *session.posargs,
        ]
    else:
        args = [
            behave_cmd,
            "-q",
            *parallel_args,
            "features/",
            *session.posargs,
        ]

    session.env["PYTHONPATH"] = str(Path("src").resolve())
    # Prevent Rich from injecting ANSI escape codes into machine-readable
    # CLI output (JSON/YAML) which causes json.loads() failures on CI.
    session.env["NO_COLOR"] = "1"
    session.run(*args)


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def docs(session: nox.Session):
    """Build documentation with MkDocs."""
    session.install("-e", ".[docs]")
    # Suppress DeprecationWarning from the third-party `rx` package which calls
    # datetime.datetime.utcfromtimestamp() (deprecated since Python 3.12).
    session.env["PYTHONWARNINGS"] = "ignore::DeprecationWarning:rx"
    session.run("mkdocs", "build")


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def serve_docs(session: nox.Session):
    """Serve docs locally for development."""
    session.install("-e", ".[docs]")
    # Suppress DeprecationWarning from the third-party `rx` package which calls
    # datetime.datetime.utcfromtimestamp() (deprecated since Python 3.12).
    session.env["PYTHONWARNINGS"] = "ignore::DeprecationWarning:rx"
    session.run("mkdocs", "serve", "--dev-addr", "0.0.0.0:8000")


@nox.session(python=SUPPORTED_PYTHONS, reuse_venv=True, venv_backend="uv")
def build(session: nox.Session):
    """Build the wheel distribution."""
    session.install("build", "pip")
    session.run("python", "-m", "build", "--wheel")


@nox.session(python=SUPPORTED_PYTHONS, reuse_venv=True, venv_backend="uv")
def integration_tests(session: nox.Session):
    """Run Robot Framework integration tests (parallel via pabot).

    Defaults to conservative parallelism (<=2 processes) to avoid
    resource pressure in CI. Override via PABOT_PROCESSES or by passing
    --processes/--processes=N in session arguments.
    """
    session.install("-e", ".[tests]")
    session.env["CLEVERAGENTS_AUTO_APPLY_MIGRATIONS"] = "true"
    session.env["NO_COLOR"] = "1"
    # Override PYTHONPATH so the editable install (src/) for this project
    # is used instead of any inherited PYTHONPATH from the outer environment.
    session.env["PYTHONPATH"] = "src"

    # Propagate venv bin to PATH so Run Process in robot files finds
    # the venv's python/robot rather than the system copies.
    venv_bin = os.path.join(session.virtualenv.location, "bin")
    session.env["PATH"] = venv_bin + os.pathsep + os.environ.get("PATH", "")

    # Ensure output directory exists (CI starts with a clean checkout)
    os.makedirs("build/reports/robot", exist_ok=True)

    # Pass the venv Python path explicitly so Run Process calls use it
    # instead of relying on PATH (which may not propagate to subprocesses).
    venv_python = os.path.join(venv_bin, "python")

    # Pre-compile bytecode so that parallel pabot workers (and the
    # Python sub-processes they spawn via ``Run Python Script``) can
    # read cached .pyc files instead of each cold-compiling every
    # module from source simultaneously — avoids a thundering-herd
    # race on CI runners with high core counts.
    session.run("python", "-m", "compileall", "-q", "src/")

    # Build a pre-migrated template DB so helper scripts that call
    # setup_workspace() can copy it instead of running 25+ Alembic
    # migrations per test — critical for parallel pabot execution.
    template_path = _create_template_db(session)
    session.env["CLEVERAGENTS_TEMPLATE_DB"] = template_path

    pabot_args, robot_args = _split_pabot_args(session.posargs)
    parallel_args = _pabot_parallel_args(pabot_args)

    # TDD expected-fail listener — inverts results for @tdd_expected_fail
    # tagged tests and validates TDD tag combinations.
    # See CONTRIBUTING.md > TDD Issue Test Tags.
    # Resolved relative to this file (not CWD) so ``nox`` invocations from
    # a non-root directory still find the listener module.
    tdd_listener = str(
        Path(__file__).parent / "robot" / "tdd_expected_fail_listener.py"
    )

    session.run(
        "pabot",
        *parallel_args,
        *pabot_args,
        "--outputdir",
        "build/reports/robot",
        "--loglevel",
        "INFO",
        "--report",
        "report.html",
        "--log",
        "log.html",
        "--xunit",
        "xunit.xml",
        "--variable",
        f"PYTHON:{venv_python}",
        "--listener",
        tdd_listener,
        "--exclude",
        "slow",
        "--exclude",
        "discovery",
        "--exclude",
        "code_blocks",
        "--exclude",
        "wip",
        "--exclude",
        "E2E",
        "--exclude",
        "tdd_fixture",
        *robot_args,
        "robot/",
    )


@nox.session(python=SUPPORTED_PYTHONS, reuse_venv=True, venv_backend="uv")
def slow_integration_tests(session: nox.Session):
    """Run Robot Framework integration tests."""
    session.install("-e", ".[tests]")
    session.env["CLEVERAGENTS_AUTO_APPLY_MIGRATIONS"] = "true"

    # TDD expected-fail listener — inverts results for @tdd_expected_fail
    # tagged tests and validates TDD tag combinations.
    # See CONTRIBUTING.md > TDD Issue Test Tags.
    # Resolved relative to this file (not CWD) so ``nox`` invocations from
    # a non-root directory still find the listener module.
    tdd_listener = str(
        Path(__file__).parent / "robot" / "tdd_expected_fail_listener.py"
    )

    session.run(
        "robot",
        "--outputdir",
        "build/reports/robot",
        "--loglevel",
        "INFO",
        "--report",
        "report.html",
        "--log",
        "log.html",
        "--xunit",
        "xunit.xml",
        "--listener",
        tdd_listener,
        "--exclude",
        "tdd_fixture",
        "robot/",
        *session.posargs,
    )


@nox.session(python=SUPPORTED_PYTHONS, reuse_venv=True, venv_backend="uv")
def e2e_tests(session: nox.Session):
    """Run end-to-end Robot Framework tests with real LLM API keys.

    E2E tests use zero mocking — they exercise the real CleverAgents CLI
    against real LLM API keys (Anthropic/OpenAI).  Tests are tagged with
    ``E2E`` and live in the ``robot/e2e/`` directory.

    This session is NOT included in the default ``nox`` run because it
    requires real API keys.  Run explicitly via ``nox -s e2e_tests``.

    Tests that require LLM API keys will skip gracefully when the keys
    are not present in the environment.
    """
    session.install("-e", ".[tests]")
    session.env["CLEVERAGENTS_AUTO_APPLY_MIGRATIONS"] = "true"
    session.env["NO_COLOR"] = "1"
    session.env["PYTHONPATH"] = "src"

    # Propagate venv bin to PATH so Run Process in robot files finds
    # the venv's python/robot rather than the system copies.
    venv_bin = os.path.join(session.virtualenv.location, "bin")
    session.env["PATH"] = venv_bin + os.pathsep + os.environ.get("PATH", "")
    venv_python = os.path.join(venv_bin, "python")

    # Ensure output directory exists
    os.makedirs("build/reports/robot-e2e", exist_ok=True)

    # Pre-compile bytecode to avoid cold-compilation overhead.
    session.run("python", "-m", "compileall", "-q", "src/")

    # Propagate LLM API keys from the environment into the session
    # so that real E2E tests can authenticate with providers.
    for key in (
        "ANTHROPIC_API_KEY",
        "OPENAI_API_KEY",
        "GOOGLE_API_KEY",
        "GEMINI_API_KEY",
    ):
        value = os.environ.get(key)
        if value:
            session.env[key] = value

    session.run(
        "robot",
        "--outputdir",
        "build/reports/robot-e2e",
        "--loglevel",
        "INFO",
        "--report",
        "report.html",
        "--log",
        "log.html",
        "--xunit",
        "xunit.xml",
        "--variable",
        f"PYTHON:{venv_python}",
        "--include",
        "E2E",
        "--listener",
        str(Path(__file__).parent / "robot" / "tdd_expected_fail_listener.py"),
        *session.posargs,
        "robot/e2e/",
    )


COVERAGE_THRESHOLD = 97


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def coverage_report(session: nox.Session):
    """Generate coverage report from Behave tests.

    Runs all behave features in a single process under slipcover.
    The in-process behave-parallel runner avoids subprocess overhead,
    so a single slipcover invocation collects coverage for the entire
    suite -- no per-worker files or merge step required.

    Coverage threshold is enforced at >=97%.

    On success, emits: COVERAGE OK: <pct>% (threshold: 97%)
    On failure, emits: COVERAGE FAILED: <pct>% < 97% threshold
    Both are single-line, CI-parseable summary strings.
    """
    session.install("-e", ".[tests]")
    _install_behave_parallel(session)

    os.makedirs("build", exist_ok=True)

    # Build a pre-migrated template DB so each scenario can copy it
    # instead of running 25 Alembic migrations from scratch.
    template_path = _create_template_db(session)
    session.env["CLEVERAGENTS_TEMPLATE_DB"] = template_path

    session.env["PYTHONPATH"] = str(Path("src").resolve())
    session.env["NO_COLOR"] = "1"
    source_paths = "src"
    omit_patterns = ",".join(
        [
            "*/tests/*",
            "*/test_*",
            "features/*",
            "*/features/*",
            "*/__pycache__/*",
            "*/site-packages/*",
            "*/dependency_injector/*",
            "*/venv/*",
            "*/.venv/*",
            "*/.nox/*",
            "src/cleveragents/discovery/*",
        ]
    )

    # Force sequential mode so slipcover wraps the entire process.
    session.env["BEHAVE_PARALLEL_COVERAGE"] = "1"

    # Clean up any existing slipcover data
    for path in Path("build").glob(".slipcover.*.json"):
        path.unlink()

    # Build behave-parallel args (sequential for coverage).
    behave_cmd = session.bin + "/behave-parallel"
    if session.posargs and session.posargs[0].endswith(".feature"):
        behave_args = [
            behave_cmd,
            "-q",
            "--tags=-discovery",
            "--no-capture",
            *session.posargs,
        ]
    else:
        behave_args = [
            behave_cmd,
            "-q",
            "--tags=-discovery",
            "--no-capture",
            "features/",
            *session.posargs,
        ]

    # Wrap the entire behave-parallel run under slipcover.
    # A single process produces a single JSON output file directly.
    # Allow exit code 1 (test failures) — coverage data is still produced.
    session.run(
        "python",
        "-m",
        "slipcover",
        "--json",
        "--out",
        "build/coverage.json",
        "--source",
        source_paths,
        "--omit",
        omit_patterns,
        "--",
        *behave_args,
        success_codes=[0, 1],
    )

    # Generate XML report for CI
    session.run(
        "python",
        "-m",
        "slipcover",
        "--merge",
        "build/coverage.json",
        "--xml",
        "--out",
        "build/coverage.xml",
    )

    # Generate terminal report and check threshold.
    # slipcover exits with code 2 if below threshold; nox intercepts the exit.
    report_path = "build/coverage-report.txt"
    session.run(
        "python",
        "-m",
        "slipcover",
        "--merge",
        "build/coverage.json",
        "--out",
        report_path,
        "--source",
        source_paths,
        "--omit",
        omit_patterns,
        f"--fail-under={COVERAGE_THRESHOLD}",
        success_codes=[0, 2],
        silent=False,
    )

    try:
        with open(report_path) as report_file:
            print(report_file.read(), end="")
    except FileNotFoundError:
        session.log(f"Slipcover report not found at {report_path}")

    # Parse the total percentage from the slipcover JSON report for the summary
    total_pct: float = 0.0
    try:
        with open("build/coverage.json") as f:
            payload = json.load(f)

        summary = payload.get("summary") or payload.get("totals") or {}
        percent_value = None
        for key in (
            "percent_covered",
            "percent_covered_display",
            "line_coverage_percent",
        ):
            percent_value = summary.get(key)
            if percent_value is not None:
                break

        if percent_value is None:
            covered = (
                summary.get("lines_covered")
                or summary.get("covered_lines")
                or summary.get("covered")
            )
            total = (
                summary.get("lines")
                or summary.get("total_lines")
                or summary.get("num_lines")
                or summary.get("total")
            )
            if covered is not None and total:
                percent_value = (float(covered) / float(total)) * 100.0

        if isinstance(percent_value, str):
            percent_value = percent_value.strip().rstrip("%")

        if percent_value is not None:
            total_pct = float(percent_value)
    except (FileNotFoundError, KeyError, TypeError, ValueError, json.JSONDecodeError):
        total_pct = 0.0

    rounded_pct = round(total_pct, 1)

    if rounded_pct >= COVERAGE_THRESHOLD:
        session.log(f"COVERAGE OK: {rounded_pct}% (threshold: {COVERAGE_THRESHOLD}%)")
    else:
        session.error(
            f"COVERAGE FAILED: {rounded_pct}% < {COVERAGE_THRESHOLD}% threshold"
        )


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def pre_commit(session: nox.Session):
    """Run all pre-commit hooks on all files."""
    session.install("-e", ".[dev]")
    session.run("pre-commit", "run", "--all-files", *session.posargs)


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def security_scan(session: nox.Session):
    """Run security checks matching Forgejo CI: bandit + semgrep + vulture.

    Steps:
      1. Bandit medium-severity report (non-blocking JSON export to build/)
      2. Bandit high-severity gate (blocks on findings)
      3. Semgrep custom rules from .semgrep.yml (blocks on ERROR-level findings)
      4. Vulture dead-code detection (blocks on confidence >=80%)

    Severity gates:
      - Bandit HIGH: hard fail (blocks merge)
      - Bandit MEDIUM: report-only (review, do not block)
      - Semgrep ERROR: hard fail (eval/exec/os.system detection)
      - Semgrep WARNING: report-only (pickle.loads)
      - Vulture >=80% confidence: hard fail
    """
    session.install("-e", ".[dev]")
    # Python 3.13 no longer bundles pkg_resources. Semgrep transitively imports
    # it via opentelemetry instrumentation, so pin setuptools to a version that
    # still provides pkg_resources.
    session.install("setuptools<81")

    # Ensure output directory exists (CI starts with a clean checkout)
    os.makedirs("build", exist_ok=True)

    # Step 1: Bandit - medium-severity report (non-blocking, mirrors CI JSON export)
    session.run(
        "bandit",
        "-c",
        "pyproject.toml",
        "-r",
        "src/cleveragents",
        "--severity-level",
        "medium",
        "--format",
        "json",
        "--output",
        "build/bandit-report.json",
        success_codes=[0, 1],
    )

    # Step 2: Bandit - high-severity gate (blocks on findings)
    session.run(
        "bandit",
        "-c",
        "pyproject.toml",
        "-r",
        "src/cleveragents",
        "--severity-level",
        "high",
    )

    # Step 3: Semgrep custom rules (blocks on ERROR-severity matches)
    semgrep_config = Path(".semgrep.yml")
    if semgrep_config.exists():
        session.run(
            "semgrep",
            "--config=.semgrep.yml",
            "--error",
            "--quiet",
            "src/",
            success_codes=[0, 1],
        )
    else:
        session.warn("No .semgrep.yml found, skipping semgrep scan")

    # Step 4: Vulture dead-code detection
    session.run(
        "vulture",
        "src/cleveragents",
        "vulture_whitelist.py",
        "--min-confidence",
        "80",
        "--exclude",
        "src/cleveragents/discovery",
    )


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def dead_code(session: nox.Session):
    """Run vulture to detect dead code."""
    session.install("-e", ".[dev]")
    session.run(
        "vulture",
        "src/cleveragents",
        "vulture_whitelist.py",
        "--min-confidence",
        "80",
        "--exclude",
        "src/cleveragents/discovery",
    )


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def complexity(session: nox.Session):
    """Check code complexity using radon."""
    session.install("-e", ".[dev]")
    session.run(
        "radon",
        "cc",
        "src/cleveragents",
        "--min",
        "C",
        "--show-complexity",
        "--total-average",
    )


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def adr_compliance(session: nox.Session):
    """Check code compliance with Architecture Decision Records."""
    session.install("-e", ".[dev]")
    session.run("python", "scripts/check-adr-compliance.py")


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def benchmark(session: nox.Session):
    """Run Airspeed Velocity benchmarks and publish results."""
    session.install("-e", ".[tests]")
    config_path = "asv.conf.json"
    session.run(
        "asv",
        "machine",
        "--machine=forgejo-runner",
        "--os=Linux_6.x",
        "--arch=x86_64",
        "--num_cpu=32",
        "--ram=32GB",
        "--cpu=AMD",
        f"--config={config_path}",
    )
    session.run(
        "asv",
        "run",
        "--machine=forgejo-runner",
        "--append-samples",
        "--launch-method=spawn",
        "--show-stderr",
        "--verbose",
        f"--config={config_path}",
        success_codes=[0, 2],
    )
    session.run("asv", "publish", f"--config={config_path}")


@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def benchmark_regression(session: nox.Session):
    """Run Airspeed Velocity benchmarks regression test."""
    session.install("-e", ".[tests]")
    config_path = "asv.conf.json"
    asv_base_sha = os.environ.get("ASV_BASE_SHA", "master")
    session.run(
        "asv",
        "machine",
        "--machine=forgejo-runner",
        "--os=Linux_6.x",
        "--arch=x86_64",
        "--num_cpu=32",
        "--ram=32GB",
        "--cpu=AMD",
        f"--config={config_path}",
    )
    session.run(
        "asv",
        "continuous",
        "--machine=forgejo-runner",
        "--append-samples",
        "--show-stderr",
        "--verbose",
        "--factor=1.50",
        f"--config={config_path}",
        asv_base_sha,
        "HEAD",
        success_codes=[0, 2],
    )
    session.run("asv", "publish", f"--config={config_path}")


# Sessions to run by default when running `nox` without arguments
nox.options.sessions = [
    "lint",  # ~5-10 seconds
    "format",  # ~5-10 seconds (auto-fixes issues)
    "typecheck",  # ~10-20 seconds
    "security_scan",  # bandit + vulture (mirrors Forgejo CI security job)
    "dead_code",  # vulture dead-code detection
    "unit_tests",  # ~30-60 seconds (without ML deps)
    "integration_tests",
    "docs",  # ~10-30 seconds
    "build",  # ~5-10 seconds
    "benchmark",  # ASV benchmarks for performance tracking
    "coverage_report",  # ~30-60 seconds (without ML deps)
]