DEVX-124: feat: extract shared utilities from infra and grm into devx

src/devx/ci/record_deployed_tag.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+"""Record the deployed git tag for a given environment.
+
+Writes the tag to a Gitea repository variable so it can be queried
+later via the Gitea API or ``devx.ci.get_deployed_tag``.
+
+Usage::
+
+    python -m devx.ci.record_deployed_tag --env production --tag v0.28.1
+    python -m devx.ci.record_deployed_tag --env staging --tag master-abc1234
+"""
+
+from __future__ import annotations
+
+import sys
+
+import click
+
+from devx.api_clients import GiteaClient
+from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
+from devx.i18n import _
+from devx.tokens import get_ci_token
+
+
+@click.command()
+@click.option(
+    "--env",
+    "env_name",
+    type=click.Choice(["staging", "production"]),
+    required=True,
+)
+@click.option("--tag", required=True, help=_("Git tag or ref that was deployed"))
+def main(env_name: str, tag: str) -> None:
+    """Record the deployed tag for the given environment."""
+    try:
+        token = get_ci_token()
+    except click.ClickException as exc:
+        click.echo(f"Error: {exc.message}", err=True)
+        sys.exit(1)
+
+    var_name = f"{env_name.upper()}_DEPLOY_TAG"
+    client = GiteaClient(GITEA_API_URL, token, REPO_OWNER, REPO_NAME)
+    client.set_repo_variable(var_name, tag)
+    click.echo(f"Recorded {var_name} = {tag}")
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

src/devx/ci/validate_deploy_ref.py

