AGENTS.md

@@ -88,7 +88,9 @@ src/devx/
 │   ├── integration_guard.py # Run pytest with cross-runner fail-fast
 │   ├── check_translations.py  # Translation completeness check
 │   ├── doc_coverage.py  # Documentation coverage check
-│   └── lint_docs.py  # Documentation linter (structure, links, headings, code blocks, orphans)
+│   ├── lint_docs.py  # Documentation linter (structure, links, headings, code blocks, orphans)
+│   ├── validate_deploy_ref.py  # Validate git tag for deployments (--github-output)
+│   └── record_deployed_tag.py  # Record deployed tag to Gitea repo variable
 ├── tools/               # Developer tooling modules (run locally or by CI)
 │   ├── setup.py         # Environment setup (venv, deps, hooks)
 │   ├── install_tools.py # Install actionlint, git-cliff, act_runner, tea, hadolint, vale
@@ -113,6 +115,16 @@ src/devx/
 │   ├── pre_push_check.py    # Validate Vikunja task existence before push
 │   └── _shared.py           # Shared tool utilities
 ├── opentofu.py          # OpenTofu output helpers (get_tofu_output, get_tofu_vm_ip, get_tofu_vm_field)
+├── utils/               # Shared utilities (reusable across projects)
+│   ├── api.py            # API response helpers (is_truthy, is_falsy)
+│   ├── ssh.py            # SSH exec + wait_for_ssh (pure-Python socket check)
+│   ├── crypto.py         # Secret generation (shell-safe passwords)
+│   ├── vault.py          # Ansible vault encrypt/decrypt helpers
+│   ├── network.py        # HTTP connectivity check + wait_for_ssh
+│   ├── confirm.py        # Typed confirmation validation for destructive ops
+│   ├── json_registry.py  # File-locked JSON registry for local state
+│   ├── step_tracker.py   # Multi-step operation tracking with reports
+│   └── logging.py        # XDG-compliant logging configuration
 └── molecule/            # Optional molecule testing helpers (for Ansible projects)
     ├── discover_runners.py    # Dynamic Gitea runner discovery
     ├── distribute_molecule.py # Distribute molecule scenarios across runners (LPT scheduling, --roles-root for multi-role)

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

tests/unit/test_record_deployed_tag.py

