cleveragents-core/robot/helper_m5_e2e_verification.py

"""Robot Framework helper for M5 E2E verification tests.

Exercises the complete M5 success criteria sequence:
  1. Large project creation with 10,000+ simulated files
  2. Resource registration and linking to a project
  3. Indexing verification (resource linked, project shows correctly)
  4. Context tier management (hot/warm/cold) via ContextConfig
  5. ACMS v1 context policy set/show with persistence round-trip
  6. Phase view inheritance (default -> strategize -> execute -> apply)
  7. Scoped context output per ACMS phase
  8. Policy clear and inheritance fallback

Each subcommand prints a sentinel string on success and exits 0.
On failure it prints a diagnostic to stderr and exits 1.

Usage:
    python robot/helper_m5_e2e_verification.py <command>
"""

from __future__ import annotations

import sys
import tempfile
import time
from collections.abc import Callable
from pathlib import Path

# Ensure src is importable when run from workspace root
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from helper_m5_e2e_context import (  # noqa: E402
    acms_phase_inheritance,
    acms_scoped_context,
    context_policy_clear,
    context_policy_set_show,
    context_tier_config,
    context_view_validation,
    llm_execute_acms_context,
)
from helper_m5_e2e_support import (  # noqa: E402
    _create_large_python_repo,
    _create_project,
    _fail,
    _setup_db,
)
from helpers_common import reset_global_state  # noqa: E402

from cleveragents.application.services.repo_indexing_service import (  # noqa: E402
    RepoIndexingService,
)
from cleveragents.domain.models.core.repo_index import IndexStatus  # noqa: E402

# -------------------------------------------------------------------
# Subcommand: project-create-large
# -------------------------------------------------------------------


def project_create_large() -> None:
    """Create a project and simulate a 10,000+ file repository.

    Verifies project creation, resource registration, linking,
    and that the project show reflects the linked resource.
    """
    reset_global_state()
    proj_repo, link_repo, svc, _sf = _setup_db()

    proj = _create_project(
        proj_repo,
        "local/large-project",
        description="Large project with 10,000+ file repo",
    )
    if proj.namespaced_name != "local/large-project":
        _fail(f"project name mismatch: {proj.namespaced_name}")

    with tempfile.TemporaryDirectory() as tmp:
        repo_path = Path(tmp) / "large-repo"
        repo_path.mkdir()

        file_count = _create_large_python_repo(repo_path)
        if file_count < 10_000:
            _fail(f"expected >= 10,000 files, created {file_count}")

        res = svc.register_resource(
            type_name="git-checkout",
            name="local/large-repo",
            location=str(repo_path),
            description="Large repository (10K+ files)",
            properties={"path": str(repo_path), "branch": "main"},
        )

        if not res.resource_id:
            _fail("resource_id is empty")

        link_repo.create_link(
            project_name="local/large-project",
            resource_id=res.resource_id,
        )

        links = link_repo.list_links("local/large-project")
        if len(links) != 1:
            _fail(f"expected 1 link, got {len(links)}")

        fetched = proj_repo.get("local/large-project")
        if fetched.description != "Large project with 10,000+ file repo":
            _fail(f"description mismatch: {fetched.description}")

    print("m5-project-create-large-ok")


# -------------------------------------------------------------------
# Subcommand: resource-register-link
# -------------------------------------------------------------------


def resource_register_link() -> None:
    """Register a git-checkout resource and link it to a project."""
    reset_global_state()
    proj_repo, link_repo, svc, _sf = _setup_db()

    proj = _create_project(proj_repo, "local/link-test")

    res1 = svc.register_resource(
        type_name="git-checkout",
        name="local/repo-alpha",
        location="/tmp/alpha",
        description="Alpha repo",
        properties={"path": "/tmp/alpha", "branch": "main"},
    )
    res2 = svc.register_resource(
        type_name="git-checkout",
        name="local/repo-beta",
        location="/tmp/beta",
        description="Beta repo",
        properties={"path": "/tmp/beta", "branch": "develop"},
    )

    link_repo.create_link(
        project_name=proj.namespaced_name,
        resource_id=res1.resource_id,
    )
    link_repo.create_link(
        project_name=proj.namespaced_name,
        resource_id=res2.resource_id,
    )

    links = link_repo.list_links(proj.namespaced_name)
    if len(links) != 2:
        _fail(f"expected 2 links, got {len(links)}")

    shown = svc.show_resource("local/repo-alpha")
    if shown.name != "local/repo-alpha":
        _fail(f"resource name mismatch: {shown.name}")
    if shown.resource_type_name != "git-checkout":
        _fail(f"type mismatch: {shown.resource_type_name}")

    print("m5-resource-register-link-ok")


