cleveragents-core/scripts/create-tests.py

#!/usr/bin/env python3
"""
Test Creation Script for CleverAgents

Generates .zsh test composition code by wrapping arbitrary test logic with
setup/teardown commands for 14 agent command types. Each function returns a
string of shell commands structured as: creation → wrapped content → removal.
"""

import logging
import os
import shlex
import textwrap
from typing import Any, Dict, List, Optional

# Configure logging
logging.basicConfig(level=logging.WARNING, format="%(levelname)s: %(message)s")
logger = logging.getLogger(__name__)


# ============================================================================
# Helper Functions
# ============================================================================


def _format_flag(param_name: str, value: Any) -> str:
    """
    Convert Python parameter name and value to CLI flag.

    Rules:
    - Underscores → dashes: automation_profile → --automation-profile
    - Booleans: True → --flag, False → omitted
    - Strings/paths: value → --flag "value"
    - Lists: each item → --flag item1 --flag item2

    Returns:
        Formatted flag string or empty string if should be omitted.
    """
    if not _should_include_param(value):
        return ""

    # Convert parameter name: underscores to dashes
    flag_name = "--" + param_name.replace("_", "-")

    # Handle boolean True (flag with no value)
    if isinstance(value, bool) and value:
        return flag_name

    # Handle lists (multiple instances of the flag)
    if isinstance(value, list):
        parts = []
        for item in value:
            parts.append(f"{flag_name} {shlex.quote(str(item))}")
        return " ".join(parts)

    # Handle strings, paths, numbers
    if isinstance(value, dict):
        # For dict args (name=value pairs)
        parts = []
        for k, v in value.items():
            parts.append(f"{flag_name} {k}={shlex.quote(str(v))}")
        return " ".join(parts)

    return f"{flag_name} {shlex.quote(str(value))}"


def _should_include_param(value: Any) -> bool:
    """
    Determine if optional parameter should appear in output.

    Rules:
    - Boolean False → exclude
    - None, "", [] → exclude
    - Otherwise → include

    Returns:
        True if should include, False if omit.
    """
    if isinstance(value, bool):
        return value  # False is omitted, True is included

    if value is None:
        return False

    if isinstance(value, str) and not value:
        return False

    if isinstance(value, (list, dict)) and not value:
        return False

    return True


def _validate_file_exists(path: str) -> None:
    """
    Log warning if file doesn't exist.

    Does not raise exception; allows execution to continue.
    """
    if not path:
        return

    if not os.path.exists(path):
        logger.warning(f"File does not exist: {path}")


def _build_command(*parts: str) -> str:
    """
    Join command parts with proper spacing.

    Handles quoting for paths with spaces. Returns formatted shell command string.
    """
    # Filter out empty parts
    filtered_parts = [p for p in parts if p]
    return " ".join(filtered_parts)


# ============================================================================
# The 14 Command Functions
# ============================================================================


def init(yes: bool = False, wrapped: str = "") -> str:
    """
    Initialize CleverAgents environment.

    Special case: No removal command.

    Args:
        yes: Suppress confirmation prompt.
        wrapped: Code to execute after initialization.

    Returns:
        Shell command string: agents init [--yes] → wrapped
    """
    startup = textwrap.dedent("""
    uv venv
    source .venv/bin/activate
    uv pip install /app
    """)

    cmd_parts = ["agents", "init"]
    if yes:
        cmd_parts.append("--yes")

    init_cmd = _build_command(*cmd_parts)

    if wrapped:
        return f"{startup}{init_cmd}\n{wrapped}"
    return f"{startup}{init_cmd}"


def session(actor: str = "", yes: bool = False, wrapped: str = "") -> str:
    """
    Create and remove a session.

    Args:
        actor: Optional actor name for session (e.g., local/orchestrator).
        yes: Include --yes flag in remove command.
        wrapped: Code to execute during session.

    Returns:
        Shell command string: create → wrapped → delete
    """
    # Creation command
    create_parts = ["agents", "session", "create"]
    if actor:
        create_parts.extend(["--actor", shlex.quote(actor)])
    create_cmd = _build_command(*create_parts)

    # Removal command (caller must capture SESSION_ID)
    remove_parts = ["agents", "session", "delete"]
    if yes:
        remove_parts.append("--yes")
    remove_parts.append('"${SESSION_ID}"')  # Placeholder for ID
    remove_cmd = _build_command(*remove_parts)

    return _build_command(
        create_cmd, "\n" + wrapped if wrapped else "", "\n" + remove_cmd
    )