@@ -0,0 +1,85 @@
+#!/usr/bin/env python3
+"""Resolve and validate the git tag to deploy.
+
+Shared between staging and production deployments.  Ensures a concrete
+git tag is used — never a moving branch ref — so deployments are
+reproducible and rollback-friendly.
+
+Usage in workflows::
+
+    # Production (tag required)
+    python -m devx.ci.validate_deploy_ref --tag "$TAG" --github-output
+
+    # Staging force-deploy (tag required)
+    python -m devx.ci.validate_deploy_ref --tag "$TAG" --github-output
+
+    # Staging PR-triggered (PR SHA is already concrete, no tag needed)
+    python -m devx.ci.validate_deploy_ref --allow-empty --github-output
+
+Writes ``deploy-ref=<tag>`` to ``$GITHUB_OUTPUT`` when ``--github-output``
+is passed, otherwise prints the ref to stdout.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess  # nosec B404
+import sys
+
+import click
+
+from devx.i18n import _
+
+
+@click.command()
+@click.option("--tag", default="", help=_("Git tag to deploy (e.g. v0.28.1)."))
+@click.option(
+    "--allow-empty",
+    is_flag=True,
+    help=_("Allow empty tag (PR mode where SHA is concrete)."),
+)
+@click.option(
+    "--github-output",
+    is_flag=True,
+    help=_("Write deploy-ref to $GITHUB_OUTPUT file."),
+)
+def main(tag: str, allow_empty: bool, github_output: bool) -> None:
+    """Resolve and validate the deploy ref, exiting non-zero on failure."""
+    if not tag:
+        if not allow_empty:
+            click.echo(
+                "::error::No tag specified. Deployments require a concrete git tag "
+                "(e.g. v0.28.1). Use --allow-empty only for PR-triggered staging deploys "
+                "where the checkout SHA is already concrete.",
+                err=True,
+            )
+            sys.exit(1)
+        ref = ""
+        click.echo("No tag specified — using checkout ref (PR mode).")
+    else:
+        result = subprocess.run(  # nosec B603, B607
+            ["git", "rev-parse", "-q", "--verify", f"refs/tags/{tag}"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        if result.returncode != 0:
+            click.echo(f"::error::Tag '{tag}' does not exist in the repository.", err=True)
+            sys.exit(1)
+        ref = tag
+        commit = result.stdout.strip()[:8]
+        click.echo(f"Deploying tag: {tag} (commit {commit})")
+
+    if github_output:
+        github_output_path = os.environ.get("GITHUB_OUTPUT")
+        if not github_output_path:
+            click.echo("::error::GITHUB_OUTPUT environment variable not set.", err=True)
+            sys.exit(1)
+        with open(github_output_path, "a") as f:
+            f.write(f"deploy-ref={ref}\n")
+    else:
+        click.echo(ref)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

src/devx/translations.json

@@ -3566,5 +3566,37 @@
     "pl": "{separator}",
     "ru": "{separator}",
     "zh": "{separator}"
+  },
+  "Allow empty tag (PR mode where SHA is concrete).": {
+    "bg": "Allow empty tag (PR mode where SHA is concrete).",
+    "de": "Allow empty tag (PR mode where SHA is concrete).",
+    "en": "Allow empty tag (PR mode where SHA is concrete).",
+    "pl": "Allow empty tag (PR mode where SHA is concrete).",
+    "ru": "Allow empty tag (PR mode where SHA is concrete).",
+    "zh": "Allow empty tag (PR mode where SHA is concrete)."
+  },
+  "Git tag or ref that was deployed": {
+    "bg": "Git tag or ref that was deployed",
+    "de": "Git tag or ref that was deployed",
+    "en": "Git tag or ref that was deployed",
+    "pl": "Git tag or ref that was deployed",
+    "ru": "Git tag or ref that was deployed",
+    "zh": "Git tag or ref that was deployed"
+  },
+  "Git tag to deploy (e.g. v0.28.1).": {
+    "bg": "Git tag to deploy (e.g. v0.28.1).",
+    "de": "Git tag to deploy (e.g. v0.28.1).",
+    "en": "Git tag to deploy (e.g. v0.28.1).",
+    "pl": "Git tag to deploy (e.g. v0.28.1).",
+    "ru": "Git tag to deploy (e.g. v0.28.1).",
+    "zh": "Git tag to deploy (e.g. v0.28.1)."
+  },
+  "Write deploy-ref to $GITHUB_OUTPUT file.": {
+    "bg": "Write deploy-ref to $GITHUB_OUTPUT file.",
+    "de": "Write deploy-ref to $GITHUB_OUTPUT file.",
+    "en": "Write deploy-ref to $GITHUB_OUTPUT file.",
+    "pl": "Write deploy-ref to $GITHUB_OUTPUT file.",
+    "ru": "Write deploy-ref to $GITHUB_OUTPUT file.",
+    "zh": "Write deploy-ref to $GITHUB_OUTPUT file."
   }
 }

src/devx/utils/confirm.py

@@ -0,0 +1,27 @@
+"""Typed confirmation validation for destructive operations.
+
+Ensures the user typed an exact confirmation phrase before proceeding
+with dangerous operations (e.g. production deploys, database migrations).
+
+Usage::
+
+    from devx.utils.confirm import validate_confirmation
+
+    if not validate_confirmation(user_input, expected="deploy-production"):
+        raise SystemExit("Confirmation does not match")
+"""
+
+from __future__ import annotations
+
+
+def validate_confirmation(confirm: str, expected: str) -> bool:
+    """Check if confirmation text matches the expected phrase.
+
+    Args:
+        confirm: The confirmation text entered by the user.
+        expected: The exact phrase that must be matched.
+
+    Returns:
+        True if confirmation matches exactly, False otherwise.
+    """
+    return confirm == expected

src/devx/utils/crypto.py

@@ -0,0 +1,73 @@
+"""Cryptographic secret generation helpers.
+
+Provides safe secret/password generators that avoid shell-option
+interpretation issues (e.g. leading ``-`` being parsed as a flag by
+``su -c`` in Docker entrypoints).
+
+Usage::
+
+    from devx.utils.crypto import generate_secret, generate_password
+
+    api_key = generate_secret()
+    db_password = generate_password(length=32)
+"""
+
+from __future__ import annotations
+
+import secrets
+
+_SYMBOLS = "!@#$%^&*()-_=+[]{}|;:,.<>?"
+_UPPER = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+_LOWER = "abcdefghijklmnopqrstuvwxyz"
+_DIGITS = "0123456789"
+
+
+def generate_secret() -> str:
+    """Generate a URL-safe secret that never starts with ``-``.
+
+    A leading ``-`` causes passwords to be interpreted as command-line
+    options when passed through shell expansion chains (e.g. Nextcloud's
+    Docker entrypoint uses ``su -c`` which strips quoting).
+
+    Returns:
+        A 43-character URL-safe base64 secret.
+    """
+    value = secrets.token_urlsafe(32)
+    while value.startswith("-"):
+        value = secrets.token_urlsafe(32)
+    return value
+
+
+def generate_password(length: int = 32) -> str:
+    """Generate a password guaranteed to contain upper, lower, digit, and symbol.
+
+    The first character is always alphanumeric to avoid being interpreted
+    as a command-line option when passed through shell expansion chains.
+
+    Args:
+        length: Desired password length (minimum 4).
+
+    Returns:
+        A password string with guaranteed character class coverage.
+    """
+    pools = [_UPPER, _LOWER, _DIGITS, _SYMBOLS]
+    chars = [secrets.choice(p) for p in pools]
+    all_chars = "".join(pools)
+    chars += [secrets.choice(all_chars) for _ in range(length - len(pools))]
+    secrets.SystemRandom().shuffle(chars)
+    while chars[0] in _SYMBOLS:
+        secrets.SystemRandom().shuffle(chars)
+    return "".join(chars)
+
+
+def generate_hex_secret(length: int = 32) -> str:
+    """Generate a hexadecimal secret of the given length.
+
+    Args:
+        length: Desired number of hex characters (doubled internally
+            since ``token_hex`` produces pairs).
+
+    Returns:
+        A hexadecimal string.
+    """
+    return secrets.token_hex(length // 2)

src/devx/utils/json_registry.py

@@ -0,0 +1,128 @@
+"""File-locked JSON registry for local state management.
+
+Provides a simple JSON-backed key-value store with ``fcntl`` file
+locking for safe concurrent access.  Useful for CLI tools that need
+to track remote resources (runners, VMs, deployments) on the local
+machine.
+
+Usage::
+
+    from devx.utils.json_registry import JsonRegistry
+
+    registry = JsonRegistry(Path("~/.local/share/myapp/state.json"))
+    registry.add("item1", host="10.0.0.1", user="deploy")
+    info = registry.get("item1")
+    registry.remove("item1")
+"""
+
+from __future__ import annotations
+
+import copy
+import fcntl
+import json
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, cast
+
+
+class JsonRegistry:
+    """Manages a local JSON file mapping names to arbitrary metadata.
+
+    Uses ``fcntl`` for file locking (shared lock for reads, exclusive
+    lock for writes) to prevent race conditions in concurrent scenarios.
+    """
+
+    def __init__(self, path: Path | None = None) -> None:
+        """Initialise the registry.
+
+        Args:
+            path: Path to the JSON file. Defaults to
+                ``~/.local/share/devx/registry.json``.
+        """
+        self._path = path or Path.home() / ".local" / "share" / "devx" / "registry.json"
+        self._data: dict[str, dict[str, Any]] = self._load()
+
+    def _load(self) -> dict[str, dict[str, Any]]:
+        if not self._path.exists():
+            return {}
+        try:
+            with open(self._path) as f:
+                fcntl.flock(f.fileno(), fcntl.LOCK_SH)
+                try:
+                    data: Any = json.load(f)
+                    if isinstance(data, dict):
+                        return cast(dict[str, dict[str, Any]], data)
+                finally:
+                    fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+        except (json.JSONDecodeError, OSError):
+            pass
+        return {}
+
+    def _save(self) -> None:
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        with open(self._path, "w") as f:
+            fcntl.flock(f.fileno(), fcntl.LOCK_EX)
+            try:
+                json.dump(self._data, f, indent=2)
+            finally:
+                fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+
+    def add(self, name: str, **fields: Any) -> None:
+        """Register or overwrite an entry in the registry.
+
+        Args:
+            name: Unique key for the entry.
+            **fields: Arbitrary metadata fields to store.
+        """
+        self._data[name] = {
+            **fields,
+            "created_at": datetime.now(UTC).isoformat(),
+        }
+        self._save()
+
+    def get(self, name: str) -> dict[str, Any] | None:
+        """Retrieve entry metadata by name.
+
+        Args:
+            name: Key to look up.
+
+        Returns:
+            A copy of the entry's metadata, or None if not found.
+        """
+        info = self._data.get(name)
+        if info:
+            return copy.deepcopy(info)
+        return None
+
+    def remove(self, name: str) -> None:
+        """Remove an entry from the registry.
+
+        Args:
+            name: Key to remove. No-op if not found.
+        """
+        if name in self._data:
+            del self._data[name]
+            self._save()
+
+    def list(self) -> dict[str, dict[str, Any]]:
+        """Return a copy of all registered entries.
+
+        Returns:
+            Dict mapping names to metadata copies.
+        """
+        return {name: copy.deepcopy(info) for name, info in self._data.items()}
+
+    def update(self, name: str, **fields: Any) -> None:
+        """Update fields for an existing entry.
+
+        Args:
+            name: Key to update.
+            **fields: Fields to update (None values are skipped).
+
+        Raises:
+            KeyError: If the entry doesn't exist.
+        """
+        if name not in self._data:
+            raise KeyError(name)
+        self._data[name].update({k: v for k, v in fields.items() if v is not None})
+        self._save()

src/devx/utils/logging.py

@@ -0,0 +1,48 @@
+"""XDG-compliant logging configuration for CLI tools.
+
+Provides a standardised logging setup that writes to
+``~/.local/state/<app>/logs/<app>.log`` following the XDG state
+directory specification.  Console output is handled separately by
+the application (e.g. via ``click.echo``).
+
+Usage::
+
+    from devx.utils.logging import get_logger
+
+    logger = get_logger("myapp")
+    logger.info("Application started")
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+
+def get_logger(name: str = "devx") -> logging.Logger:
+    """Return a configured logger that writes to an XDG state directory.
+
+    All messages (including DEBUG) are written to
+    ``~/.local/state/<name>/logs/<name>.log``.  Console output is
+    expected to be handled by the application via ``click.echo``.
+
+    Args:
+        name: Logger name and subdirectory name for log files.
+
+    Returns:
+        A configured :class:`logging.Logger` instance.
+    """
+    logger = logging.getLogger(name)
+    if logger.handlers:
+        return logger
+
+    logger.setLevel(logging.DEBUG)
+
+    log_dir = Path.home() / ".local" / "state" / name / "logs"
+    log_dir.mkdir(parents=True, exist_ok=True)
+    file_handler = logging.FileHandler(log_dir / f"{name}.log")
+    file_handler.setLevel(logging.DEBUG)
+    file_handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
+    logger.addHandler(file_handler)
+
+    return logger

