cleveragents-core/robot/helper_wf16_devcontainer.py

"""Robot Framework helper for Workflow Example 16: Devcontainer-Driven Development.

Exercises devcontainer auto-discovery, lazy container activation,
execution environment routing, and host apply via bind mount using
mocked LLM providers and mocked container operations.

Each subcommand prints a sentinel on success and exits 0/1.

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

from __future__ import annotations

import json
import os
import re
import shutil
import subprocess
import sys
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path
from typing import NoReturn

_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)
_ROBOT = str(Path(__file__).resolve().parent)
if _ROBOT not in sys.path:
    sys.path.insert(0, _ROBOT)

from helper_e2e_common import (  # noqa: E402
    cleanup_workspace,
    init_bare_git_repo,
    run_cli,
    setup_workspace,
    write_yaml,
)
from helpers_common import reset_global_state  # noqa: E402

from cleveragents.domain.models.core.container_lifecycle import (  # noqa: E402
    ContainerLifecycleState,
)
from cleveragents.resource.handlers.devcontainer import (  # noqa: E402
    activate_container,
    clear_lifecycle_registry,
    get_lifecycle_tracker,
    stop_container,
)
from cleveragents.resource.handlers.discovery import (  # noqa: E402
    discover_devcontainers,
)

_DC_JSON: str = json.dumps(
    {"name": "wf16-webapp", "image": "node:20", "workspaceFolder": "/workspaces/webapp"}
)
_ACTION_YAML: str = """\
name: local/add-dark-mode
description: Add dark mode toggle to the webapp
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: Dark mode CSS and toggle component implemented
"""
_ENV: dict[str, str] = {"COLUMNS": "500"}


# -- Mock runner (no real Docker) ------------------------------------------


@dataclass
class _MockResult:
    returncode: int = 0
    stdout: str = ""
    stderr: str = ""
    args: list[str] = field(default_factory=list)


class _MockRunner:
    def __init__(self) -> None:
        self.calls: list[tuple[list[str], dict[str, object]]] = []
        self.up_result: _MockResult = _MockResult()
        self.stop_result: _MockResult = _MockResult()

    def __call__(self, args: list[str], **kwargs: object) -> _MockResult:
        self.calls.append((list(args), dict(kwargs)))
        if len(args) >= 2 and args[0] == "devcontainer" and args[1] == "up":
            return self.up_result
        if len(args) >= 2 and args[0] == "docker" and args[1] == "stop":
            return self.stop_result
        return _MockResult()

    def set_up_result(
        self,
        *,
        cid: str = "aabbccddee5566778899",
        ws: str = "/workspaces/webapp",
    ) -> None:
        self.up_result = _MockResult(
            stdout=json.dumps(
                {"outcome": "success", "containerId": cid, "remoteWorkspaceFolder": ws}
            ),
        )

    @property
    def up_calls(self) -> list[tuple[list[str], dict[str, object]]]:
        return [
            (a, k)
            for a, k in self.calls
            if len(a) >= 2 and a[0] == "devcontainer" and a[1] == "up"
        ]


# -- Helpers ---------------------------------------------------------------


def _fail(msg: str) -> NoReturn:
    print(f"FAIL: {msg}", file=sys.stderr)
    raise SystemExit(1)


def _plan_id(output: str) -> str | None:
    m = re.search(r"\b([0-9A-Z]{26})\b", output)
    return m.group(1) if m else None


def _make_repo() -> str:
    """Create git repo with .devcontainer/devcontainer.json."""
    repo = init_bare_git_repo()
    dc = Path(repo) / ".devcontainer"
    dc.mkdir()
    (dc / "devcontainer.json").write_text(_DC_JSON, encoding="utf-8")
    subprocess.run(["git", "add", "."], cwd=repo, capture_output=True, check=True)
    subprocess.run(
        ["git", "commit", "-m", "Add devcontainer"],
        cwd=repo,
        capture_output=True,
        check=True,
    )
    return repo


def _cli(ws: str, *args: str) -> subprocess.CompletedProcess[str]:
    """Run CLI command, fail on non-zero exit."""
    r = run_cli(*args, workspace=ws, env_extra=_ENV)
    if r.returncode != 0:
        _fail(f"{' '.join(args[:2])}: {r.stderr}")
    return r


def _setup_full(ws: str, repo: str, yaml: str) -> str:
    """Register resource, project, action, plan use. Return plan_id."""
    _cli(
        ws,
        "resource",
        "add",
        "git-checkout",
        "local/webapp",
        "--path",
        repo,
        "--branch",
        "main",
    )
    _cli(ws, "project", "create", "local/webapp-project", "--resource", "local/webapp")
    _cli(ws, "action", "create", "--config", yaml)
    r = _cli(
        ws,
        "plan",
        "use",
        "local/add-dark-mode",
        "local/webapp-project",
        "--automation-profile",
        "supervised",
        "--format",
        "plain",
    )
    pid = _plan_id(r.stdout)
    if not pid:
        _fail(f"no plan_id:\n{r.stdout}")
    return pid


# -- Test commands ---------------------------------------------------------


def wf16_resource_auto_detection() -> None:
    """Step 1: resource add git-checkout, verify devcontainer discovery."""
    ws = setup_workspace(prefix="wf16_detect_")
    repo = _make_repo()
    try:
        _cli(
            ws,
            "resource",
            "add",
            "git-checkout",
            "local/webapp",
            "--path",
            repo,
            "--branch",
            "main",
            "--format",
            "plain",
        )
        r = _cli(ws, "resource", "show", "local/webapp", "--format", "plain")
        if "git-checkout" not in r.stdout:
            _fail(f"not git-checkout:\n{r.stdout}")
        results = discover_devcontainers(repo, "git-checkout")
        if len(results) != 1:
            _fail(f"expected 1 result, got {len(results)}")
        if not str(results[0].config_path).endswith("devcontainer.json"):
            _fail(f"bad config_path: {results[0].config_path}")
        if results[0].config_data.get("name") != "wf16-webapp":
            _fail(f"bad config name: {results[0].config_data.get('name')}")
        print("wf16-detect-ok")
    finally:
        shutil.rmtree(repo, ignore_errors=True)
        cleanup_workspace(ws)


def wf16_project_and_link() -> None:
    """Step 2: project create with --resource linking."""
    ws = setup_workspace(prefix="wf16_proj_")
    repo = _make_repo()
    try:
        _cli(
            ws,
            "resource",
            "add",
            "git-checkout",
            "local/webapp",
            "--path",
            repo,
            "--branch",
            "main",
        )
        _cli(
            ws,
            "project",
            "create",
            "local/webapp-project",
            "--description",
            "WF16 webapp",
            "--resource",
            "local/webapp",
            "--format",
            "plain",
        )
        r = _cli(ws, "project", "show", "local/webapp-project", "--format", "plain")
        if "local/webapp-project" not in r.stdout:
            _fail(f"project missing:\n{r.stdout}")
        print("wf16-project-ok")
    finally:
        shutil.rmtree(repo, ignore_errors=True)
        cleanup_workspace(ws)


def wf16_plan_with_devcontainer() -> None:
    """Step 3: plan use with --automation-profile supervised."""
    ws = setup_workspace(prefix="wf16_plan_")
    repo = _make_repo()
    yaml = write_yaml(_ACTION_YAML)
    try:
        pid = _setup_full(ws, repo, yaml)
        r = _cli(ws, "plan", "status", pid, "--format", "plain")
        if pid not in r.stdout:
            _fail(f"plan_id missing:\n{r.stdout}")
        print("wf16-plan-ok")
    finally:
        os.unlink(yaml)
        shutil.rmtree(repo, ignore_errors=True)
        cleanup_workspace(ws)


def wf16_execute_in_container() -> None:
    """Step 4: plan execute + domain-level lazy activation verification."""
    ws = setup_workspace(prefix="wf16_exec_")
    repo = _make_repo()
    yaml = write_yaml(_ACTION_YAML)
    try:
        pid = _setup_full(ws, repo, yaml)
        # CLI plan execute — controlled rejection (no LLM)
        r = run_cli("plan", "execute", pid, workspace=ws, env_extra=_ENV)
        out = r.stdout + r.stderr
        if "INTERNAL" in out or "Traceback" in out:
            _fail(f"plan execute crashed:\n{out}")
        # Domain-level lazy activation with mock runner
        clear_lifecycle_registry()
        runner = _MockRunner()
        runner.set_up_result(cid="aa16c0a1a1e200112233", ws="/workspaces/webapp")
        rid = "01WF16TEST0000000000010"
        tracker = activate_container(rid, repo, run_command=runner, session_id=pid)
        if tracker.current_state != ContainerLifecycleState.RUNNING:
            _fail(f"expected RUNNING: {tracker.current_state}")
        if tracker.container_id != "aa16c0a1a1e200112233":
            _fail(f"container_id: {tracker.container_id}")
        if tracker.workspace_path != "/workspaces/webapp":
            _fail(f"workspace_path: {tracker.workspace_path}")
        if tracker.host_workspace_path != os.path.realpath(repo):
            _fail(f"host_workspace_path: {tracker.host_workspace_path}")
        if len(runner.up_calls) != 1:
            _fail(f"up_calls: {len(runner.up_calls)}")
        if "--workspace-folder" not in runner.up_calls[0][0]:
            _fail(f"missing --workspace-folder: {runner.up_calls[0][0]}")
        if get_lifecycle_tracker(rid).session_id != pid:
            _fail(f"session_id: {get_lifecycle_tracker(rid).session_id}")
        print("wf16-execute-ok")
    finally:
        clear_lifecycle_registry()
        os.unlink(yaml)
        shutil.rmtree(repo, ignore_errors=True)
        cleanup_workspace(ws)


def wf16_apply_to_host() -> None:
    """Step 5: plan apply + domain-level bind-mount path verification."""
    ws = setup_workspace(prefix="wf16_apply_")
    repo = _make_repo()
    yaml = write_yaml(_ACTION_YAML)
    try:
        pid = _setup_full(ws, repo, yaml)
        # CLI lifecycle-apply — controlled rejection
        r = run_cli("plan", "lifecycle-apply", pid, workspace=ws, env_extra=_ENV)
        out = r.stdout + r.stderr
        if "INTERNAL" in out or "Traceback" in out:
            _fail(f"lifecycle-apply crashed:\n{out}")
        # Domain-level bind-mount tracking
        clear_lifecycle_registry()
        runner = _MockRunner()
        runner.set_up_result(cid="bb16a00112233445566ff", ws="/workspaces/webapp")
        rid = "01WF16TEST0000000000020"
        activate_container(rid, repo, run_command=runner, session_id=pid)
        t = get_lifecycle_tracker(rid)
        if t.host_workspace_path != os.path.realpath(repo):
            _fail(f"host_workspace_path: {t.host_workspace_path}")
        if t.workspace_path != "/workspaces/webapp":
            _fail(f"workspace_path: {t.workspace_path}")
        stop_container(rid, run_command=runner)
        final = get_lifecycle_tracker(rid)
        if final.current_state != ContainerLifecycleState.STOPPED:
            _fail(f"expected STOPPED: {final.current_state}")
        print("wf16-apply-ok")
    finally:
        clear_lifecycle_registry()
        os.unlink(yaml)
        shutil.rmtree(repo, ignore_errors=True)
        cleanup_workspace(ws)


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

_COMMANDS: dict[str, Callable[[], None]] = {
    "wf16-resource-auto-detection": wf16_resource_auto_detection,
    "wf16-project-and-link": wf16_project_and_link,
    "wf16-plan-with-devcontainer": wf16_plan_with_devcontainer,
    "wf16-execute-in-container": wf16_execute_in_container,
    "wf16-apply-to-host": wf16_apply_to_host,
}


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


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