def resource_type(
    config: str,
    name: str = "",
    update: bool = False,
    yes: bool = False,
    wrapped: str = "",
) -> str:
    """
    Create and remove a resource type.

    Args:
        config: Path to resource type config file.
        name: Resource type name.
        update: Include --update flag.
        yes: Include --yes flag in remove command.
        wrapped: Code to execute with this resource type.

    Returns:
        Shell command string: add → wrapped → remove
    """
    _validate_file_exists(config)

    if not name:
        logger.error("resource_type: name is required")
        raise ValueError("name is required for resource_type")

    # Creation command
    create_parts = ["agents", "resource", "type", "add"]
    create_parts.extend(["--config", shlex.quote(config)])
    if update:
        create_parts.append("--update")
    create_cmd = _build_command(*create_parts)

    # Removal command
    remove_parts = ["agents", "resource", "type", "remove"]
    if yes:
        remove_parts.append("--yes")
    remove_parts.append(shlex.quote(name))
    remove_cmd = _build_command(*remove_parts)

    return _build_command(
        create_cmd, "\n" + wrapped if wrapped else "", "\n" + remove_cmd
    )


def resource(
    type: str,
    name: str,
    description: str = "",
    update: bool = False,
    yes: bool = False,
    wrapped: str = "",
) -> str:
    """
    Create and remove a resource.

    Args:
        type: Resource type name.
        name: Resource name.
        description: Optional description.
        update: Include --update flag.
        yes: Include --yes flag in remove command.
        wrapped: Code to execute with this resource.

    Returns:
        Shell command string: add → wrapped → remove
    """
    if not type:
        logger.error("resource: type is required")
        raise ValueError("type is required for resource")
    if not name:
        logger.error("resource: name is required")
        raise ValueError("name is required for resource")

    # Creation command
    create_parts = ["agents", "resource", "add", shlex.quote(type), shlex.quote(name)]
    if description:
        create_parts.extend(["--description", shlex.quote(description)])
    if update:
        create_parts.append("--update")
    create_cmd = _build_command(*create_parts)

    # Removal command
    remove_parts = ["agents", "resource", "remove"]
    if yes:
        remove_parts.append("--yes")
    remove_parts.append(shlex.quote(name))
    remove_cmd = _build_command(*remove_parts)

    return _build_command(
        create_cmd, "\n" + wrapped if wrapped else "", "\n" + remove_cmd
    )


def lsp(
    config: str, name: str, update: bool = False, yes: bool = False, wrapped: str = ""
) -> str:
    """
    Create and remove an LSP server registration.

    Args:
        config: Path to LSP config file.
        name: LSP server name.
        update: Include --update flag.
        yes: Include --yes flag in remove command.
        wrapped: Code to execute with this LSP server.

    Returns:
        Shell command string: add → wrapped → remove
    """
    _validate_file_exists(config)

    if not name:
        logger.error("lsp: name is required")
        raise ValueError("name is required for lsp")

    # Creation command
    create_parts = ["agents", "lsp", "add"]
    create_parts.extend(["--config", shlex.quote(config)])
    if update:
        create_parts.append("--update")
    create_cmd = _build_command(*create_parts)

    # Removal command
    remove_parts = ["agents", "lsp", "remove"]
    if yes:
        remove_parts.append("--yes")
    remove_parts.append(shlex.quote(name))
    remove_cmd = _build_command(*remove_parts)

    return _build_command(
        create_cmd, "\n" + wrapped if wrapped else "", "\n" + remove_cmd
    )