src/devx/utils/network.py

@@ -0,0 +1,102 @@
+"""Network connectivity helpers.
+
+Provides retry-aware HTTP connectivity checks and SSH availability
+checks for deployment workflows.  Uses ``tenacity`` for exponential
+backoff retry logic.
+
+Usage::
+
+    from devx.utils.network import check_http_connectivity, wait_for_ssh
+
+    check_http_connectivity("https://auth.example.com")
+    wait_for_ssh("178.105.254.83")
+"""
+
+from __future__ import annotations
+
+import logging
+import socket
+import time
+from collections.abc import Callable
+
+import requests
+from tenacity import (
+    Retrying,
+    before_sleep_log,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_exponential,
+)
+
+
+def check_http_connectivity(
+    base_url: str,
+    max_attempts: int = 30,
+    *,
+    verify: bool = True,
+    sleep: Callable[[float], None] | None = None,
+) -> None:
+    """Verify HTTP reachability of *base_url* with retry.
+
+    Uses tenacity for retry with exponential backoff (2 s min, 10 s max).
+
+    Args:
+        base_url: URL to check via GET request.
+        max_attempts: Maximum retry attempts.
+        verify: Whether to verify TLS certificates.
+        sleep: Custom sleep function for testing (defaults to ``time.sleep``).
+
+    Raises:
+        requests.exceptions.ConnectionError: If the URL is not reachable
+            after *max_attempts*.
+    """
+    retrying = Retrying(
+        stop=stop_after_attempt(max_attempts),
+        wait=wait_exponential(multiplier=2, min=2, max=10),
+        retry=retry_if_exception_type(requests.exceptions.ConnectionError),
+        before_sleep=before_sleep_log(logging.getLogger("devx.utils.network"), logging.WARNING),
+        sleep=sleep if sleep is not None else time.sleep,
+        reraise=True,
+    )
+
+    def _check() -> None:
+        requests.get(base_url, timeout=10, verify=verify)  # nosec B501
+
+    retrying(_check)
+
+
+def wait_for_ssh(
+    host: str,
+    port: int = 22,
+    max_attempts: int = 30,
+    interval: int = 10,
+    *,
+    sleep: Callable[[float], None] | None = None,
+) -> None:
+    """Wait for SSH to be available on a host using a pure-Python socket check.
+
+    Uses socket instead of ``nc(1)`` so it works on CI runners without
+    netcat.  Uses exponential backoff: starts at 2 s, doubles each
+    attempt up to 10 s max.
+
+    Args:
+        host: VM IP address or hostname.
+        port: SSH port (default 22).
+        max_attempts: Maximum number of connection attempts.
+        interval: Base interval for backoff calculation (seconds).
+        sleep: Custom sleep function for testing (defaults to ``time.sleep``).
+
+    Raises:
+        RuntimeError: If SSH is not available after *max_attempts*.
+    """
+    _sleep = sleep if sleep is not None else time.sleep
+    for i in range(max_attempts):
+        try:
+            with socket.create_connection((host, port), timeout=5):
+                return
+        except OSError:
+            pass
+        if i < max_attempts - 1:
+            wait = min(2 * (2**i), 10)
+            _sleep(wait)
+    raise RuntimeError(f"SSH not available on {host}:{port} after {max_attempts} attempts")

