DEVX-12: feat: add opentofu helpers, CLI entry points, shared utility, and CI improvements

src/devx/__init__.py

@@ -1,3 +1,3 @@
 """devx — reusable development and CI/CD tools for oblachno-oss projects."""
 
-__version__ = "0.5.0"
+__version__ = "0.6.0"

src/devx/ci/_shared.py

@@ -0,0 +1,18 @@
+"""Shared utilities for CI modules."""
+
+from __future__ import annotations
+
+import subprocess  # nosec B404
+
+
+def get_latest_tag() -> str:
+    """Get the latest git tag, or empty string if none exists."""
+    result = subprocess.run(  # nosec B603 B607
+        ["git", "describe", "--tags", "--abbrev=0"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        return ""
+    return result.stdout.strip()

src/devx/ci/classify_changes.py

@@ -138,6 +138,7 @@ from typing import Any
 
 import click
 
+from devx.ci._shared import get_latest_tag
 from devx.i18n import _
 
 # ---------------------------------------------------------------------------
@@ -494,19 +495,6 @@ def get_changed_files(base: str, head: str) -> list[str]:
     return output.split("\n")
 
 
-def get_latest_tag() -> str:
-    """Get the latest git tag, or empty string if none exists."""
-    result = subprocess.run(  # nosec B603 B607
-        ["git", "describe", "--tags", "--abbrev=0"],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    if result.returncode != 0:
-        return ""
-    return result.stdout.strip()
-
-
 # ---------------------------------------------------------------------------
 # Backward-compatible API (used by release.py and CI workflows)
 # ---------------------------------------------------------------------------

src/devx/ci/distribute_files.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""Distribute a list of files across N parallel runners (round-robin).
+
+Generic file-based test distribution for CI matrix jobs. Discovers files
+matching a glob pattern, sorts them for deterministic ordering, then
+assigns them round-robin to *max_runners* groups. The assigned group for
+*runner_index* is written to ``$GITHUB_ENV`` for use by subsequent steps.
+
+Usage::
+
+    python3 -m devx.ci.distribute_files \\
+        --pattern "tests/integration/test_*.py" \\
+        --runner-index 1 \\
+        --max-runners 3 \\
+        --github-env --skip-if-excess
+"""
+
+from __future__ import annotations
+
+import glob
+import os
+
+import click
+
+from devx.i18n import _
+
+DEFAULT_MAX_RUNNERS = 3
+
+
+def discover_files(pattern: str) -> list[str]:
+    """Return sorted list of file paths matching *pattern*."""
+    return sorted(glob.glob(pattern))
+
+
+def distribute(files: list[str], max_runners: int) -> list[list[str]]:
+    """Split *files* into *max_runners* balanced groups (round-robin)."""
+    groups: list[list[str]] = [[] for _ in range(max_runners)]
+    for i, f in enumerate(files):
+        groups[i % max_runners].append(f)
+    return groups
+
+
+def files_for_runner(files: list[str], runner_index: int, max_runners: int) -> list[str]:
+    """Return the subset of files assigned to *runner_index* (0-based)."""
+    groups = distribute(files, max_runners)
+    if runner_index < 0 or runner_index >= len(groups):
+        raise click.ClickException(
+            _("Runner index {index} out of range (0..{max})", index=runner_index, max=max_runners - 1)
+        )
+    return groups[runner_index]
+
+
+def _write_github_env(key: str, value: str) -> None:
+    gh_env = os.environ.get("GITHUB_ENV")
+    if not gh_env:
+        raise click.ClickException("GITHUB_ENV environment variable is not set")
+    with open(gh_env, "a") as f:  # noqa: PTH123
+        f.write(f"{key}={value}\n")
+
+
+@click.command()
+@click.option("--pattern", required=True, help="Glob pattern for files to distribute.")
+@click.option(
+    "--runner-index",
+    type=int,
+    default=None,
+    help="One-based runner index. If omitted, prints all groups.",
+)
+@click.option(
+    "--max-runners",
+    type=int,
+    default=DEFAULT_MAX_RUNNERS,
+    show_default=True,
+    help="Total number of parallel runners.",
+)
+@click.option(
+    "--github-env",
+    is_flag=True,
+    default=False,
+    help="Write ASSIGNED_FILES and SKIP to $GITHUB_ENV.",
+)
+@click.option(
+    "--skip-if-excess",
+    is_flag=True,
+    default=False,
+    help="With --github-env: write SKIP=true when runner-index exceeds max-runners.",
+)
+def main(pattern: str, runner_index: int | None, max_runners: int, github_env: bool, skip_if_excess: bool) -> None:
+    files = discover_files(pattern)
+
+    if runner_index is None:
+        groups = distribute(files, max_runners)
+        for i, group in enumerate(groups):
+            labels = " ".join(group) if group else "(none)"
+            click.echo(f"Runner {i}: {labels}")
+        return
+
+    if skip_if_excess and github_env and runner_index > max_runners:
+        click.echo(f"Skipping — runner index {runner_index} > max runners {max_runners}")
+        _write_github_env("ASSIGNED_FILES", "")
+        _write_github_env("SKIP", "true")
+        return
+
+    if runner_index < 1:
+        raise click.ClickException(f"Runner index {runner_index} is out of range (must be >= 1)")
+
+    zero_based = runner_index - 1
+    assigned = files_for_runner(files, zero_based, max_runners)
+    encoded = "\n".join(assigned)
+
+    if github_env:
+        _write_github_env("ASSIGNED_FILES", encoded)
+        _write_github_env("SKIP", "false")
+        click.echo(f"Assigned {len(assigned)} files to runner {runner_index}")
+        return
+
+    click.echo(encoded)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

src/devx/ci/integration_guard.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""Run integration tests with cross-runner failure detection.
+
+Wraps ``pytest`` with the same Gitea API polling mechanism used by
+``molecule_ci_guard``. If any other integration-tests matrix runner
+reports failure, the current pytest subprocess is killed and this runner
+exits early with code 1.
+
+JUnit XML is generated via pytest's ``--junitxml`` flag (passed through
+to the pytest invocation).
+
+Usage::
+
+    python3 -m devx.ci.integration_guard \\
+        --junit-output junit-results/runner-1.xml \\
+        -- test_file1.py test_file2.py
+
+    # With pytest options
+    python3 -m devx.ci.integration_guard \\
+        --junit-output junit-results/runner-1.xml \\
+        -- -x -v --tb=short test_file1.py
+
+Environment variables:
+    GITEA_URL           Base URL of the Gitea instance.
+    REPO_TOKEN          API token with repo access.
+    RUN_ID              Workflow run ID (GITHUB_RUN_ID).
+    JOB_NAME            Base job name (GITHUB_JOB), e.g. "integration-tests".
+    MATRIX_INDEX        Current matrix index (runner-index).
+    GITEA_REPOSITORY    Repository in "owner/repo" format.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+import signal
+import subprocess  # nosec B404
+import sys
+import threading
+import time
+
+import click
+
+from devx.i18n import _
+from devx.molecule.molecule_ci_guard import (
+    poll_for_other_failures,
+)
+
+POLL_INTERVAL = 10
+
+
+@click.command(context_settings={"ignore_unknown_options": True})
+@click.argument("pytest_args", nargs=-1, type=click.UNPROCESSED, required=True)
+@click.option(
+    "--junit-output",
+    default=None,
+    help="Path for JUnit XML output (passed to pytest as --junitxml).",
+)
+def cli(pytest_args: tuple[str, ...], junit_output: str | None) -> None:
+    """Run pytest with cross-runner failure detection."""
+    gitea_url = os.environ.get("GITEA_URL", "")
+    token = os.environ.get("REPO_TOKEN", "")
+    run_id = int(os.environ.get("RUN_ID", "0"))
+    job_name = os.environ.get("JOB_NAME", "integration-tests")
+    current_index = int(os.environ.get("MATRIX_INDEX", "0"))
+    repository = os.environ.get("GITEA_REPOSITORY", "oblachno-oss/devx")
+    owner, _sep, repo = repository.partition("/")
+    if not owner or not repo:
+        owner, repo = "oblachno-oss", "devx"
+
+    if not all([gitea_url, token, run_id]):
+        click.echo(_("GITEA_URL/REPO_TOKEN/RUN_ID not set; running without cross-runner cancellation."))
+
+    stop_event = threading.Event()
+    failed_event = threading.Event()
+
+    if gitea_url and token and run_id:
+        poller = threading.Thread(
+            target=poll_for_other_failures,
+            args=(
+                gitea_url,
+                owner,
+                repo,
+                token,
+                run_id,
+                job_name,
+                current_index,
+                stop_event,
+                failed_event,
+            ),
+            daemon=True,
+        )
+        poller.start()
+
+    cmd = [sys.executable, "-m", "pytest"]
+    if junit_output:
+        cmd.extend(["--junitxml", junit_output])
+    cmd.extend(pytest_args)
+
+    click.echo(f"Running: {' '.join(cmd)}")
+
+    process = subprocess.Popen(  # nosec B603
+        cmd,
+        preexec_fn=os.setsid,
+    )
+
+    try:
+        while process.poll() is None:
+            if failed_event.is_set():
+                with contextlib.suppress(ProcessLookupError):
+                    os.killpg(os.getpgid(process.pid), signal.SIGTERM)
+                try:
+                    process.wait(timeout=10)
+                except subprocess.TimeoutExpired:
+                    with contextlib.suppress(ProcessLookupError):
+                        os.killpg(os.getpgid(process.pid), signal.SIGKILL)
+                    process.wait()
+                click.echo(_("Integration tests cancelled — another runner failed."))
+                sys.exit(1)
+            time.sleep(1)
+    except KeyboardInterrupt:
+        with contextlib.suppress(ProcessLookupError):
+            os.killpg(os.getpgid(process.pid), signal.SIGTERM)
+        process.wait()
+        sys.exit(1)
+    finally:
+        stop_event.set()
+
+    rc = process.returncode
+    if rc != 0:
+        click.echo(_("Integration tests failed with exit code {code}", code=rc))
+    else:
+        click.echo(_("Integration tests passed."))
+    sys.exit(rc)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    cli()

src/devx/ci/merge_junit.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+"""Merge multiple JUnit XML reports into a single report.
+
+Used by CI workflows to consolidate JUnit XML files produced by
+parallel matrix runners into a single merged report for archival
+and dashboard consumption.
+
+Usage::
+
+    python3 -m devx.ci.merge_junit \\
+        --pattern "junit-results/runner-*.xml" \\
+        --output junit-merged.xml
+
+Exit code is non-zero if any merged test suite reports failures,
+making this suitable as a CI gating step after matrix jobs.
+"""
+
+from __future__ import annotations
+
+import glob
+import sys
+import xml.etree.ElementTree as ET  # nosec B405
+
+import click
+
+from devx.i18n import _
+
+
+def merge_files(pattern: str) -> tuple[ET.Element, int, int]:
+    """Merge JUnit XML files matching *pattern* into a single ``<testsuites>`` element.
+
+    Returns ``(merged_element, total_tests, total_failures)``.
+    If no files match, returns an empty ``<testsuites>`` with zero counts.
+    """
+    files = sorted(glob.glob(pattern))
+    merged = ET.Element("testsuites")
+    total_tests = 0
+    total_failures = 0
+
+    for f in files:
+        tree = ET.parse(f)  # nosec B314
+        suite = tree.getroot()
+        # Handle both <testsuites> (wrapper) and <testsuite> (single) roots
+        if suite.tag == "testsuites":
+            for child in suite:
+                merged.append(child)
+                total_tests += int(child.get("tests", 0))
+                total_failures += int(child.get("failures", 0))
+        else:
+            merged.append(suite)
+            total_tests += int(suite.get("tests", 0))
+            total_failures += int(suite.get("failures", 0))
+
+    merged.set("tests", str(total_tests))
+    merged.set("failures", str(total_failures))
+    return merged, total_tests, total_failures
+
+
+@click.command()
+@click.option(
+    "--pattern",
+    default="junit-results/runner-*.xml",
+    show_default=True,
+    help="Glob pattern for input JUnit XML files.",
+)
+@click.option(
+    "--output",
+    default="junit-merged.xml",
+    show_default=True,
+    help="Output path for the merged JUnit XML file.",
+)
+def main(pattern: str, output: str) -> None:
+    merged, total_tests, total_failures = merge_files(pattern)
+
+    if total_tests == 0:
+        click.echo(_("No JUnit reports found matching {pattern} — skipping merge.", pattern=pattern))
+        return
+
+    ET.indent(merged)
+    tree = ET.ElementTree(merged)
+    tree.write(output, encoding="UTF-8", xml_declaration=True)
+    click.echo(
+        _(
+            "Merged {count} reports: {tests} tests, {failures} failures → {output}",
+            count=len(glob.glob(pattern)),
+            tests=total_tests,
+            failures=total_failures,
+            output=output,
+        )
+    )
+
+    if total_failures > 0:
+        sys.exit(1)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

src/devx/ci/notify_failure.py

@@ -10,13 +10,20 @@ Usage:
     --repo <owner/repo> \
     --run-id <run_id> \
     --workflow <workflow_name> \
-    --commit <commit_sha>
+    --commit <commit_sha> \
+    --auto-login
+
+With ``--auto-login``, the script configures the tea CLI login profile
+from ``REPO_TOKEN`` and ``DEVX_GITEA_API_URL`` before creating the issue,
+eliminating the need for a separate ``tea login add`` step in the workflow.
 """
 
 from __future__ import annotations
 
 import logging
 import os
+import shutil
+import subprocess  # nosec B404
 
 import click
 from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnknownVariableType]
@@ -30,6 +37,49 @@ load_dotenv()
 logger = logging.getLogger("devx")
 
 
+def _configure_tea_login(login_name: str = "devx") -> None:
+    """Configure tea CLI login from REPO_TOKEN and DEVX_GITEA_API_URL.
+
+    Idempotent: if a login with the same name already exists, it is not re-added.
+    Skips silently if tea is not installed or REPO_TOKEN is not set.
+    """
+    tea_bin = shutil.which("tea")
+    if tea_bin is None:
+        click.echo("notify_failure: tea not installed — skipping login configuration.")
+        return
+
+    token = os.environ.get("REPO_TOKEN", "")
+    if not token:
+        click.echo("notify_failure: REPO_TOKEN not set — skipping login configuration.")
+        return
+
+    gitea_url = GITEA_API_URL.replace("/api/v1", "")
+
+    result = subprocess.run(  # nosec B603
+        [tea_bin, "login", "list", "--output", "simple"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode == 0 and login_name in result.stdout:
+        click.echo(f"notify_failure: tea login '{login_name}' already configured.")
+        return
+
+    click.echo(f"notify_failure: configuring tea login '{login_name}' for {gitea_url}...")
+    subprocess.run(  # nosec B603
+        [tea_bin, "login", "add", "--name", login_name, "--url", gitea_url, "--token", token],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    subprocess.run(  # nosec B603
+        [tea_bin, "login", "default", login_name],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
 def _create_issue_via_tea(repo: str, title: str, body: str) -> int:
     """Create issue via tea CLI. Returns issue index.
 
@@ -62,11 +112,20 @@ def _create_issue_via_tea(repo: str, title: str, body: str) -> int:
 @click.option("--run-id", required=True, help="CI run ID.")
 @click.option("--workflow", required=True, help="Workflow name.")
 @click.option("--commit", required=True, help="Commit SHA.")
-def main(repo: str, run_id: str, workflow: str, commit: str) -> None:
+@click.option(
+    "--auto-login",
+    is_flag=True,
+    default=False,
+    help="Configure tea CLI login from REPO_TOKEN before creating the issue.",
+)
+def main(repo: str, run_id: str, workflow: str, commit: str, auto_login: bool) -> None:
     token = os.environ.get("REPO_TOKEN", "")
     if not token:
         raise click.ClickException(_("ERROR: REPO_TOKEN is not set."))
 
+    if auto_login:
+        _configure_tea_login()
+
     title = f"[CI] {workflow} workflow failed (run #{run_id})"
     body = (
         f"The **{workflow}** workflow failed.\n\n"

src/devx/ci/publish.py

@@ -164,7 +164,14 @@ def _default_gitea_registry_url() -> str:
     "or a URL derived from GITEA_API_URL. When set, publishes to Gitea PyPI "
     "instead of standard PyPI (unless PYPI_TOKEN is also set).",
 )
-def main(tag: str, repo: str, registry_url: str | None) -> None:
+@click.option(
+    "--skip-build",
+    is_flag=True,
+    default=False,
+    help="Skip package build and PyPI publish (for non-Python repos that only "
+    "need a Gitea release with git-cliff notes).",
+)
+def main(tag: str, repo: str, registry_url: str | None, skip_build: bool) -> None:
     gitea_token = os.environ.get("REPO_TOKEN", "")
     if not gitea_token:
         raise click.ClickException(_("ERROR: REPO_TOKEN is not set."))
@@ -177,21 +184,24 @@ def main(tag: str, repo: str, registry_url: str | None) -> None:
     if not registry_url:
         registry_url = _default_gitea_registry_url()
 
-    build_package()
+    if not skip_build:
+        build_package()
 
-    if pypi_token:
-        # Standard PyPI flow takes precedence when PYPI_TOKEN is set
-        publish_to_pypi(pypi_token)
-    elif registry_url:
-        # Gitea PyPI registry flow
-        publish_to_gitea_registry(registry_url, gitea_token)
-    else:
-        click.echo(
-            _(
-                "PYPI_TOKEN not set and no registry URL configured — "
-                "skipping PyPI publish. No worries, we'll just create the Gitea release."
+        if pypi_token:
+            # Standard PyPI flow takes precedence when PYPI_TOKEN is set
+            publish_to_pypi(pypi_token)
+        elif registry_url:
+            # Gitea PyPI registry flow
+            publish_to_gitea_registry(registry_url, gitea_token)
+        else:
+            click.echo(
+                _(
+                    "PYPI_TOKEN not set and no registry URL configured — "
+                    "skipping PyPI publish. No worries, we'll just create the Gitea release."
+                )
             )
-        )
+    else:
+        click.echo(_("--skip-build: skipping package build and PyPI publish."))
 
     tea = TeaCLI(repo=repo)
     release_body = generate_release_notes(tag)

src/devx/ci/push_badges.py

@@ -17,9 +17,11 @@ Usage::
 
 from __future__ import annotations
 
+import contextlib
 import re
 import subprocess  # nosec B404
 import sys
+import time
 from pathlib import Path
 from typing import Any
 
@@ -160,13 +162,34 @@ def update_readme_with_badge_sha(badges_sha: str, repo_root: Path | None = None)
     default=False,
     help="Skip updating README with cache-busting URLs (for local testing).",
 )
-def main(output_dir: str, branch: str, no_readme_update: bool) -> None:
+@click.option(
+    "--retries",
+    default=1,
+    type=int,
+    help="Number of attempts on git push failures (default: 1, no retry). "
+    "Between attempts, fetches latest master and waits 10s.",
+)
+def main(output_dir: str, branch: str, no_readme_update: bool, retries: int) -> None:
     """Generate badges and push them to the badges branch."""
-    fetch_latest_master(branch)
-    generate_badges(output_dir)
-    badges_sha = push_to_badges_branch(output_dir)
-    if not no_readme_update:
-        update_readme_with_badge_sha(badges_sha)
+    last_error: Exception | None = None
+    for attempt in range(1, retries + 1):
+        try:
+            fetch_latest_master(branch)
+            generate_badges(output_dir)
+            badges_sha = push_to_badges_branch(output_dir)
+            if not no_readme_update:
+                update_readme_with_badge_sha(badges_sha)
+            return
+        except (subprocess.CalledProcessError, RuntimeError) as exc:
+            last_error = exc
+            if attempt < retries:
+                click.echo(f"Badge push attempt {attempt}/{retries} failed — retrying: {exc}")
+                time.sleep(10)
+                with contextlib.suppress(subprocess.CalledProcessError):
+                    fetch_latest_master(branch)
+            else:
+                click.echo(f"Badge push failed after {retries} attempts: {exc}")
+    raise click.ClickException(f"Badge push failed after {retries} attempts: {last_error}")
 
 
 if __name__ == "__main__":  # pragma: no cover

src/devx/ci/release.py

@@ -43,6 +43,7 @@ import sys
 import click
 from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnknownVariableType]
 
+from devx.ci._shared import get_latest_tag
 from devx.ci.classify_changes import has_user_facing_changes  # cross-CI import, needs PYTHONPATH=.
 from devx.i18n import _
 
@@ -72,14 +73,6 @@ def run_cmd(args: list[str], check: bool = True, capture: bool = True) -> subpro
     return result
 
 
-def get_latest_tag() -> str:
-    """Get the latest git tag, or empty string if none exists."""
-    result = run_cmd(["git", "describe", "--tags", "--abbrev=0"], check=False)
-    if result.returncode != 0:
-        return ""
-    return result.stdout.strip()
-
-
 def tag_exists(tag: str) -> bool:
     """Check if a git tag already exists."""
     result = run_cmd(["git", "tag", "-l", tag], check=False)

src/devx/cli.py

@@ -151,6 +151,27 @@ def ci_validate_commit_msg(args: tuple[str, ...]) -> None:
     _run_module("devx.ci.validate_commit_msg", list(args))
 
 
+@ci.command("distribute-files")
+@click.argument("args", nargs=-1)
+def ci_distribute_files(args: tuple[str, ...]) -> None:
+    """Distribute files across parallel runners (round-robin)."""
+    _run_module("devx.ci.distribute_files", list(args))
+
+
+@ci.command("merge-junit")
+@click.argument("args", nargs=-1)
+def ci_merge_junit(args: tuple[str, ...]) -> None:
+    """Merge multiple JUnit XML reports into a single report."""
+    _run_module("devx.ci.merge_junit", list(args))
+
+
+@ci.command("integration-guard")
+@click.argument("args", nargs=-1)
+def ci_integration_guard(args: tuple[str, ...]) -> None:
+    """Run pytest with cross-runner failure detection and JUnit output."""
+    _run_module("devx.ci.integration_guard", list(args))
+
+
 @cli.group()
 def tools() -> None:
     """Development tool commands."""

src/devx/molecule/distribute_molecule.py

@@ -29,6 +29,7 @@ from devx.molecule.platforms import PLATFORMS
 
 DEFAULT_MAX_RUNNERS = 3
 MOLECULE_ROOT = Path("ansible/roles/gitea-runner/molecule")
+DEFAULT_ROLES_ROOT = Path("ansible/roles")
 
 
 @dataclass(frozen=True)
@@ -52,6 +53,31 @@ class TestPair:
         )
 
 
+@dataclass(frozen=True)
+class MultiRoleTestPair:
+    """A (role, scenario, platform) combination for multi-role projects."""
+
+    role: str
+    scenario: str
+    platform: dict[str, str]
+
+    def encode(self) -> str:
+        """Serialize to a pipe-delimited string: ``role|scenario|platform_name|image|command``."""
+        return (
+            f"{self.role}|{self.scenario}|{self.platform['name']}|{self.platform['image']}|{self.platform['command']}"
+        )
+
+    @staticmethod
+    def decode(encoded: str) -> MultiRoleTestPair:
+        """Deserialize from a pipe-delimited string."""
+        parts = encoded.split("|")
+        return MultiRoleTestPair(
+            role=parts[0],
+            scenario=parts[1],
+            platform={"name": parts[2], "image": parts[3], "command": parts[4]},
+        )
+
+
 def discover_scenarios(root: Path | None = None) -> list[str]:
     """Return sorted list of molecule scenario directory names."""
     if root is None:
@@ -62,6 +88,33 @@ def discover_scenarios(root: Path | None = None) -> list[str]:
     return sorted(scenarios)
 
 
+def discover_multi_role_scenarios(roles_root: Path | None = None) -> list[tuple[str, str]]:
+    """Discover (role, scenario) pairs across all roles under *roles_root*.
+
+    Scans ``roles_root/*/molecule/*/`` for scenario directories, skipping
+    ``common`` and directories starting with ``_``. Returns a sorted list of
+    ``(role_name, scenario_name)`` tuples.
+    """
+    if roles_root is None:
+        roles_root = DEFAULT_ROLES_ROOT
+    if not roles_root.is_dir():
+        raise click.ClickException(_("Roles directory not found: {path}", path=str(roles_root)))
+    pairs: list[tuple[str, str]] = []
+    for role_dir in sorted(roles_root.iterdir()):
+        if not role_dir.is_dir():
+            continue
+        mol_dir = role_dir / "molecule"
+        if not mol_dir.is_dir():
+            continue
+        for scenario_dir in mol_dir.iterdir():
+            if not scenario_dir.is_dir():
+                continue
+            if scenario_dir.name.startswith("_") or scenario_dir.name == "common":
+                continue
+            pairs.append((role_dir.name, scenario_dir.name))
+    return pairs
+
+
 def build_pairs(scenarios: list[str], platforms: list[dict[str, str]] | None = None) -> list[TestPair]:
     """Build the full cross-product of scenarios and platforms."""
     if platforms is None:
@@ -69,6 +122,36 @@ def build_pairs(scenarios: list[str], platforms: list[dict[str, str]] | None = N
     return [TestPair(s, p) for s in scenarios for p in platforms]
 
 
+def build_multi_role_pairs(
+    role_scenarios: list[tuple[str, str]],
+    platforms: list[dict[str, str]] | None = None,
+) -> list[MultiRoleTestPair]:
+    """Build the full cross-product of (role, scenario) pairs and platforms."""
+    if platforms is None:
+        platforms = PLATFORMS
+    return [MultiRoleTestPair(r, s, p) for r, s in role_scenarios for p in platforms]
+
+
+def distribute_multi_role(pairs: list[MultiRoleTestPair], max_runners: int) -> list[list[MultiRoleTestPair]]:
+    """Split *pairs* into *max_runners* balanced groups (round-robin)."""
+    groups: list[list[MultiRoleTestPair]] = [[] for _ in range(max_runners)]
+    for i, pair in enumerate(pairs):
+        groups[i % max_runners].append(pair)
+    return groups
+
+
+def multi_role_pairs_for_runner(
+    pairs: list[MultiRoleTestPair], runner_index: int, max_runners: int
+) -> list[MultiRoleTestPair]:
+    """Return the subset of multi-role pairs assigned to *runner_index* (0-based)."""
+    groups = distribute_multi_role(pairs, max_runners)
+    if runner_index < 0 or runner_index >= len(groups):
+        raise click.ClickException(
+            _("Runner index {index} out of range (0..{max})", index=runner_index, max=max_runners - 1)
+        )
+    return groups[runner_index]
+
+
 def distribute(pairs: list[TestPair], max_runners: int) -> list[list[TestPair]]:
     """Split *pairs* into *max_runners* balanced groups (round-robin)."""
     groups: list[list[TestPair]] = [[] for _ in range(max_runners)]
@@ -142,6 +225,19 @@ def _write_github_env(key: str, value: str) -> None:
     default=False,
     help="With --github-env: write SKIP=true when runner-index exceeds max-runners.",
 )
+@click.option(
+    "--molecule-root",
+    type=click.Path(exists=True, file_okay=False, path_type=Path),
+    default=None,
+    help="Custom molecule directory (single-role mode). Default: ansible/roles/gitea-runner/molecule.",
+)
+@click.option(
+    "--roles-root",
+    type=click.Path(exists=True, file_okay=False, path_type=Path),
+    default=None,
+    help="Roles directory for multi-role discovery (scans */molecule/*/). "
+    "Use this for projects with multiple Ansible roles. Default: disabled (single-role mode).",
+)
 def cli(
     runner_index: int | None,
     max_runners: int,
@@ -149,8 +245,47 @@ def cli(
     list_platforms: bool,
     github_env: bool,
     skip_if_excess: bool,
+    molecule_root: Path | None,
+    roles_root: Path | None,
 ) -> None:
-    scenarios = discover_scenarios()
+    # Multi-role mode: discover (role, scenario) pairs across all roles
+    if roles_root is not None:
+        role_scenarios = discover_multi_role_scenarios(roles_root)
+        if list_all:
+            for role, scenario in role_scenarios:
+                click.echo(f"{role}|{scenario}")
+            return
+        if list_platforms:
+            for p in PLATFORMS:
+                click.echo(f"{p['name']}|{p['image']}|{p['command']}")
+            return
+        pairs_mr = build_multi_role_pairs(role_scenarios)
+        if runner_index is None:
+            groups = distribute_multi_role(pairs_mr, max_runners)
+            for i, group in enumerate(groups):
+                labels = " ".join(p.encode() for p in group) if group else "(none)"
+                click.echo(f"Runner {i}: {labels}")
+            return
+        if skip_if_excess and github_env and runner_index > max_runners:
+            click.echo(f"Skipping — runner index {runner_index} > max runners {max_runners}")
+            _write_github_env("TEST_PAIRS", "")
+            _write_github_env("SKIP", "true")
+            return
+        if runner_index < 1:
+            raise click.ClickException(f"Runner index {runner_index} is out of range (must be >= 1)")
+        zero_based = runner_index - 1
+        assigned = multi_role_pairs_for_runner(pairs_mr, zero_based, max_runners)
+        encoded = " ".join(p.encode() for p in assigned)
+        if github_env:
+            _write_github_env("TEST_PAIRS", encoded)
+            _write_github_env("SKIP", "false")
+            click.echo(f"Assigned pairs: {encoded}")
+            return
+        click.echo(encoded)
+        return
+
+    # Single-role mode (default or --molecule-root)
+    scenarios = discover_scenarios(molecule_root)
     if list_all:
         for s in scenarios:
             click.echo(s)

src/devx/molecule/molecule_ci_guard.py

@@ -1,7 +1,11 @@
 #!/usr/bin/env python3
 """Run molecule tests sequentially while polling Gitea for other runner failures.
 
-Each pair is encoded as ``scenario|platform_name|platform_image|platform_command``.
+Each pair is encoded as one of:
+
+- **Single-role (4-part):** ``scenario|platform_name|platform_image|platform_command``
+- **Multi-role (5-part):** ``role|scenario|platform_name|platform_image|platform_command``
+
 Pairs are executed one at a time (molecule scenarios share temp directories and
 Docker networks, so parallel execution within a single runner is unsafe).
 
@@ -9,8 +13,17 @@ A background thread polls the Gitea API. If any other molecule matrix runner
 reports failure, the current molecule subprocess is killed and this runner
 exits early with code 1.
 
-Usage:
-    python3 -m devx.molecule.molecule_ci_guard <pair1> <pair2> ...
+JUnit XML is generated when ``--junit-output`` is provided, recording each
+pair as a testcase with pass/fail status and elapsed time.
+
+Usage::
+
+    # Single-role (grm-style)
+    python3 -m devx.molecule.molecule_ci_guard pair1 pair2 ...
+    # Multi-role (infra-style)
+    python3 -m devx.molecule.molecule_ci_guard --roles-root ansible/roles pair1 pair2 ...
+    # With JUnit output
+    python3 -m devx.molecule.molecule_ci_guard --junit-output junit-results/runner-1.xml pair1 pair2 ...
 
 Environment variables:
     GITEA_URL           Base URL of the Gitea instance.
@@ -30,6 +43,7 @@ import subprocess  # nosec B404
 import sys
 import threading
 import time
+import xml.etree.ElementTree as ET  # nosec B405
 from pathlib import Path
 
 import click
@@ -95,9 +109,23 @@ def build_molecule_cmd(scenario: str) -> list[str]:
     return cmd
 
 
+def parse_pair(pair: str) -> tuple[str, str, str, str, str]:
+    """Parse a pair string into (role, scenario, platform_name, platform_image, platform_command).
+
+    Supports both 4-part (single-role) and 5-part (multi-role) formats.
+    For 4-part pairs, role is empty (caller uses default role dir).
+    """
+    parts = pair.split("|")
+    if len(parts) == 4:
+        return "", parts[0], parts[1], parts[2], parts[3]
+    if len(parts) == 5:
+        return parts[0], parts[1], parts[2], parts[3], parts[4]
+    raise click.ClickException(f"Invalid pair format: {pair!r} (expected 4 or 5 pipe-delimited parts)")
+
+
 def build_env_for_pair(pair: str, base_env: dict[str, str]) -> dict[str, str]:
     """Build environment for a single molecule pair."""
-    scenario, platform_name, platform_image, platform_command = pair.split("|")
+    _role, _scenario, platform_name, platform_image, platform_command = parse_pair(pair)
     env = base_env.copy()
     env["MOLECULE_PLATFORM_NAME"] = platform_name
     env["MOLECULE_PLATFORM_IMAGE"] = platform_image
@@ -109,9 +137,66 @@ def build_env_for_pair(pair: str, base_env: dict[str, str]) -> dict[str, str]:
     return env
 
 
+def resolve_role_dir(role: str, roles_root: Path | None, repo_root: Path) -> Path:
+    """Resolve the working directory for a molecule pair.
+
+    For multi-role pairs (role non-empty), uses ``roles_root/role``.
+    For single-role pairs, uses ``repo_root/ansible/roles/gitea-runner``.
+    """
+    if role:
+        if roles_root is None:
+            roles_root = repo_root / "ansible" / "roles"
+        return roles_root / role
+    return repo_root / "ansible" / "roles" / "gitea-runner"
+
+
+def write_junit_report(
+    output_path: str,
+    testcases: list[dict],
+    runner_index: int,
+) -> None:
+    """Write a JUnit XML report from collected test case results.
+
+    Each testcase dict has: role, scenario, time (float), passed (bool), error (str|None).
+    """
+    suite = ET.Element(
+        "testsuite",
+        name=f"molecule-runner-{runner_index}",
+        tests=str(len(testcases)),
+        failures=str(sum(1 for tc in testcases if not tc["passed"])),
+    )
+    for tc in testcases:
+        classname = tc["role"] if tc["role"] else "molecule"
+        elem = ET.SubElement(
+            suite,
+            "testcase",
+            classname=classname,
+            name=tc["scenario"],
+            time=f"{tc['time']:.1f}",
+        )
+        if not tc["passed"]:
+            fail = ET.SubElement(elem, "failure")
+            fail.text = tc.get("error") or "molecule test failed"
+    tree = ET.ElementTree(suite)
+    ET.indent(tree)
+    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
+    tree.write(output_path, encoding="UTF-8", xml_declaration=True)
+
+
 @click.command()
 @click.argument("pairs", nargs=-1, required=True)
-def cli(pairs: tuple[str, ...]) -> None:
+@click.option(
+    "--junit-output",
+    default=None,
+    help="Path to write JUnit XML report (e.g. junit-results/runner-1.xml).",
+)
+@click.option(
+    "--roles-root",
+    type=click.Path(exists=True, file_okay=False, path_type=Path),
+    default=None,
+    help="Root directory for multi-role pairs (e.g. ansible/roles). Required when pairs use 5-part format.",
+)
+def cli(pairs: tuple[str, ...], junit_output: str | None, roles_root: Path | None) -> None:
     """Run molecule pairs sequentially, stop if another CI runner fails."""
     gitea_url = os.environ.get("GITEA_URL", "")
     token = os.environ.get("REPO_TOKEN", "")
@@ -119,7 +204,7 @@ def cli(pairs: tuple[str, ...]) -> None:
     job_name = os.environ.get("JOB_NAME", "molecule-tests")
     current_index = int(os.environ.get("MATRIX_INDEX", "0"))
     repository = os.environ.get("GITEA_REPOSITORY", "oblachno-oss/devx")
-    owner, sep, repo = repository.partition("/")
+    owner, _sep, repo = repository.partition("/")
     if not owner or not repo:
         owner, repo = "oblachno-oss", "devx"
 
@@ -127,7 +212,6 @@ def cli(pairs: tuple[str, ...]) -> None:
         click.echo(_("GITEA_URL/REPO_TOKEN/RUN_ID not set; running without cross-runner cancellation."))
 
     repo_root = Path(__file__).resolve().parent.parent.parent.parent
-    role_dir = repo_root / "ansible" / "roles" / "gitea-runner"
 
     base_env = os.environ.copy()
     base_env.setdefault("DOCKER_HOST", f"unix:///run/user/{os.getuid()}/docker.sock")
@@ -154,24 +238,24 @@ def cli(pairs: tuple[str, ...]) -> None:
         )
         poller.start()
 
+    testcases: list[dict] = []
+
     try:
         for pair in pairs:
             if failed_event.is_set():
                 sys.exit(1)
 
-            parts = pair.split("|")
-            if len(parts) < 2:
-                raise click.ClickException(f"Invalid pair format: {pair!r} (expected at least 2 pipe-delimited parts)")
-            scenario = parts[0]
-            platform_name = parts[1]
+            role, scenario, platform_name, _img, _cmd = parse_pair(pair)
             click.echo(_("Running: {scenario} on {platform}", scenario=scenario, platform=platform_name))
 
             cmd = build_molecule_cmd(scenario)
             env = build_env_for_pair(pair, base_env)
+            cwd = resolve_role_dir(role, roles_root, repo_root)
 
+            start = time.time()
             process = subprocess.Popen(  # nosec B603
                 cmd,
-                cwd=str(role_dir),
+                cwd=str(cwd),
                 env=env,
                 preexec_fn=os.setsid,
             )
@@ -187,6 +271,18 @@ def cli(pairs: tuple[str, ...]) -> None:
                             with contextlib.suppress(ProcessLookupError):
                                 os.killpg(os.getpgid(process.pid), signal.SIGKILL)
                             process.wait()
+                        elapsed = time.time() - start
+                        testcases.append(
+                            {
+                                "role": role,
+                                "scenario": scenario,
+                                "time": elapsed,
+                                "passed": False,
+                                "error": "Cancelled — another runner failed",
+                            }
+                        )
+                        if junit_output:
+                            write_junit_report(junit_output, testcases, current_index)
                         sys.exit(1)
                     time.sleep(1)
             except KeyboardInterrupt:
@@ -196,13 +292,30 @@ def cli(pairs: tuple[str, ...]) -> None:
                 sys.exit(1)
 
             rc = process.returncode
+            elapsed = time.time() - start
+            passed = rc == 0
+
+            testcases.append(
+                {
+                    "role": role,
+                    "scenario": scenario,
+                    "time": elapsed,
+                    "passed": passed,
+                    "error": f"Exit code: {rc}" if not passed else None,
+                }
+            )
+
             if rc != 0:
                 click.echo(_("FAILED: {pair} exited with code {code}", pair=pair, code=rc))
+                if junit_output:
+                    write_junit_report(junit_output, testcases, current_index)
                 sys.exit(rc)
 
             click.echo(_("PASSED: {pair}", pair=pair))
 
         click.echo(_("All molecule tests passed."))
+        if junit_output:
+            write_junit_report(junit_output, testcases, current_index)
     finally:
         stop_event.set()
 

src/devx/opentofu.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""OpenTofu output helpers for CI/CD deployment scripts.
+
+Provides reusable functions for extracting values from ``tofu output``
+in a structured way.  This eliminates duplicated ``subprocess.run``
+boilerplate across deployment and smoke-test scripts.
+
+Typical usage::
+
+    from devx.opentofu import get_tofu_output, get_tofu_vm_ip
+
+    vms = get_tofu_output("customer_vms", cwd="tofu/environments/staging",
+                          env={"HCLOUD_TOKEN": token})
+    ip = get_tofu_vm_ip("customer_vms", "oblachno", cwd="tofu/environments/staging",
+                        env={"HCLOUD_TOKEN": token})
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess  # nosec B404
+from pathlib import Path
+from typing import Any
+
+
+def get_tofu_output(
+    output_name: str,
+    cwd: str | Path | None = None,
+    env: dict[str, str] | None = None,
+) -> Any:
+    """Run ``tofu output -json <output_name>`` and return parsed JSON.
+
+    Args:
+        output_name: The OpenTofu output name to query (e.g. ``customer_vms``).
+        cwd: Directory to run the command in (the tofu env directory).
+        env: Environment variables for the subprocess (e.g. ``{"HCLOUD_TOKEN": ...}``).
+             If ``None``, inherits the current environment.
+
+    Returns:
+        Parsed JSON value from the tofu output.
+
+    Raises:
+        RuntimeError: If ``tofu output`` exits with a non-zero code.
+        json.JSONDecodeError: If stdout is not valid JSON.
+    """
+    result = subprocess.run(  # nosec B603, B607
+        ["tofu", "output", "-json", output_name],
+        cwd=str(cwd) if cwd else None,
+        capture_output=True,
+        text=True,
+        check=False,
+        env=env,
+    )
+    if result.returncode != 0:
+        raise RuntimeError(f"tofu output failed: {result.stderr}")
+    return json.loads(result.stdout)
+
+
+def get_tofu_vm_ip(
+    output_name: str,
+    vm_key: str,
+    cwd: str | Path | None = None,
+    env: dict[str, str] | None = None,
+    ip_field: str = "ipv4",
+) -> str:
+    """Extract a VM IPv4 address from a tofu output map.
+
+    The output is expected to be a JSON object mapping VM names to objects
+    containing an IP field (default ``ipv4``)::
+
+        {"staging": {"ipv4": "1.2.3.4", ...}, ...}
+
+    Args:
+        output_name: The tofu output name (e.g. ``customer_vms``).
+        vm_key: The key inside the output map (e.g. ``"staging"``).
+        cwd: Directory to run the command in.
+        env: Environment variables for the subprocess.
+        ip_field: The field name for the IP address (default ``ipv4``).
+
+    Returns:
+        The IP address string, or empty string if not found.
+    """
+    data = get_tofu_output(output_name, cwd=cwd, env=env)
+    if not isinstance(data, dict):
+        return ""
+    return str(data.get(vm_key, {}).get(ip_field, ""))
+
+
+def get_tofu_vm_field(
+    output_name: str,
+    vm_key: str,
+    field: str,
+    cwd: str | Path | None = None,
+    env: dict[str, str] | None = None,
+) -> str:
+    """Extract an arbitrary field from a VM entry in tofu output.
+
+    Like :func:`get_tofu_vm_ip` but for any field (e.g. ``volume_linux_device``).
+
+    Args:
+        output_name: The tofu output name.
+        vm_key: The key inside the output map.
+        field: The field name to extract.
+        cwd: Directory to run the command in.
+        env: Environment variables for the subprocess.
+
+    Returns:
+        The field value as a string, or empty string if not found.
+    """
+    data = get_tofu_output(output_name, cwd=cwd, env=env)
+    if not isinstance(data, dict):
+        return ""
+    return str(data.get(vm_key, {}).get(field, ""))