def automation_profile(
    config: str, name: str, update: bool = False, yes: bool = False, wrapped: str = ""
) -> str:
    """
    Create and remove an automation profile.

    Args:
        config: Path to profile config file.
        name: Profile name.
        update: Include --update flag.
        yes: Include --yes flag in remove command.
        wrapped: Code to execute with this profile.

    Returns:
        Shell command string: add → wrapped → remove
    """
    _validate_file_exists(config)

    if not name:
        logger.error("automation_profile: name is required")
        raise ValueError("name is required for automation_profile")

    # Creation command
    create_parts = ["agents", "automation-profile", "add"]
    create_parts.extend(["--config", shlex.quote(config)])
    if update:
        create_parts.append("--update")
    create_cmd = _build_command(*create_parts)

    # Removal command
    remove_parts = ["agents", "automation-profile", "remove"]
    if yes:
        remove_parts.append("--yes")
    remove_parts.append(shlex.quote(name))
    remove_cmd = _build_command(*remove_parts)

    return _build_command(
        create_cmd, "\n" + wrapped if wrapped else "", "\n" + remove_cmd
    )


def tool(
    config: str, name: str, update: bool = False, yes: bool = False, wrapped: str = ""
) -> str:
    """
    Create and remove a tool.

    Args:
        config: Path to tool config file.
        name: Tool name.
        update: Include --update flag.
        yes: Include --yes flag in remove command.
        wrapped: Code to execute with this tool.

    Returns:
        Shell command string: add → wrapped → remove
    """
    _validate_file_exists(config)

    if not name:
        logger.error("tool: name is required")
        raise ValueError("name is required for tool")

    # Creation command
    create_parts = ["agents", "tool", "add"]
    create_parts.extend(["--config", shlex.quote(config)])
    if update:
        create_parts.append("--update")
    create_cmd = _build_command(*create_parts)

    # Removal command
    remove_parts = ["agents", "tool", "remove"]
    if yes:
        remove_parts.append("--yes")
    remove_parts.append(shlex.quote(name))
    remove_cmd = _build_command(*remove_parts)

    return _build_command(
        create_cmd, "\n" + wrapped if wrapped else "", "\n" + remove_cmd
    )


def skill(
    config: str, name: str, update: bool = False, yes: bool = False, wrapped: str = ""
) -> str:
    """
    Create and remove a skill.

    Args:
        config: Path to skill config file.
        name: Skill name.
        update: Include --update flag.
        yes: Include --yes flag in remove command.
        wrapped: Code to execute with this skill.

    Returns:
        Shell command string: add → wrapped → remove
    """
    _validate_file_exists(config)

    if not name:
        logger.error("skill: name is required")
        raise ValueError("name is required for skill")

    # Creation command
    create_parts = ["agents", "skill", "add"]
    create_parts.extend(["--config", shlex.quote(config)])
    if update:
        create_parts.append("--update")
    create_cmd = _build_command(*create_parts)

    # Removal command
    remove_parts = ["agents", "skill", "remove"]
    if yes:
        remove_parts.append("--yes")
    remove_parts.append(shlex.quote(name))
    remove_cmd = _build_command(*remove_parts)

    return _build_command(
        create_cmd, "\n" + wrapped if wrapped else "", "\n" + remove_cmd
    )


def actor(config: str, name: str, update: bool = False, wrapped: str = "") -> str:
    """
    Create and remove an actor.

    Args:
        config: Path to actor config file.
        name: Actor name.
        update: Include --update flag.
        yes: Include --yes flag in remove command.
        wrapped: Code to execute with this actor.

    Returns:
        Shell command string: add → wrapped → remove
    """
    _validate_file_exists(config)

    if not name:
        logger.error("actor: name is required")
        raise ValueError("name is required for actor")

    # Creation command
    create_parts = ["agents", "actor", "add"]
    create_parts.extend(["--config", shlex.quote(config)])
    if update:
        create_parts.append("--update")
    create_cmd = _build_command(*create_parts)

    # Removal command
    remove_parts = ["agents", "actor", "remove", shlex.quote(name)]
    remove_cmd = _build_command(*remove_parts)

    return _build_command(
        create_cmd, "\n" + wrapped if wrapped else "", "\n" + remove_cmd
    )


def action(config: str, wrapped: str = "") -> str:
    """
    Create and remove an action.

    Args:
        config: Path to action config file.
        wrapped: Code to execute with this action.

    Returns:
        Shell command string: create → wrapped → archive
    """
    _validate_file_exists(config)

    # Creation command
    create_parts = ["agents", "action", "create"]
    create_parts.extend(["--config", shlex.quote(config)])
    create_cmd = _build_command(*create_parts)

    # Removal command (caller must provide ACTION_NAME)
    remove_parts = ["agents", "action", "archive", '"${ACTION_NAME}"']
    remove_cmd = _build_command(*remove_parts)

    return _build_command(
        create_cmd, "\n" + wrapped if wrapped else "", "\n" + remove_cmd
    )


