cleveragents-core/robot/helper_tdd_skill_add_regression.py

"""Helper script for tdd_skill_add_regression.robot smoke tests.

Each subcommand exercises the real CLI code path (no mocks) to reproduce
bug #980: skills registered via ``agents skill add --config <file>`` in
one CLI invocation are not visible when ``agents skill list`` is run in
a separate CLI invocation.

The existing in-process persistence tests pass because they share an
in-memory SQLAlchemy session factory between two SkillService instances.
This helper uses ``reset_container()`` between invocations to simulate
the cross-process boundary — each CLI invocation starts with a clean
DI container and must load skill data from the on-disk database.

The helper reports the **real** outcome: it exits 0 and prints the
sentinel when the skill is found cross-process (bug is fixed), and
exits 1 when the skill is not found (bug still present).  The
``tdd_expected_fail_listener`` on the Robot side handles pass/fail
inversion while the bug remains open.
"""

from __future__ import annotations

import os
import shutil
import sys
import tempfile
from collections.abc import Callable
from pathlib import Path
from typing import NoReturn

# Ensure local source tree is importable.
_ROOT = Path(__file__).resolve().parents[1]
_SRC = str(_ROOT / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from typer.testing import CliRunner  # noqa: E402

from cleveragents.application.container import reset_container  # noqa: E402
from cleveragents.cli.main import app as root_app  # noqa: E402

runner = CliRunner()

_SKILL_YAML = """\
name: local/tdd-cross-process
description: "TDD cross-process persistence test skill"

tools:
  - name: builtin/read_file
"""

_SKILL_NAME = "local/tdd-cross-process"


def _fail(message: str) -> NoReturn:
    """Print an error message to stderr and exit with code 1."""
    print(message, file=sys.stderr)
    sys.exit(1)


def _setup_env() -> tuple[str, str, dict[str, str | None]]:
    """Create a temp directory with a skill config and initialised database.

    Returns:
        Tuple of (tmpdir, config_path, saved_env_vars).
    """
    tmpdir = tempfile.mkdtemp(prefix="tdd_skill_robot_xproc_")

    # Write skill YAML config
    config_path = os.path.join(tmpdir, "test-skill.yaml")
    with open(config_path, "w") as fh:
        fh.write(_SKILL_YAML)

    # Database path
    db_dir = os.path.join(tmpdir, ".cleveragents")
    os.makedirs(db_dir, exist_ok=True)
    db_url = f"sqlite:///{db_dir}/db.sqlite"

    # Save and override env vars
    saved_env: dict[str, str | None] = {}
    for key in (
        "CLEVERAGENTS_DATABASE_URL",
        "CLEVERAGENTS_TEST_DATABASE_URL",
        "CLEVERAGENTS_HOME",
    ):
        saved_env[key] = os.environ.pop(key, None)

    os.environ["CLEVERAGENTS_HOME"] = tmpdir
    os.environ["CLEVERAGENTS_DATABASE_URL"] = db_url

    # Initialise the project (creates database + migrations)
    reset_container()
    init_result = runner.invoke(
        root_app, ["init", "--yes", "--force", "--path", tmpdir]
    )
    reset_container()

    if init_result.exit_code != 0:
        _restore_env(saved_env, tmpdir)
        _fail(
            f"Project init failed with exit code {init_result.exit_code}.\n"
            f"Output: {init_result.output}"
        )

    return tmpdir, config_path, saved_env


def _restore_env(saved_env: dict[str, str | None], tmpdir: str) -> None:
    """Restore saved env vars and clean up temp directory."""
    for key, val in saved_env.items():
        if val is not None:
            os.environ[key] = val
        else:
            os.environ.pop(key, None)
    reset_container()
    shutil.rmtree(tmpdir, ignore_errors=True)


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def cross_process_list() -> None:
    """Add a skill in one CLI invocation, list in another.

    Simulates cross-process boundary by resetting the DI container
    between invocations.  Exits 0 with sentinel when the skill is
    found (bug fixed).  Exits 1 when not found (bug present).
    """
    tmpdir, config_path, saved_env = _setup_env()
    try:
        # Invocation 1: add the skill
        reset_container()
        add_result = runner.invoke(
            root_app,
            ["skill", "add", "--config", config_path, "--format", "json"],
        )
        if add_result.exit_code != 0:
            _fail(
                f"skill add failed with exit code {add_result.exit_code}.\n"
                f"Output: {add_result.output}"
            )

        # Invocation 2: list skills (cross-process boundary)
        reset_container()
        list_result = runner.invoke(
            root_app,
            ["skill", "list", "--format", "json"],
        )

        if _SKILL_NAME not in list_result.output:
            _fail(
                f"Skill '{_SKILL_NAME}' not found in cross-process list.\n"
                f"Add output: {add_result.output}\n"
                f"List exit code: {list_result.exit_code}\n"
                f"List output: {list_result.output}"
            )

        print("tdd-skill-add-cross-process-list-ok")
    finally:
        _restore_env(saved_env, tmpdir)


def cross_process_show() -> None:
    """Add a skill in one CLI invocation, show in another.

    Simulates cross-process boundary by resetting the DI container
    between invocations.  Exits 0 with sentinel when the skill show
    succeeds (bug fixed).  Exits 1 when not found (bug present).
    """
    tmpdir, config_path, saved_env = _setup_env()
    try:
        # Invocation 1: add the skill
        reset_container()
        add_result = runner.invoke(
            root_app,
            ["skill", "add", "--config", config_path, "--format", "json"],
        )
        if add_result.exit_code != 0:
            _fail(
                f"skill add failed with exit code {add_result.exit_code}.\n"
                f"Output: {add_result.output}"
            )

        # Invocation 2: show the skill (cross-process boundary)
        reset_container()
        show_result = runner.invoke(
            root_app,
            ["skill", "show", _SKILL_NAME, "--format", "json"],
        )

        if show_result.exit_code != 0 or _SKILL_NAME not in show_result.output:
            _fail(
                f"Skill '{_SKILL_NAME}' not found in cross-process show.\n"
                f"Add output: {add_result.output}\n"
                f"Show exit code: {show_result.exit_code}\n"
                f"Show output: {show_result.output}"
            )

        print("tdd-skill-add-cross-process-show-ok")
    finally:
        _restore_env(saved_env, tmpdir)


# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, Callable[[], None]] = {
    "cross-process-list": cross_process_list,
    "cross-process-show": cross_process_show,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(
            f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>",
            file=sys.stderr,
        )
        sys.exit(1)
    cmd = _COMMANDS[sys.argv[1]]
    cmd()