release: v0.40.0 [skip ci]

.env.example

@@ -1,6 +1,19 @@
-# Gitea API token (required for CI scripts that interact with Gitea)
+# Role-based Gitea API tokens.
+# Each token serves a specific role. For small teams the developer and CI
+# tokens may belong to the same user, but the reviewer token MUST belong to a
+# different Gitea user than the PR author so Gitea accepts approval reviews.
 # Create at: https://git.oblachno.oblachno.fyi/user/settings/applications
-CI_GITEA_TOKEN=
+
+# Developer token — used by local tooling: create-task, create-pr, setup, etc.
+DEVELOPER_GITEA_API_TOKEN=
+
+# CI token — used by CI workflows and scripts that do not post approvals.
+# Legacy CI_GITEA_TOKEN is also accepted.
+CI_GITEA_API_TOKEN=
+
+# Reviewer token — used by the auto-merge workflow to post APPROVE reviews.
+# This must be a different Gitea user from the developer/CI user.
+REVIEWER_GITEA_API_TOKEN=
 
 # Vikunja API token (required for post-merge task updates)
 # Create at: https://work.oblachno.oblachno.fyi/settings/tokens

.gitea/workflows/build-images.yml

@@ -38,6 +38,8 @@ jobs:
         with:
           fetch-depth: 1
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-ci
       - name: Check if this is a release commit
         id: check
@@ -62,18 +64,22 @@ jobs:
           fetch-depth: 0
       - name: Set up environment
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-release
       - name: Docker registry login
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           CI_GITEA_USERNAME: ${{ vars.CI_GITEA_USERNAME }}
         run: |
           . .venv/bin/activate
-          echo "$CI_GITEA_TOKEN" | docker login git.oblachno.oblachno.fyi -u "$CI_GITEA_USERNAME" --password-stdin
+          _TOKEN="$CI_GITEA_API_TOKEN"
+          [ -z "$_TOKEN" ] && _TOKEN="$DEVELOPER_GITEA_API_TOKEN"
+          [ -z "$_TOKEN" ] && _TOKEN="$CI_GITEA_TOKEN"
+          if [ -z "$_TOKEN" ]; then echo "Gitea API token not set — skipping Docker login"; exit 1; fi
+          echo "$_TOKEN" | docker login git.oblachno.oblachno.fyi -u "$CI_GITEA_USERNAME" --password-stdin
       - name: Build and push tier images
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           CI_GITEA_USERNAME: ${{ vars.CI_GITEA_USERNAME }}
           PYTHONPATH: src
         run: |
@@ -103,7 +109,7 @@ jobs:
       - name: Notify on failure
         if: failure()
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           . .venv/bin/activate 2>/dev/null || true
@@ -125,10 +131,12 @@ jobs:
         with:
           fetch-depth: 1
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-ci
       - name: Clean up old image versions
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           . .venv/bin/activate

.gitea/workflows/ci.yml

@@ -16,6 +16,8 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Lint all
         run: |
@@ -79,6 +81,8 @@ jobs:
         with:
           fetch-depth: 0
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Detect changed paths
         id: detect
@@ -106,10 +110,11 @@ jobs:
           fetch-depth: 0
       - name: Set up environment
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Release dry-run validation
         env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           . .venv/bin/activate 2>/dev/null || true
@@ -127,10 +132,12 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Run automated PR review
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           set -euo pipefail
@@ -160,12 +167,14 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          token: ${{ secrets.CI_GITEA_TOKEN }}
+          token: ${{ secrets.CI_GITEA_API_TOKEN }}
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Post approval review
         env:
-          CI_GITEA_TOKEN: ${{ secrets.REVIEW_GITEA_TOKEN }}
+          REVIEWER_GITEA_API_TOKEN: ${{ secrets.REVIEWER_GITEA_API_TOKEN }}
           PR_NUMBER: ${{ github.event.number }}
           REPOSITORY: ${{ github.repository }}
           PYTHONPATH: src
@@ -180,7 +189,7 @@ jobs:
             --body "Auto-approved: all CI checks passed (quality, pr-review, release-dry-run)."
       - name: Squash merge with task ID
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           VIKUNJA_TOKEN: ${{ secrets.VIKUNJA_TOKEN }}
           DEVX_VIKUNJA_PROJECT_ID: "8"
           PYTHONPATH: src

.gitea/workflows/post-merge.yml

@@ -47,6 +47,8 @@ jobs:
         with:
           fetch-depth: 1
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Check if this is a release commit
         id: check
@@ -70,6 +72,8 @@ jobs:
         with:
           fetch-depth: 1
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Validate latest commit message
         env:
@@ -95,10 +99,10 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          token: ${{ secrets.CI_GITEA_TOKEN }}
+          token: ${{ secrets.CI_GITEA_API_TOKEN }}
       - name: Set up environment
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Configure git
         run: |
@@ -107,6 +111,7 @@ jobs:
       - name: Run release
         id: release-tag
         env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           . .venv/bin/activate 2>/dev/null || true
@@ -115,7 +120,7 @@ jobs:
       - name: Notify on failure
         if: failure()
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           . .venv/bin/activate 2>/dev/null || true
@@ -142,10 +147,12 @@ jobs:
           fetch-depth: 0
           ref: ${{ needs.release.outputs.tag }}
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image EXTRAS=release
       - name: Build and publish release
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           . .venv/bin/activate 2>/dev/null || true
@@ -154,7 +161,7 @@ jobs:
       - name: Notify on failure
         if: failure()
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           . .venv/bin/activate 2>/dev/null || true
@@ -183,10 +190,12 @@ jobs:
         with:
           fetch-depth: 0
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Sync documentation to wiki
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           . .venv/bin/activate 2>/dev/null || true
@@ -194,7 +203,7 @@ jobs:
       - name: Notify on failure
         if: failure()
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           export PATH="$HOME/.local/bin:$PATH"
@@ -219,15 +228,18 @@ jobs:
         with:
           fetch-depth: 0
           ref: master
-          token: ${{ secrets.CI_GITEA_TOKEN }}
+          token: ${{ secrets.CI_GITEA_API_TOKEN }}
       - name: Fetch latest master
         run: |
           git fetch origin master
           git reset --hard origin/master
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Generate and push badges
         env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PRE_COMMIT_ALLOW_NO_CONFIG: "1"
         run: |
           . .venv/bin/activate 2>/dev/null || true
@@ -235,7 +247,7 @@ jobs:
       - name: Notify on failure
         if: failure()
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           export PATH="$HOME/.local/bin:$PATH"
@@ -260,6 +272,8 @@ jobs:
         with:
           fetch-depth: 0
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Update Vikunja task
         env:
@@ -272,7 +286,7 @@ jobs:
       - name: Notify on failure
         if: failure()
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           export PATH="$HOME/.local/bin:$PATH"
@@ -295,10 +309,12 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - name: Set up environment
+        env:
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
         run: make setup-image
       - name: Ensure branch protection and labels
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
           DEVX_REPO_NAME: devx
           DEVX_REPO_OWNER: oblachno-oss
@@ -308,7 +324,7 @@ jobs:
       - name: Notify on failure
         if: failure()
         env:
-          CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
+          CI_GITEA_API_TOKEN: ${{ secrets.CI_GITEA_API_TOKEN }}
           PYTHONPATH: src
         run: |
           export PATH="$HOME/.local/bin:$PATH"

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)
@@ -148,6 +160,12 @@ The following rules are enforced for `master`:
 ### 1. Create Vikunja Task
 Create a task in Vikunja to get a `DEVX-N` identifier.
 
+**IMPORTANT:** The task title must NOT include the `DEVX-N:` prefix.
+The `make create-pr` and `check_auto_merge_ready` commands automatically
+prepend `DEVX-N: ` to the Vikunja task title when forming the PR title.
+If the Vikunja task title already includes the prefix, the PR title will
+have a double prefix and auto-merge validation will fail.
+
 ### 2. Create Branch
 ```bash
 git checkout master && git pull

CHANGELOG.md

@@ -2,6 +2,24 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.40.0] - 2026-07-11
+
+### Features
+
+- Detect double-prefix in Vikunja task title during pre-merge validation
+
+## [0.39.0] - 2026-07-09
+
+### Features
+
+- Extract shared utilities from infra and grm into devx
+
+## [0.38.0] - 2026-07-08
+
+### Features
+
+- Introduce role-based Gitea API token environment variables
+
 ## [0.37.0] - 2026-07-07
 
 ### Features

README.md

@@ -16,12 +16,12 @@ quality badges.
 
 [![CI](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions/workflows/ci.yml/badge.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
 [![License: GPL-3.0](https://img.shields.io/badge/license-GPL--3.0-blue)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE)
-[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
-[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
-[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
+[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
-[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
-[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
+[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
-[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/python.svg)](https://www.python.org/downloads/)
+[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/python.svg)](https://www.python.org/downloads/)
 
 ## Why devx?
 
@@ -87,7 +87,7 @@ extra index and list devx in your dependencies:
 ```toml
 [project]
 dependencies = [
-    "devx>=0.37.0",
+    "devx>=0.40.0",
 ]
 
 [tool.pip]