def project(
    name: str,
    description: str = "",
    resources: Optional[List[str]] = None,
    invariants: Optional[List[str]] = None,
    invariant_actor: str = "",
    yes: bool = False,
    wrapped: str = "",
) -> str:
    """
    Create and remove a project.

    Args:
        name: Project name.
        description: Optional description.
        resources: List of resource names to link (assume they exist).
        invariants: List of invariant texts to attach.
        invariant_actor: Optional invariant reconciliation actor name.
        yes: Include --yes flag in delete command.
        wrapped: Code to execute with this project.

    Returns:
        Shell command string: create → wrapped → delete
    """
    if not name:
        logger.error("project: name is required")
        raise ValueError("name is required for project")

    if resources is None:
        resources = []
    if invariants is None:
        invariants = []

    # Creation command
    create_parts = ["agents", "project", "create"]
    if description:
        create_parts.extend(["--description", shlex.quote(description)])
    for resource in resources:
        create_parts.extend(["--resource", shlex.quote(resource)])
    for invariant in invariants:
        create_parts.extend(["--invariant", shlex.quote(invariant)])
    if invariant_actor:
        create_parts.extend(["--invariant-actor", shlex.quote(invariant_actor)])
    create_parts.append(shlex.quote(name))
    create_cmd = _build_command(*create_parts)

    # Removal command
    remove_parts = ["agents", "project", "delete"]
    if yes:
        remove_parts.append("--yes")
    remove_parts.append(shlex.quote(name))
    remove_cmd = _build_command(*remove_parts)

    return _build_command(
        create_cmd, "\n" + wrapped if wrapped else "", "\n" + remove_cmd
    )


def plan(
    action: str,
    projects: List[str],
    automation_profile: str = "",
    strategy_actor: str = "",
    execution_actor: str = "",
    estimation_actor: str = "",
    invariant_actor: str = "",
    invariants: Optional[List[str]] = None,
    args: Optional[Dict[str, str]] = None,
    yes: bool = False,
    wrapped: str = "",
) -> str:
    """
    Create and execute a plan, then cancel it.

    Args:
        action: Action name.
        projects: List of project names (at least one required).
        automation_profile: Optional profile name.
        strategy_actor: Optional strategy actor override.
        execution_actor: Optional execution actor override.
        estimation_actor: Optional estimation actor override.
        invariant_actor: Optional invariant actor override.
        invariants: Optional list of invariant IDs.
        args: Optional action arguments (dict → --arg name=value).
        yes: Include --yes flag in cancel command.
        wrapped: Code to execute during plan.

    Returns:
        Shell command string: use → wrapped → cancel
    """
    if not action:
        logger.error("plan: action is required")
        raise ValueError("action is required for plan")
    if not projects:
        logger.error("plan: at least one project is required")
        raise ValueError("projects is required for plan")

    if invariants is None:
        invariants = []
    if args is None:
        args = {}

    # Creation command (agents plan use)
    use_parts = ["agents", "plan", "use"]
    if automation_profile:
        use_parts.extend(["--automation-profile", shlex.quote(automation_profile)])
    if strategy_actor:
        use_parts.extend(["--strategy-actor", shlex.quote(strategy_actor)])
    if execution_actor:
        use_parts.extend(["--execution-actor", shlex.quote(execution_actor)])
    if estimation_actor:
        use_parts.extend(["--estimation-actor", shlex.quote(estimation_actor)])
    if invariant_actor:
        use_parts.extend(["--invariant-actor", shlex.quote(invariant_actor)])
    for invariant in invariants:
        use_parts.extend(["--invariant", shlex.quote(invariant)])
    for key, value in args.items():
        use_parts.extend(["--arg", f"{key}={shlex.quote(value)}"])
    use_parts.append(shlex.quote(action))
    for project in projects:
        use_parts.append(shlex.quote(project))
    use_cmd = _build_command(*use_parts)

    # Removal command (caller must capture PLAN_ID)
    cancel_parts = ["agents", "plan", "cancel"]
    if yes:
        cancel_parts.append("--yes")
    cancel_parts.append('"${PLAN_ID}"')  # Placeholder for ID
    cancel_cmd = _build_command(*cancel_parts)

    return _build_command(use_cmd, "\n" + wrapped if wrapped else "", "\n" + cancel_cmd)