@@ -0,0 +1,62 @@
+"""Unit tests for devx.ci.record_deployed_tag."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from click.testing import CliRunner
+
+from devx.ci.record_deployed_tag import main
+
+
+class TestRecordDeployedTag:
+    @patch("devx.ci.record_deployed_tag.GiteaClient")
+    @patch("devx.ci.record_deployed_tag.get_ci_token")
+    def test_records_production_tag(self, mock_token: MagicMock, mock_client: MagicMock) -> None:
+        mock_token.return_value = "fake-token"
+        client_instance = MagicMock()
+        mock_client.return_value = client_instance
+
+        runner = CliRunner()
+        result = runner.invoke(main, ["--env", "production", "--tag", "v1.0.0"])
+
+        assert result.exit_code == 0
+        assert "PRODUCTION_DEPLOY_TAG" in result.output
+        assert "v1.0.0" in result.output
+        client_instance.set_repo_variable.assert_called_once_with("PRODUCTION_DEPLOY_TAG", "v1.0.0")
+
+    @patch("devx.ci.record_deployed_tag.GiteaClient")
+    @patch("devx.ci.record_deployed_tag.get_ci_token")
+    def test_records_staging_tag(self, mock_token: MagicMock, mock_client: MagicMock) -> None:
+        mock_token.return_value = "fake-token"
+        client_instance = MagicMock()
+        mock_client.return_value = client_instance
+
+        runner = CliRunner()
+        result = runner.invoke(main, ["--env", "staging", "--tag", "master-abc123"])
+
+        assert result.exit_code == 0
+        assert "STAGING_DEPLOY_TAG" in result.output
+        client_instance.set_repo_variable.assert_called_once_with("STAGING_DEPLOY_TAG", "master-abc123")
+
+    @patch("devx.ci.record_deployed_tag.get_ci_token")
+    def test_token_error_exits_nonzero(self, mock_token: MagicMock) -> None:
+        import click
+
+        mock_token.side_effect = click.ClickException("No token available")
+
+        runner = CliRunner()
+        result = runner.invoke(main, ["--env", "production", "--tag", "v1.0.0"])
+
+        assert result.exit_code == 1
+        assert "No token available" in result.output
+
+    def test_invalid_env_choice(self) -> None:
+        runner = CliRunner()
+        result = runner.invoke(main, ["--env", "invalid", "--tag", "v1.0.0"])
+        assert result.exit_code != 0
+
+    def test_missing_tag_option(self) -> None:
+        runner = CliRunner()
+        result = runner.invoke(main, ["--env", "production"])
+        assert result.exit_code != 0

tests/unit/test_utils_confirm.py

@@ -0,0 +1,28 @@
+"""Unit tests for devx.utils.confirm."""
+
+from __future__ import annotations
+
+from devx.utils.confirm import validate_confirmation
+
+
+class TestValidateConfirmation:
+    def test_exact_match(self) -> None:
+        assert validate_confirmation("deploy-production", "deploy-production") is True
+
+    def test_mismatch(self) -> None:
+        assert validate_confirmation("deploy-staging", "deploy-production") is False
+
+    def test_empty_string(self) -> None:
+        assert validate_confirmation("", "deploy-production") is False
+
+    def test_case_sensitive(self) -> None:
+        assert validate_confirmation("Deploy-Production", "deploy-production") is False
+
+    def test_partial_match(self) -> None:
+        assert validate_confirmation("deploy", "deploy-production") is False
+
+    def test_extra_whitespace(self) -> None:
+        assert validate_confirmation("deploy-production ", "deploy-production") is False
+
+    def test_custom_expected(self) -> None:
+        assert validate_confirmation("yes-delete-all", "yes-delete-all") is True

tests/unit/test_utils_crypto.py

@@ -0,0 +1,69 @@
+"""Unit tests for devx.utils.crypto."""
+
+from __future__ import annotations
+
+import re
+
+from devx.utils.crypto import (
+    _DIGITS,
+    _LOWER,
+    _SYMBOLS,
+    _UPPER,
+    generate_hex_secret,
+    generate_password,
+    generate_secret,
+)
+
+
+class TestGenerateSecret:
+    def test_returns_url_safe_string(self) -> None:
+        secret = generate_secret()
+        assert isinstance(secret, str)
+        assert len(secret) > 0
+        # URL-safe base64 characters only
+        assert re.match(r"^[A-Za-z0-9_-]+$", secret)
+
+    def test_never_starts_with_dash(self) -> None:
+        for _ in range(1000):
+            secret = generate_secret()
+            assert not secret.startswith("-")
+
+
+class TestGeneratePassword:
+    def test_default_length(self) -> None:
+        pw = generate_password()
+        assert len(pw) == 32
+
+    def test_custom_length(self) -> None:
+        pw = generate_password(length=64)
+        assert len(pw) == 64
+
+    def test_contains_all_char_classes(self) -> None:
+        pw = generate_password(length=32)
+        assert any(c in _UPPER for c in pw), "Missing uppercase"
+        assert any(c in _LOWER for c in pw), "Missing lowercase"
+        assert any(c in _DIGITS for c in pw), "Missing digits"
+        assert any(c in _SYMBOLS for c in pw), "Missing symbols"
+
+    def test_first_char_alphanumeric(self) -> None:
+        for _ in range(1000):
+            pw = generate_password()
+            assert pw[0] not in _SYMBOLS, f"First char '{pw[0]}' is a symbol"
+
+    def test_minimum_length_4(self) -> None:
+        pw = generate_password(length=4)
+        assert len(pw) == 4
+
+
+class TestGenerateHexSecret:
+    def test_returns_hex_string(self) -> None:
+        secret = generate_hex_secret(length=32)
+        assert re.match(r"^[0-9a-f]+$", secret)
+
+    def test_correct_length(self) -> None:
+        secret = generate_hex_secret(length=20)
+        assert len(secret) == 20
+
+    def test_empty_for_zero(self) -> None:
+        secret = generate_hex_secret(length=0)
+        assert secret == ""

tests/unit/test_utils_json_registry.py

@@ -0,0 +1,110 @@
+"""Unit tests for devx.utils.json_registry."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from devx.utils.json_registry import JsonRegistry
+
+
+class TestJsonRegistry:
+    def test_add_and_get(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "state.json")
+        reg.add("item1", host="10.0.0.1", user="deploy")
+        info = reg.get("item1")
+        assert info is not None
+        assert info["host"] == "10.0.0.1"
+        assert info["user"] == "deploy"
+        assert "created_at" in info
+
+    def test_get_nonexistent(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "state.json")
+        assert reg.get("nope") is None
+
+    def test_remove(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "state.json")
+        reg.add("item1", host="10.0.0.1")
+        reg.remove("item1")
+        assert reg.get("item1") is None
+
+    def test_remove_nonexistent_is_noop(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "state.json")
+        reg.remove("nonexistent")  # should not raise
+
+    def test_list(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "state.json")
+        reg.add("a", host="1.1.1.1")
+        reg.add("b", host="2.2.2.2")
+        items = reg.list()
+        assert set(items.keys()) == {"a", "b"}
+        assert items["a"]["host"] == "1.1.1.1"
+
+    def test_list_empty(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "state.json")
+        assert reg.list() == {}
+
+    def test_update_existing(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "state.json")
+        reg.add("item", host="1.1.1.1", status="active")
+        reg.update("item", status="inactive")
+        info = reg.get("item")
+        assert info["status"] == "inactive"
+        assert info["host"] == "1.1.1.1"  # unchanged
+
+    def test_update_nonexistent_raises(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "state.json")
+        with pytest.raises(KeyError):
+            reg.update("nonexistent", host="1.1.1.1")
+
+    def test_update_skips_none_values(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "state.json")
+        reg.add("item", host="1.1.1.1")
+        reg.update("item", host=None, status="active")
+        info = reg.get("item")
+        assert info["host"] == "1.1.1.1"  # not overwritten by None
+        assert info["status"] == "active"
+
+    def test_persistence_across_instances(self, tmp_path: Path) -> None:
+        path = tmp_path / "state.json"
+        reg1 = JsonRegistry(path)
+        reg1.add("item", host="10.0.0.1")
+        reg2 = JsonRegistry(path)
+        info = reg2.get("item")
+        assert info is not None
+        assert info["host"] == "10.0.0.1"
+
+    def test_overwrite_existing(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "state.json")
+        reg.add("item", host="1.1.1.1")
+        reg.add("item", host="2.2.2.2")
+        info = reg.get("item")
+        assert info["host"] == "2.2.2.2"
+
+    def test_corrupt_json_returns_empty(self, tmp_path: Path) -> None:
+        path = tmp_path / "state.json"
+        path.write_text("{invalid json")
+        reg = JsonRegistry(path)
+        assert reg.list() == {}
+
+    def test_nonexistent_file_returns_empty(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "nonexistent.json")
+        assert reg.list() == {}
+
+    def test_creates_parent_dirs(self, tmp_path: Path) -> None:
+        path = tmp_path / "subdir" / "deeper" / "state.json"
+        reg = JsonRegistry(path)
+        reg.add("item", host="1.1.1.1")
+        assert path.exists()
+
+    def test_get_returns_copy(self, tmp_path: Path) -> None:
+        reg = JsonRegistry(tmp_path / "state.json")
+        reg.add("item", host="1.1.1.1", tags=["a", "b"])
+        info = reg.get("item")
+        assert info is not None
+        info["tags"].append("c")
+        # Original should be unchanged
+        info2 = reg.get("item")
+        assert info2 is not None
+        assert info2["tags"] == ["a", "b"]