# -------------------------------------------------------------------
# Subcommand: indexing-complete
# -------------------------------------------------------------------


def indexing_complete() -> None:
    """Verify indexing completes: resource is linked, project shows OK."""
    reset_global_state()
    proj_repo, link_repo, svc, sf = _setup_db()

    proj = _create_project(proj_repo, "local/idx-project")
    with tempfile.TemporaryDirectory(prefix="m5_idx_repo_") as tmp:
        repo_root = Path(tmp) / "idx-repo"
        repo_root.mkdir(parents=True, exist_ok=True)

        file_count = _create_large_python_repo(repo_root)
        if file_count < 10_000:
            _fail(f"expected >= 10,000 files, created {file_count}")

        res = svc.register_resource(
            type_name="git-checkout",
            name="local/idx-repo",
            location=str(repo_root),
            description="Index test",
            properties={"path": str(repo_root), "branch": "main"},
        )
        link_repo.create_link(
            project_name=proj.namespaced_name,
            resource_id=res.resource_id,
        )

        links = link_repo.list_links(proj.namespaced_name)
        if len(links) != 1:
            _fail(f"expected 1 link, got {len(links)}")

        index_service = RepoIndexingService(session_factory=sf)
        timeout_seconds = 120.0
        started = time.monotonic()
        idx = index_service.index_resource(
            res.resource_id,
            repo_root,
            timeout_seconds=timeout_seconds,
        )
        elapsed = time.monotonic() - started

        if elapsed > timeout_seconds:
            _fail(
                "indexing exceeded timeout bound: "
                f"{elapsed:.2f}s > {timeout_seconds:.2f}s"
            )
        if idx.metadata.file_count < 10_000:
            _fail(f"expected >= 10,000 indexed files, got {idx.metadata.file_count}")
        if idx.metadata.status != IndexStatus.READY:
            _fail(f"expected READY status, got {idx.metadata.status}")

        fetched_res = svc.show_resource("local/idx-repo")
        if fetched_res.resource_id != res.resource_id:
            _fail(f"resource_id mismatch after fetch: {fetched_res.resource_id}")
        if fetched_res.location != str(repo_root):
            _fail(f"location mismatch: {fetched_res.location}")

        status = index_service.get_index_status(res.resource_id)
        if status is None:
            _fail("index status missing after indexing")
        if status.file_count != idx.metadata.file_count:
            _fail(
                "index status file_count mismatch: "
                f"{status.file_count} != {idx.metadata.file_count}"
            )

    print("m5-indexing-complete-ok")


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

_COMMANDS: dict[str, Callable[[], None]] = {
    "project-create-large": project_create_large,
    "resource-register-link": resource_register_link,
    "indexing-complete": indexing_complete,
    "context-tier-config": context_tier_config,
    "context-policy-set-show": context_policy_set_show,
    "acms-phase-inheritance": acms_phase_inheritance,
    "acms-scoped-context": acms_scoped_context,
    "context-policy-clear": context_policy_clear,
    "context-view-validation": context_view_validation,
    "llm-execute-acms-context": llm_execute_acms_context,
}


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print(
            f"Usage: helper_m5_e2e_verification.py <{'|'.join(_COMMANDS)}>",
        )
        return 1
    command = sys.argv[1]
    handler = _COMMANDS.get(command)
    if handler is None:
        print(f"Unknown command: {command}")
        return 1
    handler()
    return 0


if __name__ == "__main__":
    sys.exit(main())