src/devx/utils/ssh.py

@@ -0,0 +1,132 @@
+"""SSH helpers for running commands on remote hosts.
+
+Provides a simple wrapper around the ``ssh`` CLI for executing commands
+on remote machines (e.g. customer VMs, CI runners) without requiring
+Ansible.  Includes a pure-Python ``wait_for_ssh`` that uses socket
+instead of ``nc(1)`` so it works on minimal CI containers.
+
+Usage::
+
+    from devx.utils.ssh import ssh_exec, wait_for_ssh
+
+    wait_for_ssh("178.105.254.83")
+    result = ssh_exec("178.105.254.83", "uname -a")
+    print(result.stdout)
+"""
+
+from __future__ import annotations
+
+import socket
+import subprocess  # nosec B404
+import sys
+import time
+
+SSH_CONNECT_TIMEOUT = "10"
+SSH_HOST_KEY_CHECKING = "no"
+
+
+def ssh_exec(
+    host: str,
+    command: str,
+    *,
+    user: str = "deploy",
+    timeout: int = 30,
+    check: bool = True,
+) -> subprocess.CompletedProcess[str]:
+    """Run *command* on *host* via SSH and return the result.
+
+    Args:
+        host: VM IP address or hostname.
+        command: Shell command to execute on the remote host.
+        user: SSH user (default ``deploy``).
+        timeout: Subprocess timeout in seconds.
+        check: If True, raise ``CalledProcessError`` on non-zero exit.
+
+    Returns:
+        The completed process result with stdout/stderr captured.
+    """
+    result = subprocess.run(  # nosec B603, B607, B607
+        [
+            "ssh",
+            "-o",
+            f"StrictHostKeyChecking={SSH_HOST_KEY_CHECKING}",
+            "-o",
+            f"ConnectTimeout={SSH_CONNECT_TIMEOUT}",
+            f"{user}@{host}",
+            command,
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=timeout,
+    )
+    if check and result.returncode != 0:
+        print(f"SSH command failed on {host}: {command}", file=sys.stderr)
+        print(f"  stdout: {result.stdout.strip()}", file=sys.stderr)
+        print(f"  stderr: {result.stderr.strip()}", file=sys.stderr)
+        result.check_returncode()
+    return result
+
+
+def docker_exec_on_vm(
+    host: str,
+    container: str,
+    command: str,
+    *,
+    user: str = "deploy",
+    db_user: str | None = None,
+    db_name: str | None = None,
+    timeout: int = 30,
+) -> str:
+    """Run a command inside a Docker container on a remote VM via SSH.
+
+    For PostgreSQL commands, set *db_user* and *db_name* to run
+    ``psql -U <db_user> -d <db_name> -c <command>`` inside the container.
+
+    Args:
+        host: VM IP address or hostname.
+        container: Docker container name on the remote host.
+        command: Command to execute inside the container (or SQL if db_user/db_name set).
+        user: SSH user (default ``deploy``).
+        db_user: PostgreSQL user name (enables psql mode).
+        db_name: PostgreSQL database name (enables psql mode).
+        timeout: Subprocess timeout in seconds.
+
+    Returns:
+        Stripped stdout from the command.
+    """
+    if db_user and db_name:
+        escaped_sql = command.replace("'", "'\"'\"'")
+        remote_cmd = f'docker exec {container} psql -U {db_user} -d {db_name} -t -A -c "{escaped_sql}"'
+    else:
+        remote_cmd = f"docker exec {container} {command}"
+    result = ssh_exec(host, remote_cmd, user=user, timeout=timeout)
+    return result.stdout.strip()
+
+
+def wait_for_ssh(host: str, port: int = 22, max_attempts: int = 30, interval: int = 10) -> None:
+    """Wait for SSH to be available on a host using a pure-Python socket check.
+
+    Uses socket instead of ``nc(1)`` so it works on CI runners without
+    netcat.  Uses exponential backoff: starts at 2 s, doubles each
+    attempt up to 10 s max.
+
+    Args:
+        host: VM IP address or hostname.
+        port: SSH port (default 22).
+        max_attempts: Maximum number of connection attempts.
+        interval: Base interval for backoff calculation (seconds).
+
+    Raises:
+        RuntimeError: If SSH is not available after *max_attempts*.
+    """
+    for i in range(max_attempts):
+        try:
+            with socket.create_connection((host, port), timeout=5):
+                return
+        except OSError:
+            pass
+        if i < max_attempts - 1:
+            wait = min(2 * (2**i), 10)
+            time.sleep(wait)
+    raise RuntimeError(f"SSH not available on {host}:{port} after {max_attempts} attempts")

src/devx/utils/step_tracker.py

@@ -0,0 +1,102 @@
+"""Operation step tracking with translated reports.
+
+Provides a context manager that tracks multi-step operations and prints
+a status report on exit.  Steps are marked as pending, in_progress,
+completed, or failed.  On exception, the last in-progress step is
+marked as failed.
+
+Usage::
+
+    from devx.utils.step_tracker import track_steps
+
+    with track_steps() as tracker:
+        tracker.begin("Install dependencies")
+        install_deps()
+        tracker.done()
+
+        tracker.begin("Run tests")
+        run_tests()
+        tracker.done()
+"""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+from contextlib import contextmanager
+
+import click
+
+_STATUS_ICONS = {
+    "completed": "✓",
+    "failed": "✗",
+    "pending": "○",
+    "in_progress": "◌",
+}
+
+_STATUS_COLORS = {
+    "completed": "green",
+    "failed": "red",
+    "in_progress": "yellow",
+    "pending": "white",
+}
+
+
+class Step:
+    """A single tracked step in an operation."""
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self.status = "pending"
+
+
+class StepTracker:
+    """Tracks steps of an operation and prints a report on exit."""
+
+    def __init__(self) -> None:
+        self.steps: list[Step] = []
+
+    def begin(self, name: str) -> None:
+        """Start a new step.
+
+        Args:
+            name: Human-readable step name.
+        """
+        step = Step(name)
+        self.steps.append(step)
+        step.status = "in_progress"
+
+    def done(self) -> None:
+        """Mark the most recent in-progress step as completed."""
+        if self.steps and self.steps[-1].status == "in_progress":
+            self.steps[-1].status = "completed"
+
+
+@contextmanager
+def track_steps() -> Generator[StepTracker, None, None]:
+    """Context manager that tracks steps and prints a report on exit.
+
+    On exception the last in-progress step is marked as failed.
+    The report is printed in the ``finally`` block so it always appears.
+
+    Yields:
+        A :class:`StepTracker` instance to track steps with.
+    """
+    tracker = StepTracker()
+    try:
+        yield tracker
+    except Exception:
+        for step in reversed(tracker.steps):
+            if step.status == "in_progress":
+                step.status = "failed"
+        raise
+    finally:
+        _print_report(tracker.steps)
+
+
+def _print_report(steps: list[Step]) -> None:
+    """Print an operation report to stdout."""
+    click.secho("=== Operation Report ===", fg="bright_cyan")
+    for step in steps:
+        icon = _STATUS_ICONS.get(step.status, "?")
+        color = _STATUS_COLORS.get(step.status)
+        click.secho(f"  {icon} {step.name} ({step.status})", fg=color)

src/devx/utils/vault.py

@@ -0,0 +1,135 @@
+"""Ansible Vault helpers for encrypting and decrypting YAML files.
+
+Wraps ``ansible-vault`` to provide a convenient API for loading and
+saving vault-encrypted YAML files.  Falls back to plain YAML when no
+vault-password file is available, making it safe to use in both
+local (with vault) and CI (without vault) environments.
+
+Usage::
+
+    from devx.utils.vault import load_vault_yaml, save_vault_yaml
+
+    data = load_vault_yaml(Path("secrets.yml"), vault_pass=Path("vault-password"))
+    data["new_key"] = "value"
+    save_vault_yaml(Path("secrets.yml"), data, vault_pass=Path("vault-password"))
+"""
+
+from __future__ import annotations
+
+import subprocess  # nosec B404
+from pathlib import Path
+
+import yaml
+
+
+def encrypt_file(path: Path, vault_pass: Path) -> None:
+    """Encrypt a file in-place using ansible-vault.
+
+    Args:
+        path: File to encrypt.
+        vault_pass: Path to the vault-password file.
+    """
+    subprocess.run(  # nosec B603, B607
+        [
+            "ansible-vault",
+            "encrypt",
+            str(path),
+            "--vault-password-file",
+            str(vault_pass),
+            "--encrypt-vault-id",
+            "default",
+        ],
+        check=True,
+    )
+
+
+def decrypt_file(path: Path, vault_pass: Path) -> None:
+    """Decrypt a file in-place using ansible-vault.
+
+    Args:
+        path: File to decrypt.
+        vault_pass: Path to the vault-password file.
+    """
+    subprocess.run(  # nosec B603, B607
+        [
+            "ansible-vault",
+            "decrypt",
+            str(path),
+            "--vault-password-file",
+            str(vault_pass),
+        ],
+        check=True,
+    )
+
+
+def load_vault_yaml(path: Path, vault_pass: Path | None = None) -> dict:
+    """Load a YAML file, decrypting with ansible-vault if vault-password exists.
+
+    If *vault_pass* is None or doesn't exist, the file is read as plain
+    YAML.  If decryption fails (file not vault-encrypted), it falls back
+    to plain YAML.
+
+    Args:
+        path: YAML file path.
+        vault_pass: Path to the vault-password file (optional).
+
+    Returns:
+        Parsed YAML content as a dict (empty dict if file is empty).
+    """
+    if vault_pass is None or not vault_pass.exists():
+        with open(path, encoding="utf-8") as f:
+            return yaml.safe_load(f) or {}
+    result = subprocess.run(  # nosec B603, B607
+        ["ansible-vault", "view", str(path), "--vault-password-file", str(vault_pass)],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode == 0:
+        return yaml.safe_load(result.stdout) or {}
+    if "is not vault encrypted" in result.stderr:
+        with open(path, encoding="utf-8") as f:
+            return yaml.safe_load(f) or {}
+    result.check_returncode()  # pragma: no cover
+    return {}  # pragma: no cover
+
+
+def save_vault_yaml(path: Path, data: dict, vault_pass: Path | None = None) -> None:
+    """Write YAML data, encrypting with ansible-vault if vault-password exists.
+
+    Args:
+        path: Destination YAML file path.
+        data: Data to serialize.
+        vault_pass: Path to the vault-password file (optional).
+    """
+    plain = yaml.dump(data, default_flow_style=False, sort_keys=False)
+    with open(path, "w", encoding="utf-8") as f:
+        f.write(plain)
+    if vault_pass is not None and vault_pass.exists():
+        subprocess.run(  # nosec B603, B607
+            [
+                "ansible-vault",
+                "encrypt",
+                str(path),
+                "--vault-password-file",
+                str(vault_pass),
+                "--encrypt-vault-id",
+                "default",
+            ],
+            capture_output=True,
+            check=True,
+        )
+
+
+def is_encrypted(path: Path) -> bool:
+    """Check if a file is ansible-vault encrypted.
+
+    Args:
+        path: File to check.
+
+    Returns:
+        True if the file starts with the ``$ANSIBLE_VAULT`` marker.
+    """
+    with open(path, encoding="utf-8") as f:
+        first_line = f.readline()
+    return "$ANSIBLE_VAULT" in first_line