def validation(
    config: str,
    resource: str,
    attachment_args: Optional[Dict[str, str]] = None,
    yes: bool = False,
    wrapped: str = "",
) -> str:
    """
    Create and remove a validation attachment.

    Args:
        config: Path to validation config file.
        resource: Resource name to attach validation to.
        attachment_args: Dict of attachment-specific args (e.g., {"project": "myproj"}).
        yes: Include --yes flag in detach command.
        wrapped: Code to execute with this validation.

    Returns:
        Shell command string: add → attach → wrapped → detach
    """
    _validate_file_exists(config)

    if not resource:
        logger.error("validation: resource is required")
        raise ValueError("resource is required for validation")

    if attachment_args is None:
        attachment_args = {}

    # Add command
    add_parts = ["agents", "validation", "add"]
    add_parts.extend(["--config", shlex.quote(config)])
    add_cmd = _build_command(*add_parts)

    # Attach command (caller must extract validation name)
    attach_parts = [
        "agents",
        "validation",
        "attach",
        shlex.quote(resource),
        '"${VALIDATION_NAME}"',
    ]
    for key, value in attachment_args.items():
        attach_parts.append(f"--{key.replace('_', '-')} {shlex.quote(value)}")
    attach_cmd = _build_command(*attach_parts)

    # Detach command (caller must capture ATTACHMENT_ID)
    detach_parts = ["agents", "validation", "detach"]
    if yes:
        detach_parts.append("--yes")
    detach_parts.append('"${ATTACHMENT_ID}"')  # Placeholder for ID
    detach_cmd = _build_command(*detach_parts)

    full_creation = _build_command(add_cmd, "\n" + attach_cmd)
    return _build_command(
        full_creation, "\n" + wrapped if wrapped else "", "\n" + detach_cmd
    )


def invariant(
    text: str,
    global_scope: bool = False,
    project: str = "",
    plan: str = "",
    action: str = "",
    yes: bool = False,
    wrapped: str = "",
) -> str:
    """
    Create and remove an invariant.

    Args:
        text: Invariant text.
        global_scope: If True, use --global flag.
        project: Optional project scope.
        plan: Optional plan ID scope.
        action: Optional action scope.
        yes: Include --yes flag in remove command.
        wrapped: Code to execute with this invariant.

    Returns:
        Shell command string: add → wrapped → remove
    """
    if not text:
        logger.error("invariant: text is required")
        raise ValueError("text is required for invariant")

    # Creation command
    add_parts = ["agents", "invariant", "add"]
    if global_scope:
        add_parts.append("--global")
    if project:
        add_parts.extend(["--project", shlex.quote(project)])
    if plan:
        add_parts.extend(["--plan", shlex.quote(plan)])
    if action:
        add_parts.extend(["--action", shlex.quote(action)])
    add_parts.append(shlex.quote(text))
    add_cmd = _build_command(*add_parts)

    # Removal command (caller must capture INVARIANT_ID)
    remove_parts = ["agents", "invariant", "remove"]
    if yes:
        remove_parts.append("--yes")
    remove_parts.append('"${INVARIANT_ID}"')  # Placeholder for ID
    remove_cmd = _build_command(*remove_parts)

    return _build_command(add_cmd, "\n" + wrapped if wrapped else "", "\n" + remove_cmd)


# ============================================================================
# Main Entry Point (Placeholder for Future Integration)
# ============================================================================


def main():
    """
    Main entry point.

    Placeholder for future integration with .zsh file writing and execution.
    """
    result = init(
        yes=True,
        wrapped=automation_profile(
            config="supervised2.yaml",
            name="supervised2",
            update=False,
            yes=True,
            wrapped="agents automation-profile list"
        ) + "\n" + automation_profile(
            config="supervised2.yaml",
            name="supervised2",
            update=True,
            yes=True,
            wrapped="agents automation-profile list"
        )
    )

    print(result)

if __name__ == "__main__":
    main()