tests/unit/test_utils_logging.py

@@ -0,0 +1,53 @@
+"""Unit tests for devx.utils.logging."""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from unittest.mock import patch
+
+from devx.utils.logging import get_logger
+
+
+class TestGetLogger:
+    def test_returns_logger_with_handlers(self) -> None:
+        logger = get_logger("test_devx_unit_1")
+        assert logger.handlers
+        assert isinstance(logger.handlers[0], logging.FileHandler)
+
+    def test_idempotent(self) -> None:
+        logger1 = get_logger("test_devx_unit_2")
+        initial_count = len(logger1.handlers)
+        logger2 = get_logger("test_devx_unit_2")
+        assert logger1 is logger2
+        assert len(logger2.handlers) == initial_count
+
+    def test_log_level_is_debug(self) -> None:
+        logger = get_logger("test_devx_unit_3")
+        assert logger.level == logging.DEBUG
+
+    def test_file_handler_level_is_debug(self) -> None:
+        logger = get_logger("test_devx_unit_4")
+        file_handler = logger.handlers[0]
+        assert file_handler.level == logging.DEBUG
+
+    def test_default_name(self) -> None:
+        logger = get_logger()
+        assert logger.name == "devx"
+
+    def test_creates_log_directory(self, tmp_path: Path) -> None:
+        with patch.object(Path, "home", return_value=tmp_path):
+            get_logger("test_app_creates_dir")
+            log_dir = tmp_path / ".local" / "state" / "test_app_creates_dir" / "logs"
+            assert log_dir.exists()
+            assert (log_dir / "test_app_creates_dir.log").exists()
+
+    def test_formatter_includes_timestamp(self) -> None:
+        logger = get_logger("test_devx_unit_5")
+        file_handler = logger.handlers[0]
+        fmt = file_handler.formatter
+        assert fmt is not None
+        assert "%(asctime)s" in fmt._fmt
+        assert "%(levelname)s" in fmt._fmt
+        assert "%(name)s" in fmt._fmt
+        assert "%(message)s" in fmt._fmt