@@ -101,8 +101,8 @@ pip install -e .
 ```
 
 > **Note:** If your project requires a specific devx version, pin it in
-> `dependencies` (for example, `"devx==0.37.0"`) or use a version constraint
+> `dependencies` (for example, `"devx==0.40.0"`) or use a version constraint
-> (for example, `"devx>=0.37.0,<0.38"`).
+> (for example, `"devx>=0.40.0,<0.41"`).
 
 ### Optional extras
 

docs/index.md

@@ -12,12 +12,12 @@ project to be reusable across all oblachno-oss repositories.
 
 [![CI](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions/workflows/ci.yml/badge.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
 [![License: GPL-3.0](https://img.shields.io/badge/license-GPL--3.0-blue)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE)
-[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
-[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
-[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
+[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
-[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
-[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
+[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
-[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/bff19e053993255b932b7b20cb91dd6a056bf409/python.svg)](https://www.python.org/downloads/)
+[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/c9f25c1348d9703783e473e54f9d624879667bbc/python.svg)](https://www.python.org/downloads/)
 
 ## Overview
 
@@ -74,14 +74,14 @@ Add devx to your `pyproject.toml` dependencies and configure the registry:
 ```toml
 [project]
 dependencies = [
-    "devx>=0.37.0",
+    "devx>=0.40.0",
 ]
 
 [tool.pip]
 extra-index-url = "https://git.oblachno.oblachno.fyi/api/packages/oblachno-oss/pypi/simple"
 ```
 
-Pin a specific version if needed: `"devx==0.37.0"` or `"devx>=0.37.0,<0.38"`.
+Pin a specific version if needed: `"devx==0.40.0"` or `"devx>=0.40.0,<0.41"`.
 
 ### Optional extras
 

docs/user/getting-started.md

@@ -48,12 +48,12 @@ Add devx to your `pyproject.toml`:
 ```toml
 [project]
 dependencies = [
-    "devx>=0.37.0",
+    "devx>=0.40.0",
 ]
 
 [project.optional-dependencies]
 dev = [
-    "devx>=0.37.0",
+    "devx>=0.40.0",
 ]
 ```
 

src/devx/__init__.py

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

src/devx/ci/auto_merge.py

@@ -17,10 +17,9 @@ This allows the PR title to be a human-friendly Vikunja task title
 while the squashed commit follows conventional commits.
 
 Usage:
-  CI_GITEA_TOKEN=<token> python3 -m devx.ci.auto_merge <branch> <pr_title> <repo> <pr_number>
+  CI_GITEA_API_TOKEN=<token> VIKUNJA_TOKEN=<token> python3 -m devx.ci.auto_merge <branch> <pr_title> <repo> <pr_number>
 """
 
-import os
 import re
 from pathlib import Path
 from typing import Any
@@ -40,6 +39,7 @@ from devx.config import (
 )
 from devx.exceptions import APIError
 from devx.i18n import _
+from devx.tokens import get_ci_token, get_vikunja_token
 
 # Strip leading task ID prefix (e.g. "DEVX-12: " or "OBL-INFRA-364: ") from commit subjects.
 _TASK_ID_PREFIX_RE = re.compile(rf"^{TASK_PREFIX}-\d+:\s*")
@@ -115,9 +115,12 @@ def get_vikunja_task_title(task_id: str) -> str:
 
     Raises ClickException if VIKUNJA_TOKEN is not set or the task is not found.
     """
-    token = os.environ.get("VIKUNJA_TOKEN", "")
+    try:
-    if not token:
+        token = get_vikunja_token()
-        raise click.ClickException(_("VIKUNJA_TOKEN is not set. This is required in CI to validate PR titles."))
+    except click.ClickException:
+        raise click.ClickException(
+            _("VIKUNJA_TOKEN is not set. This is required in CI to validate PR titles.")
+        ) from None
     client = VikunjaClient(VIKUNJA_API_URL, token)
     page = 1
     while True:
@@ -197,9 +200,10 @@ def extract_conventional_msg(commits: list[dict[str, Any]]) -> str:
 @click.argument("repo")
 @click.argument("pr_number")
 def main(branch: str, pr_title: str, repo: str, pr_number: str) -> None:
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_ci_token()
-        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set."))
+    except click.ClickException:
+        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set.")) from None
 
     # Validate PR number is an integer
     try:

src/devx/ci/check_auto_merge_ready.py

@@ -15,7 +15,7 @@ Exit code 1 = NOT ready — fix issues before pushing.
 
 Usage::
 
-  # CI (with VIKUNJA_TOKEN and CI_GITEA_TOKEN):
+  # CI (with VIKUNJA_TOKEN and CI_GITEA_API_TOKEN):
   python3 -m devx.ci.check_auto_merge_ready \\
       --branch "$HEAD_REF" \\
       --pr-title "$PR_TITLE" \\
@@ -34,13 +34,12 @@ skipped (with a warning) — this allows local pre-push hooks to run
 without CI secrets.  In CI, the token is always set and the check is
 mandatory.
 
-If ``CI_GITEA_TOKEN`` is not set and ``--pr-number`` is not provided, only
+If ``CI_GITEA_API_TOKEN`` is not set and ``--pr-number`` is not provided, only
 branch-name and PR-title-format checks run (local mode).
 """
 
 from __future__ import annotations
 
-import os
 import subprocess  # nosec B404
 
 import click
@@ -55,6 +54,7 @@ from devx.config import (
 )
 from devx.exceptions import APIError
 from devx.i18n import _
+from devx.tokens import get_ci_token, get_vikunja_token
 
 load_dotenv()
 
@@ -99,10 +99,13 @@ def is_branch_behind_master(branch: str) -> bool:
 def get_pr_title_from_gitea(repo: str, pr_number: int) -> str | None:
     """Fetch the PR title from the Gitea API.
 
-    Returns ``None`` if ``CI_GITEA_TOKEN`` is not set or the PR cannot be fetched.
+    Returns ``None`` if no token is set or the PR cannot be fetched.
     """
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token or "/" not in repo:
+        token = get_ci_token()
+    except click.ClickException:
+        return None
+    if "/" not in repo:
         return None
     owner, repo_name = repo.split("/", 1)
     client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
@@ -120,8 +123,9 @@ def get_vikunja_title_optional(task_id: str) -> str | None:
     raise when ``VIKUNJA_TOKEN`` is missing — it returns ``None`` so the
     caller can skip the check in local mode.
     """
-    token = os.environ.get("VIKUNJA_TOKEN", "")
+    try:
-    if not token:
+        token = get_vikunja_token()
+    except click.ClickException:
         return None
     client = VikunjaClient(VIKUNJA_API_URL, token)
     from devx.config import DEFAULT_PER_PAGE
@@ -221,7 +225,11 @@ def cli(
     if not skip_vikunja:
         vikunja_title = get_vikunja_title_optional(task_id)
         if vikunja_title is None:
-            token_set = bool(os.environ.get("VIKUNJA_TOKEN", ""))
+            try:
+                get_vikunja_token()
+                token_set = True
+            except click.ClickException:
+                token_set = False
             if token_set:
                 errors.append(
                     _(
@@ -233,17 +241,33 @@ def cli(
             else:
                 click.echo("[pre-merge-check] WARNING: VIKUNJA_TOKEN not set — skipping Vikunja title match check.")
         else:
-            expected = f"{task_id}: {vikunja_title}"
+            # Defensive check: warn if the Vikunja task title already includes
-            if pr_title != expected:
+            # the task ID prefix. The expected PR title is
+            # f"{task_id}: {vikunja_title}" — if vikunja_title already starts
+            # with "{task_id}:", the PR title will have a double prefix.
+            if vikunja_title.startswith(f"{task_id}:"):
                 errors.append(
                     _(
-                        "PR title does not match Vikunja task title.\n  Expected: {expected}\n  Got:      {title}",
+                        "Vikunja task title '{title}' starts with '{prefix}:'. "
-                        expected=expected,
+                        "The task title should NOT include the '{prefix}' prefix — "
-                        title=pr_title,
+                        "it is automatically added to the PR title. "
+                        "Update the Vikunja task title to remove the prefix.",
+                        title=vikunja_title,
+                        prefix=task_id,
                     ),
                 )
             else:
-                click.echo(f"[pre-merge-check] Vikunja title match OK: {expected}")
+                expected = f"{task_id}: {vikunja_title}"
+                if pr_title != expected:
+                    errors.append(
+                        _(
+                            "PR title does not match Vikunja task title.\n  Expected: {expected}\n  Got:      {title}",
+                            expected=expected,
+                            title=pr_title,
+                        ),
+                    )
+                else:
+                    click.echo(f"[pre-merge-check] Vikunja title match OK: {expected}")
 
     # 6. Branch behind master (skip if --skip-behind-check)
     if not skip_behind_check:

src/devx/ci/discover_runners.py

@@ -31,6 +31,7 @@ import requests
 
 from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
 from devx.i18n import _
+from devx.tokens import get_ci_token
 
 DEFAULT_MAX_RUNNERS = 3
 
@@ -96,7 +97,7 @@ def query_runners(api_url: str, token: str, owner: str, repo: str) -> int:
     return total
 
 
-def get_runner_count(api_url: str, token: str, owner: str, repo: str) -> int:
+def get_runner_count(api_url: str, token: str | None, owner: str, repo: str) -> int:
     """Determine the number of available runners.
 
     Tries the Gitea API first, then falls back to env vars, then default.
@@ -152,7 +153,10 @@ def main(
     output_indices: bool,
     github_output: bool,
 ) -> None:
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
+        token = get_ci_token()
+    except click.ClickException:
+        token = None
 
     if owner is None:
         owner = os.environ.get("DEVX_REPO_OWNER", "") or REPO_OWNER

src/devx/ci/integration_guard.py

@@ -17,7 +17,7 @@ Usage::
 
 Environment variables:
     GITEA_URL           Base URL of the Gitea instance.
-    CI_GITEA_TOKEN      API token with repo access.
+    CI_GITEA_API_TOKEN  API token with repo access (CI_GITEA_TOKEN accepted for legacy).
     RUN_ID              Workflow run ID (GITHUB_RUN_ID).
     JOB_NAME            Base job name (GITHUB_JOB), e.g. "integration-tests".
     MATRIX_INDEX        Current matrix index (runner-index).
@@ -41,6 +41,7 @@ from devx.i18n import _
 from devx.molecule.molecule_ci_guard import (
     poll_for_other_failures,
 )
+from devx.tokens import get_ci_token
 
 POLL_INTERVAL = 10
 
@@ -50,7 +51,10 @@ POLL_INTERVAL = 10
 def cli(pytest_args: tuple[str, ...]) -> None:
     """Run pytest with cross-runner failure detection."""
     gitea_url = os.environ.get("GITEA_URL", "")
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
+        token = get_ci_token()
+    except click.ClickException:
+        token = None
     run_id = int(os.environ.get("RUN_ID", "0"))
     job_name = os.environ.get("JOB_NAME", "integration-tests")
     current_index = int(os.environ.get("MATRIX_INDEX", "0"))

src/devx/ci/notify_failure.py

@@ -6,7 +6,7 @@ otherwise go unnoticed in the Actions tab. Uses the ``tea`` Gitea CLI
 for issue creation — tea must be installed and configured.
 
 Usage:
-  CI_GITEA_TOKEN=<token> python3 -m devx.ci.notify_failure \
+  CI_GITEA_API_TOKEN=<token> python3 -m devx.ci.notify_failure \
     --repo <owner/repo> \
     --run-id <run_id> \
     --workflow <workflow_name> \
@@ -14,14 +14,13 @@ Usage:
     --auto-login
 
 With ``--auto-login``, the script configures the tea CLI login profile
-from ``CI_GITEA_TOKEN`` and ``DEVX_GITEA_API_URL`` before creating the issue,
+from the CI API token and ``DEVX_GITEA_API_URL`` before creating the issue,
 eliminating the need for a separate ``tea login add`` step in the workflow.
 """
 
 from __future__ import annotations
 
 import logging
-import os
 
 import click
 from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnknownVariableType]
@@ -29,6 +28,7 @@ from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnk
 from devx.config import GITEA_API_URL
 from devx.gitea_cli import TeaCLI, TeaCLIError, configure_tea_login
 from devx.i18n import _
+from devx.tokens import get_ci_token
 
 load_dotenv()
 
@@ -74,9 +74,10 @@ def _create_issue_via_tea(repo: str, title: str, body: str) -> int:
     help="Configure tea CLI login from CI_GITEA_TOKEN before creating the issue.",
 )
 def main(repo: str, run_id: str, workflow: str, commit: str, auto_login: bool) -> None:
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        get_ci_token()
-        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set."))
+    except click.ClickException:
+        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set.")) from None
 
     if auto_login:
         configure_tea_login()

src/devx/ci/post_merge.py

@@ -5,7 +5,6 @@ Usage:
   VIKUNJA_TOKEN=<token> python3 -m devx.ci.post_merge <commit_msg> [--commit-sha <sha>]
 """
 
-import os
 import re
 import subprocess  # nosec B404
 
@@ -17,6 +16,7 @@ from devx.ci._shared import extract_task_id as _extract_task_id
 from devx.config import DEFAULT_PER_PAGE, TASK_PREFIX, VIKUNJA_API_URL, VIKUNJA_PROJECT_ID
 from devx.exceptions import APIError
 from devx.i18n import _
+from devx.tokens import get_vikunja_token
 
 load_dotenv()
 
@@ -127,9 +127,10 @@ def main(commit_msg: str | None, commit_sha: str, from_git: bool, git_sha: str)
             commit_sha = _get_git_commit_sha()
     if not commit_msg:
         raise click.ClickException("commit_msg argument is required (or use --from-git or --git-sha)")
-    token = os.environ.get("VIKUNJA_TOKEN", "")
+    try:
-    if not token:
+        token = get_vikunja_token()
-        raise click.ClickException(_("ERROR: VIKUNJA_TOKEN is not set."))
+    except click.ClickException:
+        raise click.ClickException(_("ERROR: VIKUNJA_TOKEN is not set.")) from None
 
     task_id = extract_task_id(commit_msg)
     if not task_id:

src/devx/ci/pr_review.py

@@ -17,12 +17,11 @@ Checks performed:
   8. Commit conventions — conventional commit format on branch commits
 
 Usage:
-  CI_GITEA_TOKEN=<token> python3 -m devx.ci.pr_review <pr_number> <owner/repo>
+  CI_GITEA_API_TOKEN=<token> [REVIEWER_GITEA_API_TOKEN=<token>] python3 -m devx.ci.pr_review <pr_number> <owner/repo>
 """
 
 from __future__ import annotations
 
-import os
 import re
 from dataclasses import dataclass, field
 from typing import Any
@@ -34,6 +33,7 @@ from devx.api_clients import GiteaClient
 from devx.config import GITEA_API_URL
 from devx.exceptions import APIError
 from devx.i18n import _
+from devx.tokens import get_ci_token, get_reviewer_token
 
 load_dotenv()
 
@@ -636,9 +636,10 @@ def main(
     Without --event: runs automated checks and posts COMMENT/REQUEST_CHANGES.
     With --event: posts a manual review (skips automated checks).
     """
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_reviewer_token() if (event and event.upper() == "APPROVE") else get_ci_token()
-        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set."))
+    except click.ClickException:
+        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set.")) from None
 
     owner, repo_name = repo.split("/")
     client = GiteaClient(GITEA_API_URL, token, owner, repo_name)

src/devx/ci/publish.py

@@ -9,14 +9,14 @@ Publishing destinations (checked in order):
      ``DEVX_PYPI_REGISTRY_URL`` env var is set, or ``GITEA_API_URL``
      is converted to a packages URL). Uses ``twine upload
      --repository-url <url> -u <token> -p <token>`` with the
-     ``CI_GITEA_TOKEN`` as both username and password.
+     CI API token as both username and password.
   2. **Standard PyPI** — if ``PYPI_TOKEN`` is set. Uses the standard
      ``twine upload -u __token__ -p <token>`` flow.
   3. **Skip** — if neither is configured, only the Gitea release is created.
 
 Usage:
-  CI_GITEA_TOKEN=<token> [PYPI_TOKEN=<token>] python3 -m devx.ci.publish <tag> <repo>
+  CI_GITEA_API_TOKEN=<token> [PYPI_TOKEN=<token>] python3 -m devx.ci.publish <tag> <repo>
-  CI_GITEA_TOKEN=<token> python3 -m devx.ci.publish <tag> <repo> --registry-url https://git.example.com/api/packages/owner/pypi
+  CI_GITEA_API_TOKEN=<token> python3 -m devx.ci.publish <tag> <repo> --registry-url https://git.example.com/api/packages/owner/pypi
 """
 
 import os
@@ -31,6 +31,7 @@ from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnk
 from devx.config import GITEA_API_URL, REPO_OWNER
 from devx.gitea_cli import TeaCLI, TeaCLIError, configure_tea_login
 from devx.i18n import _
+from devx.tokens import get_ci_token
 
 load_dotenv()
 
@@ -253,9 +254,10 @@ def main(
 
     if not tag:
         raise click.ClickException(_("Tag is required (or use --from-tag)."))
-    gitea_token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not gitea_token:
+        gitea_token = get_ci_token()
-        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set."))
+    except click.ClickException:
+        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set.")) from None
 
     pypi_token = os.environ.get("PYPI_TOKEN", "")
 

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/release.py

@@ -29,7 +29,7 @@ version.  This prevents duplicate release commits (a common issue when CI
 checkouts don't fetch tags) and ensures tag/version/commit alignment.
 
 Usage:
-  CI_GITEA_TOKEN=<token> python3 -m devx.ci.release [--dry-run] [--skip-tests]
+  CI_GITEA_API_TOKEN=<token> python3 -m devx.ci.release [--dry-run] [--skip-tests]
   python3 -m devx.ci.release --verify   # Check tag/version/release alignment
 """
 

src/devx/ci/sync_wiki.py

@@ -21,7 +21,7 @@ Link transformations:
 - Anchor-only links (``#section``) are preserved
 
 Usage:
-  CI_GITEA_TOKEN=<token> python3 -m devx.ci.sync_wiki [--dry-run] [--repo owner/repo]
+  CI_GITEA_API_TOKEN=<token> python3 -m devx.ci.sync_wiki [--dry-run] [--repo owner/repo]
 """
 
 from __future__ import annotations
@@ -40,6 +40,7 @@ from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnk
 
 from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
 from devx.i18n import _
+from devx.tokens import get_ci_token
 
 load_dotenv()
 
@@ -274,9 +275,10 @@ def commit_and_push(wiki_dir: Path, wiki_url: str, dry_run: bool) -> bool:
 )
 def main(dry_run: bool, repo: str | None, verify: bool) -> None:
     """Sync documentation to the Gitea wiki via Git."""
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_ci_token()
-        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set."))
+    except click.ClickException:
+        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set.")) from None
 
     if repo is None:
         owner = os.environ.get("DEVX_REPO_OWNER", "") or REPO_OWNER

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/gitea_cli.py

@@ -40,7 +40,6 @@ Usage::
 from __future__ import annotations
 
 import json
-import os
 import shutil
 import subprocess  # nosec B404
 from typing import Any
@@ -49,6 +48,7 @@ import click
 
 from devx.config import GITEA_API_URL
 from devx.i18n import _
+from devx.tokens import get_ci_token
 
 
 class TeaCLIError(Exception):
@@ -56,10 +56,10 @@ class TeaCLIError(Exception):
 
 
 def configure_tea_login(login_name: str = "devx") -> None:
-    """Configure tea CLI login from CI_GITEA_TOKEN and DEVX_GITEA_API_URL.
+    """Configure tea CLI login from CI_GITEA_API_TOKEN and DEVX_GITEA_API_URL.
 
     Idempotent: if a login with the same name already exists, it is not re-added.
-    Skips silently if tea is not installed or CI_GITEA_TOKEN is not set.
+    Skips silently if tea is not installed or no token is set.
 
     Used by CI scripts (publish, notify_failure) that need tea login but
     run in containerized environments where ``make setup`` was not called.
@@ -69,8 +69,9 @@ def configure_tea_login(login_name: str = "devx") -> None:
         click.echo(_("tea not installed — skipping login configuration."))
         return
 
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_ci_token()
+    except click.ClickException:
         click.echo(_("CI_GITEA_TOKEN not set — skipping login configuration."))
         return
 

src/devx/make/devx.mak

@@ -61,10 +61,12 @@ DEVX_VALE_LEVEL ?= warning
 # Usage: $(DEVX_PIP_INSTALL) install -e '.[ci,lint]'
 # CI_GITEA_USERNAME can be set in .env, as an env var, or as a Make variable.
 # Projects can alias: PIP_INSTALL = $(DEVX_PIP_INSTALL)
-DEVX_PIP_INSTALL := if [ -z "$$CI_GITEA_TOKEN" ]; then . ./.env 2>/dev/null; fi; \
+DEVX_PIP_INSTALL := if [ -z "$$CI_GITEA_API_TOKEN" ] && [ -z "$$DEVELOPER_GITEA_API_TOKEN" ] && [ -z "$$CI_GITEA_TOKEN" ]; then . ./.env 2>/dev/null; fi; \
-	CI_GITEA_TOKEN="$$CI_GITEA_TOKEN"; \
+	_TOKEN="$$CI_GITEA_API_TOKEN"; \
+	[ -z "$$_TOKEN" ] && _TOKEN="$$DEVELOPER_GITEA_API_TOKEN"; \
+	[ -z "$$_TOKEN" ] && _TOKEN="$$CI_GITEA_TOKEN"; \
 	_PYPI_USER="$${CI_GITEA_USERNAME:-emil}"; \
-	if [ -n "$$CI_GITEA_TOKEN" ] && [ -n "$$_PYPI_USER" ]; then export PIP_EXTRA_INDEX_URL="https://$$_PYPI_USER:$$CI_GITEA_TOKEN@$(DEVX_GITEA_PYPI_HOST)/api/packages/$(DEVX_GITEA_PYPI_ORG)/pypi/simple/"; fi; \
+	if [ -n "$$_TOKEN" ] && [ -n "$$_PYPI_USER" ]; then export PIP_EXTRA_INDEX_URL="https://$$_PYPI_USER:$$_TOKEN@$(DEVX_GITEA_PYPI_HOST)/api/packages/$(DEVX_GITEA_PYPI_ORG)/pypi/simple/"; fi; \
 	$(DEVX_BIN)/pip
 
 # ── Virtual environment management ────────────────────────────────────────────
@@ -196,12 +198,14 @@ devx-pr-rebase:
 # ── Environment setup ─────────────────────────────────────────────────────────
 
 # Configure Gitea private PyPI registry so pip can find devx and other
-# private packages. In CI, CI_GITEA_TOKEN is set as a secret. Locally, it's in .env.
+# private packages. In CI, CI_GITEA_API_TOKEN is set as a secret. Locally, DEVELOPER_GITEA_API_TOKEN or CI_GITEA_TOKEN can be used.
 devx-configure-gitea-pypi:
-	@if [ -z "$$CI_GITEA_TOKEN" ]; then . ./.env 2>/dev/null; fi; \
+	@if [ -z "$$CI_GITEA_API_TOKEN" ] && [ -z "$$DEVELOPER_GITEA_API_TOKEN" ] && [ -z "$$CI_GITEA_TOKEN" ]; then . ./.env 2>/dev/null; fi; \
-	CI_GITEA_TOKEN="$$CI_GITEA_TOKEN"; \
+	_TOKEN="$$CI_GITEA_API_TOKEN"; \
-	if [ -z "$$CI_GITEA_TOKEN" ]; then echo "[configure-gitea-pypi] CI_GITEA_TOKEN not set — skipping (devx must be on public PyPI)"; exit 0; fi; \
+	[ -z "$$_TOKEN" ] && _TOKEN="$$DEVELOPER_GITEA_API_TOKEN"; \
-	echo "[configure-gitea-pypi] Gitea PyPI registry configured (CI_GITEA_TOKEN present)."
+	[ -z "$$_TOKEN" ] && _TOKEN="$$CI_GITEA_TOKEN"; \
+	if [ -z "$$_TOKEN" ]; then echo "[configure-gitea-pypi] Gitea API token not set — skipping (devx must be on public PyPI)"; exit 0; fi; \
+	echo "[configure-gitea-pypi] Gitea PyPI registry configured (token present)."
 
 # Create .env from .env.example if it doesn't exist
 devx-env:
@@ -268,7 +272,7 @@ devx-workflow-check: devx-workflow-lint devx-workflow-dryrun
 
 # Notify on CI failure — creates a Gitea issue via devx.ci.notify_failure.
 # Usage: make devx-notify-failure WORKFLOW=post-merge/release
-# Requires: CI_GITEA_TOKEN, GITHUB_REPOSITORY, GITHUB_RUN_ID, GITHUB_SHA
+# Requires: CI_GITEA_API_TOKEN, GITHUB_REPOSITORY, GITHUB_RUN_ID, GITHUB_SHA
 devx-notify-failure:
 	@. $(DEVX_VENV)/bin/activate 2>/dev/null || true; \
 	export PATH="$(HOME)/.local/bin:$$PATH"; \

src/devx/molecule/discover_runners.py

@@ -30,6 +30,7 @@ import click
 import requests
 
 from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
+from devx.tokens import get_ci_token
 
 DEFAULT_MAX_RUNNERS = 3
 
@@ -86,7 +87,7 @@ def query_runners(api_url: str, token: str, owner: str, repo: str) -> int:
     return total
 
 
-def get_runner_count(api_url: str, token: str, owner: str, repo: str) -> int:
+def get_runner_count(api_url: str, token: str | None, owner: str, repo: str) -> int:
     """Determine the number of available runners.
 
     Tries the Gitea API first, then falls back to env vars, then default.
@@ -142,7 +143,10 @@ def main(
     output_indices: bool,
     github_output: bool,
 ) -> None:
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
+        token = get_ci_token()
+    except click.ClickException:
+        token = None
 
     if owner is None:
         owner = os.environ.get("DEVX_REPO_OWNER", "") or REPO_OWNER

src/devx/molecule/molecule_ci_guard.py

@@ -22,7 +22,7 @@ Usage::
 
 Environment variables:
     GITEA_URL           Base URL of the Gitea instance.
-    CI_GITEA_TOKEN      API token with repo access.
+    CI_GITEA_API_TOKEN  API token with repo access (CI_GITEA_TOKEN accepted for legacy).
     RUN_ID              Workflow run ID (GITHUB_RUN_ID).
     JOB_NAME            Base job name (GITHUB_JOB), e.g. "molecule-tests".
     MATRIX_INDEX        Current matrix index (runner-index).
@@ -45,6 +45,7 @@ import requests
 
 from devx.config import REPO_NAME, REPO_OWNER
 from devx.i18n import _
+from devx.tokens import get_ci_token
 
 POLL_INTERVAL = 10
 
@@ -164,7 +165,10 @@ def resolve_role_dir(role: str, roles_root: Path | None, repo_root: Path) -> Pat
 def cli(pairs: tuple[str, ...], roles_root: Path | None) -> None:
     """Run molecule pairs sequentially, stop if another CI runner fails."""
     gitea_url = os.environ.get("GITEA_URL", "")
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
+        token = get_ci_token()
+    except click.ClickException:
+        token = None
     run_id = int(os.environ.get("RUN_ID", "0"))
     job_name = os.environ.get("JOB_NAME", "molecule-tests")
     current_index = int(os.environ.get("MATRIX_INDEX", "0"))

src/devx/tokens.py

@@ -0,0 +1,76 @@
+"""Token resolution helpers for devx tools.
+
+Centralizes Gitea/Vikunja token discovery with role-based environment
+variable names and backwards compatibility with the legacy
+``CI_GITEA_TOKEN`` / ``REVIEW_GITEA_TOKEN`` naming convention.
+
+Roles:
+- ``CI_GITEA_API_TOKEN``: CI workflows (read actions, post status, merge, etc.)
+- ``REVIEWER_GITEA_API_TOKEN``: PR approval reviews (must be a different user
+  from the PR author for Gitea to accept the review as an approval)
+- ``DEVELOPER_GITEA_API_TOKEN``: local development tools (create-task,
+  create-pr, setup, etc.)
+
+Fallbacks:
+- New role names are checked first.
+- Legacy names (``CI_GITEA_TOKEN``, ``REVIEW_GITEA_TOKEN``) are accepted for
+  backwards compatibility.
+- If no role-specific token is set, the generic CI tokens are tried last.
+"""
+
+from __future__ import annotations
+
+import os
+
+import click
+
+from devx.i18n import _
+
+# Token environment variable names, in lookup priority order.
+CI_TOKEN_NAMES = ["CI_GITEA_API_TOKEN", "CI_GITEA_TOKEN"]
+REVIEWER_TOKEN_NAMES = [
+    "REVIEWER_GITEA_API_TOKEN",
+    # Legacy name used before role-based tokens.
+    "REVIEW_GITEA_TOKEN",
+    *CI_TOKEN_NAMES,
+]
+DEVELOPER_TOKEN_NAMES = ["DEVELOPER_GITEA_API_TOKEN", *CI_TOKEN_NAMES]
+
+VIKUNJA_TOKEN_NAMES = ["VIKUNJA_TOKEN"]
+
+
+def get_token(*names: str) -> str:
+    """Return the first non-empty value from the listed environment variables.
+
+    Raises a ``click.ClickException`` if none of the listed variables are set.
+    """
+    for name in names:
+        token = os.environ.get(name, "").strip()
+        if token:
+            return token
+    raise click.ClickException(
+        _(
+            "Gitea API token not set. Set one of: {names}",
+            names=", ".join(names),
+        )
+    )
+
+
+def get_ci_token() -> str:
+    """Resolve the CI Gitea API token."""
+    return get_token(*CI_TOKEN_NAMES)
+
+
+def get_reviewer_token() -> str:
+    """Resolve the reviewer Gitea API token used for PR approvals."""
+    return get_token(*REVIEWER_TOKEN_NAMES)
+
+
+def get_developer_token() -> str:
+    """Resolve the developer Gitea API token used for local tooling."""
+    return get_token(*DEVELOPER_TOKEN_NAMES)
+
+
+def get_vikunja_token() -> str:
+    """Resolve the Vikunja API token."""
+    return get_token(*VIKUNJA_TOKEN_NAMES)

src/devx/tools/_shared.py

@@ -8,6 +8,8 @@ import subprocess  # nosec B404
 
 import click
 
+from devx.tokens import get_developer_token
+
 
 def arch_string() -> str:
     """Return the architecture string used by release assets.
@@ -45,8 +47,9 @@ def detect_pr_number() -> int | None:
     if branch == "HEAD":
         return None
 
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_developer_token()
+    except click.ClickException:
         return None
 
     owner = os.environ.get("DEVX_REPO_OWNER", "")

src/devx/tools/build_image.py

@@ -34,8 +34,8 @@ The manifest file is a JSON list of dicts, each with:
     - ``context``: build context directory (optional, defaults to repo root)
     - ``tags``: list of tags (optional, defaults to ``["latest"]``)
 
-Registry authentication uses ``CI_GITEA_TOKEN`` and ``CI_GITEA_USERNAME``
+Registry authentication uses ``CI_GITEA_API_TOKEN`` (or legacy ``CI_GITEA_TOKEN``)
-environment variables, matching the existing CI workflow patterns.
+and ``CI_GITEA_USERNAME`` environment variables, matching the existing CI workflow patterns.
 """
 
 from __future__ import annotations
@@ -49,6 +49,7 @@ from pathlib import Path
 import click
 
 from devx.i18n import _
+from devx.tokens import get_developer_token
 
 
 @dataclass
@@ -221,9 +222,12 @@ def push_image(
 
 def _get_registry_creds() -> tuple[str, str]:
     """Get registry credentials from environment variables."""
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
+        token = get_developer_token()
+    except click.ClickException:
+        token = None
     username = os.environ.get("CI_GITEA_USERNAME", "")
-    return username, token
+    return username, token or ""
 
 
 @click.command()

src/devx/tools/clean_images.py

@@ -28,12 +28,11 @@ Usage::
         --keep 2 \\
         --dry-run
 
-Authentication uses ``CI_GITEA_TOKEN`` environment variable.
+Authentication uses ``CI_GITEA_API_TOKEN`` environment variable (or legacy ``CI_GITEA_TOKEN``).
 """
 
 from __future__ import annotations
 
-import os
 import time
 from typing import Any
 
@@ -42,6 +41,7 @@ import requests
 
 from devx.config import GITEA_API_URL, REPO_OWNER
 from devx.i18n import _
+from devx.tokens import get_developer_token
 
 
 def list_package_versions(
@@ -187,9 +187,10 @@ def main(
     api_url: str | None,
 ) -> None:
     """Clean up old Docker image versions from a Gitea registry."""
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_developer_token()
-        raise click.ClickException(_("CI_GITEA_TOKEN environment variable required"))
+    except click.ClickException:
+        raise click.ClickException(_("CI_GITEA_TOKEN environment variable required")) from None
     if not owner:
         owner = REPO_OWNER
     if not owner:

src/devx/tools/configure_repo.py

@@ -7,8 +7,8 @@ ci-improvement, doc-improvement, workflow-improvement) are created
 idempotently via ``ensure_label``.
 
 Usage:
-  CI_GITEA_TOKEN=<token> python3 -m devx.tools.configure_repo --repo my-repo
+  DEVELOPER_GITEA_API_TOKEN=<token> python3 -m devx.tools.configure_repo --repo my-repo
-  CI_GITEA_TOKEN=<token> python3 -m devx.tools.configure_repo --repo my-repo --owner my-org
+  DEVELOPER_GITEA_API_TOKEN=<token> python3 -m devx.tools.configure_repo --repo my-repo --owner my-org
 """
 
 from __future__ import annotations
@@ -23,6 +23,7 @@ from devx.api_clients import GiteaClient
 from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
 from devx.exceptions import APIError
 from devx.i18n import _
+from devx.tokens import get_developer_token
 
 
 def _default_status_checks() -> list[str]:
@@ -175,7 +176,10 @@ def configure_repo(
 )
 def main(repo: str | None, owner: str | None, branch: str, api_url: str | None) -> None:
     """Configure branch protection and repository settings via the Gitea API."""
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
+        token = get_developer_token()
+    except click.ClickException:
+        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set.")) from None
 
     if repo is None:
         repo = os.environ.get("DEVX_REPO_NAME", "") or REPO_NAME

src/devx/tools/create_pr.py

@@ -42,6 +42,7 @@ from devx.config import (
     VIKUNJA_PROJECT_ID,
 )
 from devx.i18n import _
+from devx.tokens import get_developer_token, get_vikunja_token
 
 load_dotenv()
 
@@ -72,9 +73,10 @@ def get_vikunja_task_title(task_id: str) -> str:
 
     Raises ClickException if VIKUNJA_TOKEN is not set or the task is not found.
     """
-    token = os.environ.get("VIKUNJA_TOKEN", "")
+    try:
-    if not token:
+        token = get_vikunja_token()
-        raise click.ClickException(_("VIKUNJA_TOKEN is not set. Required to derive PR title."))
+    except click.ClickException:
+        raise click.ClickException(_("VIKUNJA_TOKEN is not set. Required to derive PR title.")) from None
     client = VikunjaClient(VIKUNJA_API_URL, token)
     task = client.find_task_by_identifier(VIKUNJA_PROJECT_ID, task_id, per_page=DEFAULT_PER_PAGE)
     if not task:
@@ -118,9 +120,10 @@ def create_pr(
             ),
         )
 
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_developer_token()
-        raise click.ClickException(_("CI_GITEA_TOKEN is not set. Required to create a PR."))
+    except click.ClickException:
+        raise click.ClickException(_("CI_GITEA_TOKEN is not set. Required to create a PR.")) from None
 
     vikunja_title = get_vikunja_task_title(task_id)
     pr_title = f"{task_id}: {vikunja_title}"

src/devx/tools/create_task.py

@@ -17,14 +17,13 @@ and ``DEVX_TASK_PREFIX`` environment variables (or ``.env``).
 
 from __future__ import annotations
 
-import os
-
 import click
 from dotenv import load_dotenv
 
 from devx.api_clients import VikunjaClient
 from devx.config import TASK_PREFIX, VIKUNJA_API_URL, VIKUNJA_PROJECT_ID
 from devx.i18n import _
+from devx.tokens import get_vikunja_token
 
 load_dotenv()
 
@@ -39,9 +38,10 @@ load_dotenv()
 @click.option("--project-id", type=int, default=None, help="Vikunja project ID (default: DEVX_VIKUNJA_PROJECT_ID).")
 def cli(title: str, description: str, project_id: int | None) -> None:
     """Create a Vikunja task and print its identifier."""
-    token = os.environ.get("VIKUNJA_TOKEN", "")
+    try:
-    if not token:
+        token = get_vikunja_token()
-        raise click.ClickException(_("VIKUNJA_TOKEN is not set. Set it in .env or environment."))
+    except click.ClickException:
+        raise click.ClickException(_("VIKUNJA_TOKEN is not set. Set it in .env or environment.")) from None
 
     pid = project_id if project_id is not None else VIKUNJA_PROJECT_ID
 

src/devx/tools/pr_label.py

@@ -22,14 +22,13 @@ The repository is auto-detected from ``DEVX_REPO_OWNER`` /
 
 from __future__ import annotations
 
-import os
-
 import click
 from dotenv import load_dotenv
 
 from devx.api_clients import GiteaClient
 from devx.config import GITEA_API_URL, REPO_OWNER
 from devx.i18n import _
+from devx.tokens import get_developer_token
 from devx.tools.create_pr import get_repo_name
 from devx.tools.pr_status import _get_current_branch_pr
 
@@ -48,9 +47,10 @@ def cli(
     repo: str | None,
 ) -> None:
     """Add one or more labels to a pull request (idempotent)."""
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_developer_token()
-        raise click.ClickException(_("CI_GITEA_TOKEN is not set."))
+    except click.ClickException:
+        raise click.ClickException(_("CI_GITEA_TOKEN is not set.")) from None
 
     repo_owner = owner or REPO_OWNER
     if not repo_owner:

src/devx/tools/pr_logs.py

@@ -25,14 +25,13 @@ The repository is auto-detected from ``DEVX_REPO_OWNER`` /
 
 from __future__ import annotations
 
-import os
-
 import click
 from dotenv import load_dotenv
 
 from devx.api_clients import APIError, GiteaClient
 from devx.config import GITEA_API_URL, REPO_OWNER
 from devx.i18n import _
+from devx.tokens import get_developer_token
 from devx.tools.create_pr import get_repo_name
 from devx.tools.pr_status import _get_current_branch_pr
 
@@ -126,9 +125,10 @@ def cli(
     repo: str | None,
 ) -> None:
     """Fetch logs for failed CI jobs on a pull request."""
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_developer_token()
-        raise click.ClickException(_("CI_GITEA_TOKEN is not set."))
+    except click.ClickException:
+        raise click.ClickException(_("CI_GITEA_TOKEN is not set.")) from None
 
     repo_owner = owner or REPO_OWNER
     if not repo_owner:

src/devx/tools/pr_rebase.py

@@ -32,6 +32,7 @@ from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnk
 from devx.api_clients import APIError, GiteaClient
 from devx.config import GITEA_API_URL
 from devx.i18n import _
+from devx.tokens import get_developer_token
 from devx.tools._shared import detect_pr_number
 
 
@@ -41,9 +42,10 @@ def main(pr: int | None) -> None:
     """Rebase a pull request's head branch onto master via Gitea API."""
     load_dotenv()
 
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_developer_token()
-        raise click.ClickException(_("CI_GITEA_TOKEN is not set. Add it to .env or export it."))
+    except click.ClickException:
+        raise click.ClickException(_("CI_GITEA_TOKEN is not set. Add it to .env or export it.")) from None
 
     pr_num = pr or detect_pr_number()
     if not pr_num:

src/devx/tools/pr_status.py

@@ -24,7 +24,6 @@ The repository is auto-detected from ``DEVX_REPO_OWNER`` /
 
 from __future__ import annotations
 
-import os
 import subprocess  # nosec B404
 import time
 
@@ -34,6 +33,7 @@ from dotenv import load_dotenv
 from devx.api_clients import GiteaClient
 from devx.config import GITEA_API_URL, REPO_OWNER
 from devx.i18n import _
+from devx.tokens import get_developer_token
 from devx.tools.create_pr import get_repo_name
 
 load_dotenv()
@@ -139,9 +139,10 @@ def cli(
     repo: str | None,
 ) -> None:
     """Check CI status for a pull request or commit."""
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_developer_token()
-        raise click.ClickException(_("CI_GITEA_TOKEN is not set."))
+    except click.ClickException:
+        raise click.ClickException(_("CI_GITEA_TOKEN is not set.")) from None
 
     repo_owner = owner or REPO_OWNER
     if not repo_owner:

src/devx/tools/pre_push_check.py

@@ -20,7 +20,6 @@ Exit codes:
 
 from __future__ import annotations
 
-import os
 import subprocess  # nosec B404
 
 import click
@@ -29,6 +28,7 @@ from dotenv import load_dotenv
 from devx.api_clients import VikunjaClient
 from devx.config import DEFAULT_PER_PAGE, TASK_ID_RE, TASK_PREFIX, VIKUNJA_API_URL, VIKUNJA_PROJECT_ID
 from devx.i18n import _
+from devx.tokens import get_vikunja_token
 
 load_dotenv()
 
@@ -55,8 +55,9 @@ def task_exists(task_id: str) -> bool:
 
     Returns ``False`` if VIKUNJA_TOKEN is not set (soft-fail in local mode).
     """
-    token = os.environ.get("VIKUNJA_TOKEN", "")
+    try:
-    if not token:
+        token = get_vikunja_token()
+    except click.ClickException:
         return False
     client = VikunjaClient(VIKUNJA_API_URL, token)
     return client.find_task_by_identifier(VIKUNJA_PROJECT_ID, task_id, per_page=DEFAULT_PER_PAGE) is not None
@@ -84,8 +85,9 @@ def validate(branch: str) -> None:
             )
         )
 
-    token = os.environ.get("VIKUNJA_TOKEN", "")
+    try:
-    if not token:
+        get_vikunja_token()
+    except click.ClickException:
         click.echo(
             _(
                 "WARNING: VIKUNJA_TOKEN not set — skipping task existence check. "

src/devx/tools/setup.py

@@ -16,6 +16,8 @@ from pathlib import Path
 import click
 from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnknownVariableType]
 
+from devx.tokens import get_developer_token
+
 load_dotenv()
 
 
@@ -64,19 +66,20 @@ def _install_ansible_collections(bin_dir: str) -> None:
 
 
 def _configure_tea_login() -> None:
-    """Configure tea CLI login from .env if CI_GITEA_TOKEN is set.
+    """Configure tea CLI login from .env if a Gitea token is set.
 
     Idempotent: if a login with the same name already exists, it is not re-added.
-    Skips if tea is not installed or CI_GITEA_TOKEN is not set.
+    Skips if tea is not installed or no Gitea token is set.
     """
     tea_bin = shutil.which("tea")
     if tea_bin is None:
         click.echo("tea: not installed — run 'make install-tools' to install it.")
         return
 
-    token = os.environ.get("CI_GITEA_TOKEN", "")
+    try:
-    if not token:
+        token = get_developer_token()
-        click.echo("tea: CI_GITEA_TOKEN not set — skipping login configuration.")
+    except click.ClickException:
+        click.echo("tea: Gitea API token not set — skipping login configuration.")
         return
 
     api_url = os.environ.get("DEVX_GITEA_API_URL", "https://git.oblachno.oblachno.fyi/api/v1")

src/devx/tools/setup_image.py

@@ -24,6 +24,8 @@ from pathlib import Path
 
 import click
 
+from devx.tokens import get_developer_token
+
 DEFAULT_VENV = ".venv"
 OPT_VENV = "/opt/venv"
 FALLBACK_TARGET = "setup-ci"
@@ -67,7 +69,10 @@ def _install_in_image(
     cmd = [pip_bin, "install", "--no-cache-dir", "-e", spec]
 
     env = os.environ.copy()
-    token = env.get("CI_GITEA_TOKEN", "")
+    try:
+        token = get_developer_token()
+    except click.ClickException:
+        token = None
     if token:
         username = env.get("CI_GITEA_USERNAME", "emil")
         env["PIP_EXTRA_INDEX_URL"] = _build_pip_extra_index_url(

src/devx/translations.json

@@ -959,6 +959,14 @@
     "ru": "CI checks failed.",
     "zh": "CI checks failed."
   },
+  "Gitea API token not set. Set one of: {names}": {
+    "bg": "Gitea API token not set. Set one of: {names}",
+    "de": "Gitea API token not set. Set one of: {names}",
+    "en": "Gitea API token not set. Set one of: {names}",
+    "pl": "Gitea API token not set. Set one of: {names}",
+    "ru": "Gitea API token not set. Set one of: {names}",
+    "zh": "Gitea API token not set. Set one of: {names}"
+  },
   "CI_GITEA_TOKEN environment variable required": {
     "bg": "CI_GITEA_TOKEN environment variable required",
     "de": "CI_GITEA_TOKEN environment variable required",
@@ -3558,5 +3566,45 @@
     "pl": "{separator}",
     "ru": "{separator}",
     "zh": "{separator}"
+  },
+  "Allow empty tag (PR mode where SHA is concrete).": {
+    "bg": "Позволи празен таг (PR режим, където SHA е конкретен).",
+    "de": "Leeren Tag zulassen (PR-Modus, in dem SHA konkret ist).",
+    "en": "Allow empty tag (PR mode where SHA is concrete).",
+    "pl": "Zezwalaj na pusty tag (tryb PR, w którym SHA jest konkretne).",
+    "ru": "Разрешить пустой тег (режим PR, где SHA конкретен).",
+    "zh": "允许空标签（SHA 为具体值的 PR 模式）。"
+  },
+  "Git tag or ref that was deployed": {
+    "bg": "Git таг или референция, която беше разгърната",
+    "de": "Git-Tag oder Ref, der bereitgestellt wurde",
+    "en": "Git tag or ref that was deployed",
+    "pl": "Tag Git lub ref, który został wdrożony",
+    "ru": "Git-тег или ссылка, которые были развёрнуты",
+    "zh": "已部署的 Git 标签或引用"
+  },
+  "Git tag to deploy (e.g. v0.28.1).": {
+    "bg": "Git таг за разгръщане (напр. v0.28.1).",
+    "de": "Git-Tag für Bereitstellung (z.B. v0.28.1).",
+    "en": "Git tag to deploy (e.g. v0.28.1).",
+    "pl": "Tag Git do wdrożenia (np. v0.28.1).",
+    "ru": "Git-тег для развёртывания (напр. v0.28.1).",
+    "zh": "要部署的 Git 标签（例如 v0.28.1）。"
+  },
+  "Write deploy-ref to $GITHUB_OUTPUT file.": {
+    "bg": "Запиши deploy-ref в $GITHUB_OUTPUT файла.",
+    "de": "Deploy-ref in $GITHUB_OUTPUT-Datei schreiben.",
+    "en": "Write deploy-ref to $GITHUB_OUTPUT file.",
+    "pl": "Zapisz deploy-ref do pliku $GITHUB_OUTPUT.",
+    "ru": "Записать deploy-ref в файл $GITHUB_OUTPUT.",
+    "zh": "将 deploy-ref 写入 $GITHUB_OUTPUT 文件。"
+  },
+  "Vikunja task title '{title}' starts with '{prefix}:'. The task title should NOT include the '{prefix}' prefix — it is automatically added to the PR title. Update the Vikunja task title to remove the prefix.": {
+    "bg": "Заглавието на задачата във Vikunja '{title}' започва с '{prefix}:'. Заглавието на задачата НЕ трябва да съдържа префикса '{prefix}' — той се добавя автоматично към заглавието на PR. Актуализирайте заглавието на задачата във Vikunja, за да премахнете префикса.",
+    "de": "Der Vikunja-Aufgabentitel '{title}' beginnt mit '{prefix}:'. Der Aufgabentitel darf NICHT den Präfix '{prefix}' enthalten — er wird automatisch zum PR-Titel hinzugefügt. Aktualisieren Sie den Vikunja-Aufgabentitel, um den Präfix zu entfernen.",
+    "en": "Vikunja task title '{title}' starts with '{prefix}:'. The task title should NOT include the '{prefix}' prefix — it is automatically added to the PR title. Update the Vikunja task title to remove the prefix.",
+    "pl": "Tytuł zadania Vikunja '{title}' zaczyna się od '{prefix}:'. Tytuł zadania nie powinien zawierać prefiksu '{prefix}' — jest on automatycznie dodawany do tytułu PR. Zaktualizuj tytuł zadania Vikunja, aby usunąć prefiks.",
+    "ru": "Заголовок задачи Vikunja '{title}' начинается с '{prefix}:'. Заголовок задачи НЕ должен включать префикс '{prefix}' — он автоматически добавляется к заголовку PR. Обновите заголовок задачи Vikunja, чтобы удалить префикс.",
+    "zh": "Vikunja 任务标题 '{title}' 以 '{prefix}:' 开头。任务标题不应包含 '{prefix}' 前缀 — 它会自动添加到 PR 标题中。请更新 Vikunja 任务标题以删除前缀。"
   }
 }

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_check_auto_merge_ready.py

@@ -292,3 +292,18 @@ class TestCli:
             )
         assert result.exit_code != 0
         assert "does not match Vikunja" in result.output
+
+    def test_fails_with_double_prefix_in_vikunja_title(self) -> None:
+        """Vikunja title with task ID prefix causes double-prefix in PR title."""
+        runner = CliRunner()
+        with (
+            patch.dict("os.environ", {"DEVX_TASK_PREFIX": "DEVX", "VIKUNJA_TOKEN": "tok"}, clear=True),
+            patch("devx.ci.check_auto_merge_ready.is_branch_behind_master", return_value=False),
+            patch("devx.ci.check_auto_merge_ready.get_vikunja_title_optional", return_value="DEVX-1: Fix foo"),
+        ):
+            result = runner.invoke(
+                cli,
+                ["--branch", "DEVX-1-fix-foo", "--pr-title", "DEVX-1: Fix foo"],
+            )
+        assert result.exit_code != 0
+        assert "should NOT include" in result.output

tests/unit/test_discover_runners.py

@@ -4,6 +4,7 @@ import json
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
+import click
 import pytest
 from click.testing import CliRunner
 
@@ -249,3 +250,14 @@ class TestMain:
         args, kwargs = mock_count.call_args
         assert "myorg" in args
         assert "myrepo" in args
+
+    @patch("devx.ci.discover_runners.get_ci_token", side_effect=click.ClickException("no token"))
+    @patch("devx.ci.discover_runners.get_runner_count", return_value=3)
+    def test_missing_token_runs_without_api(self, mock_count: MagicMock, mock_token: MagicMock) -> None:
+        """When no token is available, runner discovery falls back to env/default."""
+        runner = CliRunner()
+        result = runner.invoke(main, ["--count"])
+        assert result.exit_code == 0
+        assert result.output.strip() == "3"
+        args, _ = mock_count.call_args
+        assert args[1] is None  # token passed as None when missing

tests/unit/test_molecule_ci_guard.py

@@ -186,6 +186,28 @@ class TestCli:
                 timeout=60,
             )
 
+    @patch("devx.molecule.molecule_ci_guard.get_ci_token", side_effect=click.ClickException("no token"))
+    def test_missing_token_runs_without_polling(self, mock_token: MagicMock) -> None:
+        """When no token is available, cross-runner polling is skipped."""
+        from click.testing import CliRunner
+
+        with (
+            patch("devx.molecule.molecule_ci_guard.subprocess.Popen") as mock_popen,
+            patch("devx.molecule.molecule_ci_guard.subprocess.run") as mock_run,
+            patch("devx.molecule.molecule_ci_guard.poll_for_other_failures") as mock_poll,
+            patch("time.sleep"),
+        ):
+            proc = MagicMock()
+            proc.poll.return_value = 0
+            proc.returncode = 0
+            mock_popen.return_value = proc
+            mock_run.return_value = MagicMock(returncode=0)
+
+            runner = CliRunner()
+            result = runner.invoke(cli, ["default|ubuntu-2204|img:latest|"])
+            assert result.exit_code == 0
+            mock_poll.assert_not_called()
+
     def test_invalid_pair_format_raises(self) -> None:
         """Pair with fewer than 2 parts should raise."""
         from click.testing import CliRunner

tests/unit/test_molecule_discover_runners.py

@@ -4,6 +4,7 @@ import json
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
+import click
 import pytest
 from click.testing import CliRunner
 
@@ -219,3 +220,14 @@ class TestMain:
         args, kwargs = mock_count.call_args
         assert "myorg" in args
         assert "myrepo" in args
+
+    @patch("devx.molecule.discover_runners.get_ci_token", side_effect=click.ClickException("no token"))
+    @patch("devx.molecule.discover_runners.get_runner_count", return_value=3)
+    def test_missing_token_runs_without_api(self, mock_count: MagicMock, mock_token: MagicMock) -> None:
+        """When no token is available, runner discovery falls back to env/default."""
+        runner = CliRunner()
+        result = runner.invoke(main, ["--count"])
+        assert result.exit_code == 0
+        assert result.output.strip() == "3"
+        args, _ = mock_count.call_args
+        assert args[1] is None  # token passed as None when missing

tests/unit/test_pr_label.py

@@ -12,9 +12,14 @@ from devx.tools.pr_label import cli
 
 class TestCli:
     def test_no_token_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        monkeypatch.delenv("CI_GITEA_TOKEN", raising=False)
+        for name in ("DEVELOPER_GITEA_API_TOKEN", "CI_GITEA_API_TOKEN", "CI_GITEA_TOKEN"):
+            monkeypatch.delenv(name, raising=False)
         runner = CliRunner()
-        result = runner.invoke(cli, ["--pr", "42", "--label", "ready-to-merge"])
+        result = runner.invoke(
+            cli,
+            ["--pr", "42", "--label", "ready-to-merge"],
+            env={"DEVELOPER_GITEA_API_TOKEN": "", "CI_GITEA_API_TOKEN": "", "CI_GITEA_TOKEN": ""},
+        )
         assert result.exit_code != 0
         assert "CI_GITEA_TOKEN" in result.output
 

tests/unit/test_pr_logs.py

@@ -153,9 +153,14 @@ class TestPrintLogs:
 
 class TestCli:
     def test_no_token_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        monkeypatch.delenv("CI_GITEA_TOKEN", raising=False)
+        for name in ("DEVELOPER_GITEA_API_TOKEN", "CI_GITEA_API_TOKEN", "CI_GITEA_TOKEN"):
+            monkeypatch.delenv(name, raising=False)
         runner = CliRunner()
-        result = runner.invoke(cli, ["--pr", "42"])
+        result = runner.invoke(
+            cli,
+            ["--pr", "42"],
+            env={"DEVELOPER_GITEA_API_TOKEN": "", "CI_GITEA_API_TOKEN": "", "CI_GITEA_TOKEN": ""},
+        )
         assert result.exit_code != 0
         assert "CI_GITEA_TOKEN" in result.output
 

tests/unit/test_pr_review.py

@@ -756,9 +756,10 @@ class TestMain:
         result = runner.invoke(main, ["42", "my-org/my-repo"], env={"CI_GITEA_TOKEN": "fake"})
         assert result.exit_code != 0
 
+    @patch.dict("os.environ", {"CI_GITEA_API_TOKEN": "", "CI_GITEA_TOKEN": ""})
     def test_no_token_raises(self) -> None:
         runner = CliRunner()
-        result = runner.invoke(main, ["42", "my-org/my-repo"], env={"CI_GITEA_TOKEN": ""})
+        result = runner.invoke(main, ["42", "my-org/my-repo"], env={"CI_GITEA_API_TOKEN": "", "CI_GITEA_TOKEN": ""})
         assert result.exit_code != 0
         assert "CI_GITEA_TOKEN" in result.output
 

tests/unit/test_pr_status.py

@@ -122,9 +122,14 @@ class TestWaitForCompletion:
 
 class TestCli:
     def test_no_token_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        monkeypatch.delenv("CI_GITEA_TOKEN", raising=False)
+        for name in ("DEVELOPER_GITEA_API_TOKEN", "CI_GITEA_API_TOKEN", "CI_GITEA_TOKEN"):
+            monkeypatch.delenv(name, raising=False)
         runner = CliRunner()
-        result = runner.invoke(cli, ["--pr", "42"])
+        result = runner.invoke(
+            cli,
+            ["--pr", "42"],
+            env={"DEVELOPER_GITEA_API_TOKEN": "", "CI_GITEA_API_TOKEN": "", "CI_GITEA_TOKEN": ""},
+        )
         assert result.exit_code != 0
         assert "CI_GITEA_TOKEN" in result.output
 

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_sync_wiki.py

@@ -256,9 +256,10 @@ class TestCommitAndPush:
 
 class TestMain:
     def test_no_token_raises(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
-        monkeypatch.delenv("CI_GITEA_TOKEN", raising=False)
+        for name in ("CI_GITEA_API_TOKEN", "CI_GITEA_TOKEN"):
+            monkeypatch.delenv(name, raising=False)
         runner = CliRunner()
-        result = runner.invoke(main, [])
+        result = runner.invoke(main, [], env={"CI_GITEA_API_TOKEN": "", "CI_GITEA_TOKEN": ""})
         assert result.exit_code != 0
         assert "CI_GITEA_TOKEN" in result.output
 

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()