tests/unit/test_utils_network.py

@@ -0,0 +1,71 @@
+"""Unit tests for devx.utils.network."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+import requests
+
+from devx.utils.network import check_http_connectivity, wait_for_ssh
+
+_no_sleep = MagicMock()
+
+
+class TestCheckHttpConnectivity:
+    @patch("devx.utils.network.requests.get")
+    def test_success(self, mock_get: MagicMock) -> None:
+        mock_get.return_value = MagicMock(status_code=200)
+        check_http_connectivity("https://example.com", max_attempts=3)
+        mock_get.assert_called_once()
+
+    @patch("devx.utils.network.requests.get")
+    def test_retries_on_connection_error(self, mock_get: MagicMock) -> None:
+        mock_get.side_effect = [
+            requests.exceptions.ConnectionError("refused"),
+            requests.exceptions.ConnectionError("refused"),
+            MagicMock(status_code=200),
+        ]
+        check_http_connectivity("https://example.com", max_attempts=5, sleep=_no_sleep)
+        assert mock_get.call_count == 3
+
+    @patch("devx.utils.network.requests.get")
+    def test_raises_after_max_attempts(self, mock_get: MagicMock) -> None:
+        mock_get.side_effect = requests.exceptions.ConnectionError("refused")
+        with pytest.raises(requests.exceptions.ConnectionError):
+            check_http_connectivity("https://example.com", max_attempts=2, sleep=_no_sleep)
+        assert mock_get.call_count == 2
+
+    @patch("devx.utils.network.requests.get")
+    def test_verify_false(self, mock_get: MagicMock) -> None:
+        mock_get.return_value = MagicMock(status_code=200)
+        check_http_connectivity("https://example.com", verify=False)
+        mock_get.assert_called_once_with("https://example.com", timeout=10, verify=False)
+
+
+class TestWaitForSsh:
+    @patch("devx.utils.network.socket.create_connection")
+    def test_immediate_success(self, mock_conn: MagicMock) -> None:
+        mock_conn.return_value.__enter__ = MagicMock()
+        mock_conn.return_value.__exit__ = MagicMock(return_value=False)
+        wait_for_ssh("10.0.0.1")
+        mock_conn.assert_called_once()
+
+    @patch("devx.utils.network.socket.create_connection")
+    def test_retries_until_success(self, mock_conn: MagicMock) -> None:
+        mock_conn.side_effect = [
+            OSError("refused"),
+            OSError("refused"),
+            MagicMock(),
+        ]
+        mock_conn.return_value.__enter__ = MagicMock()
+        mock_conn.return_value.__exit__ = MagicMock(return_value=False)
+        wait_for_ssh("10.0.0.1", max_attempts=5, sleep=_no_sleep)
+        assert mock_conn.call_count == 3
+
+    @patch("devx.utils.network.socket.create_connection")
+    def test_timeout_after_max_attempts(self, mock_conn: MagicMock) -> None:
+        mock_conn.side_effect = OSError("refused")
+        with pytest.raises(RuntimeError, match="SSH not available"):
+            wait_for_ssh("10.0.0.1", max_attempts=3, sleep=_no_sleep)
+        assert mock_conn.call_count == 3

tests/unit/test_utils_ssh.py

@@ -0,0 +1,96 @@
+"""Unit tests for devx.utils.ssh."""
+
+from __future__ import annotations
+
+import subprocess
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from devx.utils.ssh import docker_exec_on_vm, ssh_exec, wait_for_ssh
+
+
+class TestSshExec:
+    @patch("devx.utils.ssh.subprocess.run")
+    def test_success(self, mock_run: MagicMock) -> None:
+        mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
+        result = ssh_exec("10.0.0.1", "uname -a")
+        assert result.returncode == 0
+        mock_run.assert_called_once()
+
+    @patch("devx.utils.ssh.subprocess.run")
+    def test_failure_with_check(self, mock_run: MagicMock) -> None:
+        mock_result = MagicMock(returncode=1, stdout="", stderr="error")
+        mock_result.check_returncode.side_effect = subprocess.CalledProcessError(1, "ssh")
+        mock_run.return_value = mock_result
+        with pytest.raises(subprocess.CalledProcessError):
+            ssh_exec("10.0.0.1", "false")
+
+    @patch("devx.utils.ssh.subprocess.run")
+    def test_failure_without_check(self, mock_run: MagicMock) -> None:
+        mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="error")
+        result = ssh_exec("10.0.0.1", "false", check=False)
+        assert result.returncode == 1
+
+    @patch("devx.utils.ssh.subprocess.run")
+    def test_custom_user(self, mock_run: MagicMock) -> None:
+        mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
+        ssh_exec("10.0.0.1", "whoami", user="root")
+        cmd = mock_run.call_args[0][0]
+        assert "root@10.0.0.1" in cmd
+
+
+class TestDockerExecOnVm:
+    @patch("devx.utils.ssh.ssh_exec")
+    def test_simple_command(self, mock_ssh: MagicMock) -> None:
+        mock_ssh.return_value = MagicMock(stdout="output\n")
+        result = docker_exec_on_vm("10.0.0.1", "mycontainer", "ls /")
+        assert result == "output"
+        mock_ssh.assert_called_once_with("10.0.0.1", "docker exec mycontainer ls /", user="deploy", timeout=30)
+
+    @patch("devx.utils.ssh.ssh_exec")
+    def test_psql_mode(self, mock_ssh: MagicMock) -> None:
+        mock_ssh.return_value = MagicMock(stdout="result\n")
+        result = docker_exec_on_vm("10.0.0.1", "db", "SELECT 1", db_user="postgres", db_name="mydb")
+        assert result == "result"
+        call_args = mock_ssh.call_args[0][1]
+        assert "psql -U postgres -d mydb" in call_args
+        assert "SELECT 1" in call_args
+
+    @patch("devx.utils.ssh.ssh_exec")
+    def test_psql_escapes_single_quotes(self, mock_ssh: MagicMock) -> None:
+        mock_ssh.return_value = MagicMock(stdout="\n")
+        docker_exec_on_vm("10.0.0.1", "db", "SELECT 'it''s ok'", db_user="pg", db_name="db")
+        call_args = mock_ssh.call_args[0][1]
+        assert "'\"'\"'" in call_args
+
+
+class TestWaitForSsh:
+    @patch("devx.utils.ssh.socket.create_connection")
+    def test_immediate_success(self, mock_conn: MagicMock) -> None:
+        mock_conn.return_value.__enter__ = MagicMock()
+        mock_conn.return_value.__exit__ = MagicMock(return_value=False)
+        wait_for_ssh("10.0.0.1")
+        mock_conn.assert_called_once()
+
+    @patch("devx.utils.ssh.socket.create_connection")
+    @patch("devx.utils.ssh.time.sleep")
+    def test_retries_until_success(self, mock_sleep: MagicMock, mock_conn: MagicMock) -> None:
+        # Fail twice, then succeed
+        mock_conn.side_effect = [
+            OSError("refused"),
+            OSError("refused"),
+            MagicMock(),
+        ]
+        mock_conn.return_value.__enter__ = MagicMock()
+        mock_conn.return_value.__exit__ = MagicMock(return_value=False)
+        wait_for_ssh("10.0.0.1", max_attempts=5)
+        assert mock_conn.call_count == 3
+
+    @patch("devx.utils.ssh.socket.create_connection")
+    @patch("devx.utils.ssh.time.sleep")
+    def test_timeout_after_max_attempts(self, mock_sleep: MagicMock, mock_conn: MagicMock) -> None:
+        mock_conn.side_effect = OSError("refused")
+        with pytest.raises(RuntimeError, match="SSH not available"):
+            wait_for_ssh("10.0.0.1", max_attempts=3)
+        assert mock_conn.call_count == 3

tests/unit/test_utils_step_tracker.py

@@ -0,0 +1,124 @@
+"""Unit tests for devx.utils.step_tracker."""
+
+from __future__ import annotations
+
+import click
+import pytest
+from click.testing import CliRunner
+
+from devx.utils.step_tracker import Step, StepTracker, track_steps
+
+
+class TestStep:
+    def test_initial_status_is_pending(self) -> None:
+        step = Step("install")
+        assert step.status == "pending"
+        assert step.name == "install"
+
+
+class TestStepTracker:
+    def test_begin_adds_step_as_in_progress(self) -> None:
+        tracker = StepTracker()
+        tracker.begin("install deps")
+        assert len(tracker.steps) == 1
+        assert tracker.steps[0].status == "in_progress"
+
+    def test_done_marks_last_in_progress_as_completed(self) -> None:
+        tracker = StepTracker()
+        tracker.begin("step1")
+        tracker.done()
+        assert tracker.steps[0].status == "completed"
+
+    def test_done_no_op_if_no_in_progress(self) -> None:
+        tracker = StepTracker()
+        tracker.begin("step1")
+        tracker.done()
+        tracker.done()  # should not raise, no-op
+        assert tracker.steps[0].status == "completed"
+
+    def test_done_no_op_if_empty(self) -> None:
+        tracker = StepTracker()
+        tracker.done()  # should not raise
+
+    def test_multiple_steps(self) -> None:
+        tracker = StepTracker()
+        tracker.begin("step1")
+        tracker.done()
+        tracker.begin("step2")
+        tracker.done()
+        assert len(tracker.steps) == 2
+        assert tracker.steps[0].status == "completed"
+        assert tracker.steps[1].status == "completed"
+
+
+class TestTrackSteps:
+    def test_successful_operation(self) -> None:
+        runner = CliRunner()
+        with runner.isolation():
+            with track_steps() as tracker:
+                tracker.begin("step1")
+                tracker.done()
+                tracker.begin("step2")
+                tracker.done()
+            assert len(tracker.steps) == 2
+            assert all(s.status == "completed" for s in tracker.steps)
+
+    def test_exception_marks_in_progress_as_failed(self) -> None:
+        runner = CliRunner()
+        with runner.isolation():
+            with pytest.raises(ValueError, match="boom"):
+                with track_steps() as tracker:
+                    tracker.begin("step1")
+                    tracker.done()
+                    tracker.begin("step2")
+                    raise ValueError("boom")
+            assert tracker.steps[0].status == "completed"
+            assert tracker.steps[1].status == "failed"
+
+    def test_pending_step_stays_pending_on_exception(self) -> None:
+        runner = CliRunner()
+        with runner.isolation():
+            with pytest.raises(ValueError):
+                with track_steps() as tracker:
+                    tracker.begin("step1")
+                    tracker.done()
+                    tracker.begin("step2")
+                    tracker.done()
+                    tracker.begin("step3")  # in_progress
+                    # step4 is pending (not started)
+                    raise ValueError("oops")
+            assert tracker.steps[2].status == "failed"
+
+    def test_empty_operation(self) -> None:
+        runner = CliRunner()
+        with runner.isolation():
+            with track_steps() as tracker:
+                pass
+            assert tracker.steps == []
+
+    def test_report_printed_on_success(self) -> None:
+        runner = CliRunner()
+        result = runner.invoke(_cmd_success, [], color=False)
+        assert result.exit_code == 0
+        assert "Operation Report" in result.output
+        assert "step1" in result.output
+
+    def test_report_printed_on_failure(self) -> None:
+        runner = CliRunner()
+        result = runner.invoke(_cmd_failure, [], color=False)
+        assert result.exit_code != 0
+        assert "Operation Report" in result.output
+
+
+@click.command()
+def _cmd_success() -> None:
+    with track_steps() as tracker:
+        tracker.begin("step1")
+        tracker.done()
+
+
+@click.command()
+def _cmd_failure() -> None:
+    with track_steps() as tracker:
+        tracker.begin("step1")
+        raise ValueError("oops")

tests/unit/test_utils_vault.py

@@ -0,0 +1,134 @@
+"""Unit tests for devx.utils.vault."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from devx.utils.vault import (
+    decrypt_file,
+    encrypt_file,
+    is_encrypted,
+    load_vault_yaml,
+    save_vault_yaml,
+)
+
+
+class TestIsEncrypted:
+    def test_encrypted_file(self, tmp_path: Path) -> None:
+        f = tmp_path / "secret.yml"
+        f.write_text("$ANSIBLE_VAULT;1.1;AES256\n9382928...\n")
+        assert is_encrypted(f) is True
+
+    def test_plain_file(self, tmp_path: Path) -> None:
+        f = tmp_path / "plain.yml"
+        f.write_text("key: value\n")
+        assert is_encrypted(f) is False
+
+
+class TestLoadVaultYaml:
+    def test_plain_yaml_no_vault_pass(self, tmp_path: Path) -> None:
+        f = tmp_path / "data.yml"
+        f.write_text("key: value\nlist:\n  - a\n  - b\n")
+        data = load_vault_yaml(f)
+        assert data == {"key": "value", "list": ["a", "b"]}
+
+    def test_empty_file(self, tmp_path: Path) -> None:
+        f = tmp_path / "empty.yml"
+        f.write_text("")
+        data = load_vault_yaml(f)
+        assert data == {}
+
+    def test_vault_pass_not_exists(self, tmp_path: Path) -> None:
+        f = tmp_path / "data.yml"
+        f.write_text("key: value\n")
+        data = load_vault_yaml(f, vault_pass=tmp_path / "nonexistent")
+        assert data == {"key": "value"}
+
+    @patch("devx.utils.vault.subprocess.run")
+    def test_encrypted_file_success(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        f = tmp_path / "secret.yml"
+        f.write_text("$ANSIBLE_VAULT\n...")
+        vp = tmp_path / "vault-password"
+        vp.write_text("secret")
+
+        mock_run.return_value = MagicMock(returncode=0, stdout="key: decrypted\n", stderr="")
+        data = load_vault_yaml(f, vault_pass=vp)
+        assert data == {"key": "decrypted"}
+
+    @patch("devx.utils.vault.subprocess.run")
+    def test_not_vault_encrypted_fallback(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        f = tmp_path / "plain.yml"
+        f.write_text("key: value\n")
+        vp = tmp_path / "vault-password"
+        vp.write_text("secret")
+
+        mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="is not vault encrypted")
+        data = load_vault_yaml(f, vault_pass=vp)
+        assert data == {"key": "value"}
+
+
+class TestSaveVaultYaml:
+    def test_save_plain(self, tmp_path: Path) -> None:
+        f = tmp_path / "output.yml"
+        save_vault_yaml(f, {"key": "value"})
+        content = f.read_text()
+        assert "key: value" in content
+
+    def test_save_with_vault_pass_not_exists(self, tmp_path: Path) -> None:
+        f = tmp_path / "output.yml"
+        vp = tmp_path / "nonexistent"
+        save_vault_yaml(f, {"key": "value"}, vault_pass=vp)
+        # Should save as plain YAML
+        content = f.read_text()
+        assert "key: value" in content
+        assert "$ANSIBLE_VAULT" not in content
+
+    @patch("devx.utils.vault.subprocess.run")
+    def test_save_and_encrypt(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        f = tmp_path / "output.yml"
+        vp = tmp_path / "vault-password"
+        vp.write_text("secret")
+
+        save_vault_yaml(f, {"key": "value"}, vault_pass=vp)
+        # File should be written
+        assert f.exists()
+        # ansible-vault encrypt should be called
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        assert "ansible-vault" in cmd
+        assert "encrypt" in cmd
+
+
+class TestEncryptFile:
+    @patch("devx.utils.vault.subprocess.run")
+    def test_calls_ansible_vault(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        f = tmp_path / "file.yml"
+        f.write_text("key: value")
+        vp = tmp_path / "vault-password"
+        vp.write_text("secret")
+
+        encrypt_file(f, vp)
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        assert "ansible-vault" in cmd
+        assert "encrypt" in cmd
+        assert str(f) in cmd
+        assert str(vp) in cmd
+
+
+class TestDecryptFile:
+    @patch("devx.utils.vault.subprocess.run")
+    def test_calls_ansible_vault(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        f = tmp_path / "file.yml"
+        f.write_text("$ANSIBLE_VAULT\n...")
+        vp = tmp_path / "vault-password"
+        vp.write_text("secret")
+
+        decrypt_file(f, vp)
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        assert "ansible-vault" in cmd
+        assert "decrypt" in cmd
+        assert str(f) in cmd
+        assert str(vp) in cmd

tests/unit/test_validate_deploy_ref.py

@@ -0,0 +1,70 @@
+"""Unit tests for devx.ci.validate_deploy_ref."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from click.testing import CliRunner
+
+from devx.ci.validate_deploy_ref import main
+
+
+class TestValidateDeployRef:
+    def test_valid_tag_prints_ref(self, tmp_path: Path) -> None:
+        runner = CliRunner()
+        with patch("devx.ci.validate_deploy_ref.subprocess.run") as mock_run:
+            mock_run.return_value = MagicMock(returncode=0, stdout="abcdef1234567890\n", stderr="")
+            result = runner.invoke(main, ["--tag", "v1.0.0"])
+        assert result.exit_code == 0
+        assert "v1.0.0" in result.output
+
+    def test_invalid_tag_exits_nonzero(self) -> None:
+        runner = CliRunner()
+        with patch("devx.ci.validate_deploy_ref.subprocess.run") as mock_run:
+            mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="error")
+            result = runner.invoke(main, ["--tag", "nonexistent"])
+        assert result.exit_code == 1
+        assert "does not exist" in result.output
+
+    def test_no_tag_without_allow_empty_exits_nonzero(self) -> None:
+        runner = CliRunner()
+        result = runner.invoke(main, [])
+        assert result.exit_code == 1
+        assert "No tag specified" in result.output
+
+    def test_allow_empty_prints_pr_mode(self) -> None:
+        runner = CliRunner()
+        result = runner.invoke(main, ["--allow-empty"])
+        assert result.exit_code == 0
+        assert "PR mode" in result.output
+
+    def test_github_output_writes_ref(self, tmp_path: Path) -> None:
+        runner = CliRunner()
+        gh_output = tmp_path / "github_output"
+        gh_output.write_text("")
+        with patch("devx.ci.validate_deploy_ref.subprocess.run") as mock_run:
+            mock_run.return_value = MagicMock(returncode=0, stdout="abcdef12\n", stderr="")
+            with runner.isolation(env={"GITHUB_OUTPUT": str(gh_output)}):
+                result = runner.invoke(main, ["--tag", "v1.0.0", "--github-output"])
+        assert result.exit_code == 0
+        content = gh_output.read_text()
+        assert "deploy-ref=v1.0.0" in content
+
+    def test_github_output_without_env_var_exits_nonzero(self) -> None:
+        runner = CliRunner()
+        with patch("devx.ci.validate_deploy_ref.subprocess.run") as mock_run:
+            mock_run.return_value = MagicMock(returncode=0, stdout="abcdef12\n", stderr="")
+            with runner.isolation(env={"GITHUB_OUTPUT": ""}):
+                result = runner.invoke(main, ["--tag", "v1.0.0", "--github-output"])
+        assert result.exit_code == 1
+        assert "GITHUB_OUTPUT" in result.output
+
+    def test_allow_empty_with_github_output(self, tmp_path: Path) -> None:
+        runner = CliRunner()
+        gh_output = tmp_path / "github_output"
+        gh_output.write_text("")
+        with runner.isolation(env={"GITHUB_OUTPUT": str(gh_output)}):
+            result = runner.invoke(main, ["--allow-empty", "--github-output"])
+        assert result.exit_code == 0
+        assert "deploy-ref=" in gh_output.read_text()