diff --git a/.github/workflows/test-backup-database.yaml b/.github/workflows/test-backup-database.yaml
new file mode 100644
index 00000000..cb867e5c
--- /dev/null
+++ b/.github/workflows/test-backup-database.yaml
@@ -0,0 +1,38 @@
+name: Test backup-database
+
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - gh-actions/infra/backup-database/**
+      - .github/workflows/test-backup-database.yaml
+  pull_request:
+    paths:
+      - gh-actions/infra/backup-database/**
+      - .github/workflows/test-backup-database.yaml
+
+jobs:
+  check-python-syntax:
+    name: py_compile
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v7
+
+      - name: Syntax check all Python files
+        run: |
+          while IFS= read -r file; do
+            echo "Checking $file"
+            python3 -m py_compile "$file"
+          done < <(find gh-actions/infra/backup-database -type f -name '*.py' -print0 | sort -z | tr '\0' '\n')
+
+  unit-tests:
+    name: Python unit tests
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v7
+
+      - name: Run unit tests
+        run: python3 -m unittest discover -s gh-actions/infra/backup-database/tests -p 'test_*.py' -v
diff --git a/.gitignore b/.gitignore
index 6620007e..988af421 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,6 @@
 .vscode
 .idea
 .DS_Store
+
+# Python
+__pycache__/
diff --git a/gh-actions/infra/backup-database/README.md b/gh-actions/infra/backup-database/README.md
new file mode 100644
index 00000000..f642383e
--- /dev/null
+++ b/gh-actions/infra/backup-database/README.md
@@ -0,0 +1,59 @@
+# Database backup action
+
+This composite GitHub Action selects a healthy unit of a Juju application and
+runs a database charm backup action on it. It requires an already authenticated
+Juju session and reports the result in the GitHub job summary.
+
+The [jaas-auth](../jaas-auth/action.yml) action can be used to install Juju and
+authenticate to JAAS beforehand.
+
+## Inputs
+
+| Input         | Required | Default         | Description                                        |
+| ------------- | -------- | --------------- | -------------------------------------------------- |
+| `model`       | yes      |                 | Juju model name                                    |
+| `model-owner` | no       |                 | Juju model owner, for models owned by another user |
+| `application` | yes      |                 | Application whose units will be considered         |
+| `action`      | no       | `create-backup` | Juju action name                                   |
+| `parameters`  | no       | `{}`            | JSON object containing action parameters           |
+| `unit-role`   | no       | `non-primary`   | `non-primary`, `primary`, or `any`                 |
+| `timeout`     | no       | `6h`            | Duration passed to Juju's `--wait`, e.g. `30m`     |
+| `dry-run`     | no       | `false`         | Skip backup creation but still list backups        |
+
+Only `model` and `application` are strictly required. `model-owner` is only
+needed when targeting a model owned by another user.
+
+## Unit selection
+
+Units with an unhealthy workload (`blocked` or `error`) or agent (`error` or
+`lost`) status are excluded from selection. The unit with the workload status
+message `Primary` is treated as the primary. With `non-primary`, the first
+eligible non-primary unit is selected, falling back to the primary with a
+degraded warning when no eligible replica remains. `any` skips role discovery.
+Excluded units and degraded fallbacks are noted in the job summary.
+
+## Usage
+
+```yaml
+steps:
+  - uses: actions/checkout@v7
+  - name: Authenticate to JAAS
+    id: jaas-auth
+    uses: canonical/desktop-engineering/gh-actions/infra/jaas-auth@main
+    with:
+      jaas-controller: ${{ vars.JUJU_CONTROLLER }}
+      jaas-controller-host: ${{ vars.JUJU_CONTROLLER_HOST }}
+      juju-client-id: ${{ secrets.JUJU_CLIENT_ID }}
+      juju-client-secret: ${{ secrets.JUJU_CLIENT_SECRET }}
+  - uses: canonical/desktop-engineering/gh-actions/infra/backup-database@main
+    env:
+      JUJU_DATA: ${{ steps.jaas-auth.outputs.juju-data }}
+    with:
+      model: example-model
+      model-owner: ${{ vars.JUJU_MODEL_OWNER }}
+      application: database
+      dry-run: "true"
+```
+
+Invoke the action once per backup target. Run with `dry-run: "true"` first to
+validate model access and unit selection before enabling backup creation.
diff --git a/gh-actions/infra/backup-database/action.yaml b/gh-actions/infra/backup-database/action.yaml
new file mode 100644
index 00000000..db8bdb76
--- /dev/null
+++ b/gh-actions/infra/backup-database/action.yaml
@@ -0,0 +1,66 @@
+name: Database backup
+description: Select a healthy Juju unit and run a database backup action
+
+inputs:
+  model:
+    description: Juju model name
+    required: true
+  model-owner:
+    description: Juju model owner; only needed for models owned by another user
+    required: false
+  application:
+    description: Juju application to back up
+    required: true
+  action:
+    description: Juju backup action
+    required: false
+    default: create-backup
+  parameters:
+    description: JSON object containing Juju action parameters
+    required: false
+    default: "{}"
+  unit-role:
+    description: Unit role to select (non-primary, primary, or any)
+    required: false
+    default: non-primary
+  timeout:
+    description: Duration passed to Juju's --wait option, e.g. 30m or 6h
+    required: false
+    default: 6h
+  dry-run:
+    description: Validate model access and unit selection without running the action
+    required: false
+    default: "false"
+
+runs:
+  using: composite
+  steps:
+    - name: Run database backup
+      id: database-backup
+      shell: bash
+      env:
+        MODEL: ${{ inputs.model }}
+        MODEL_OWNER: ${{ inputs.model-owner }}
+        APPLICATION: ${{ inputs.application }}
+        ACTION: ${{ inputs.action }}
+        PARAMETERS_JSON: ${{ inputs.parameters }}
+        UNIT_ROLE: ${{ inputs.unit-role }}
+        TIMEOUT: ${{ inputs.timeout }}
+        DRY_RUN: ${{ inputs.dry-run }}
+      run: python3 "${{ github.action_path }}/backup.py"
+
+    - name: Write job summary
+      if: always()
+      shell: bash
+      env:
+        BACKUP_RESULT: ${{ steps.database-backup.outputs.result }}
+      run: |
+        set -eo pipefail
+        [[ -n "${BACKUP_RESULT}" ]] || exit 0
+        {
+          echo "## Database backup"
+          echo
+          echo "| Operation | Target | Unit | Result | Notes |"
+          echo "| --------- | ------ | ---- | ------ | ----- |"
+          jq -Rr 'select(length > 0) | fromjson | [.operation, "`\(.target)`", (.unit // "—" | "`\(.)`"), .result, (.notes // "—")] | "| " + join(" | ") + " |"' <<< "${BACKUP_RESULT}"
+        } >> "${GITHUB_STEP_SUMMARY}"
diff --git a/gh-actions/infra/backup-database/backup.py b/gh-actions/infra/backup-database/backup.py
new file mode 100644
index 00000000..7a3ae5df
--- /dev/null
+++ b/gh-actions/infra/backup-database/backup.py
@@ -0,0 +1,478 @@
+#!/usr/bin/env python3
+
+"""Select a healthy Juju unit and run a database backup."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import re
+import tempfile
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from backup_helpers import (
+    CommandCapture,
+    CommandRunner,
+    join_notes,
+    juju_run_succeeded,
+    parameter_value,
+    read_json_object,
+    run_command,
+    set_result_output,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class SelectionError(ValueError):
+    """Raised when no unit satisfies the requested selection policy."""
+
+
+@dataclass(frozen=True)
+class BackupTarget:
+    """Configuration for one database backup target."""
+
+    application: str
+    model: str
+    model_owner: str = ""
+    action: str = "create-backup"
+    parameters: Mapping[str, Any] = field(default_factory=dict)
+    unit_role: str = "non-primary"
+    timeout: str = "6h"
+
+    @property
+    def qualified_model(self) -> str:
+        """Return the owner-qualified model name expected by Juju."""
+
+        if not self.model_owner:
+            return self.model
+        return f"{self.model_owner}/{self.model}"
+
+
+@dataclass(frozen=True)
+class Exclusion:
+    """A Juju unit excluded from selection and the reasons why."""
+
+    unit: str
+    reasons: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class Selection:
+    """Selected Juju unit and the health information behind that choice."""
+
+    unit: str
+    requested_role: str
+    selected_role: str
+    degraded: bool
+    warning: str | None
+    workload: str
+    agent: str
+    excluded: tuple[Exclusion, ...]
+
+
+@dataclass(frozen=True)
+class RunContext:
+    """Resources and reporting details shared by one target run."""
+
+    target: BackupTarget
+    dry_run: bool
+    output_path: Path | None
+    runner: CommandRunner
+    temporary_directory: Path
+
+    @property
+    def target_name(self) -> str:
+        """Return the target name used in logs and summaries."""
+
+        return f"{self.target.model}/{self.target.application}"
+
+    def capture(self, name: str) -> CommandCapture:
+        """Create command capture paths in this run's temporary directory."""
+
+        return CommandCapture.create(self.temporary_directory, name)
+
+    def record_result(self, *, unit: str | None, result: str, notes: str) -> None:
+        """Record this target's outcome for the final summary step."""
+
+        set_result_output(
+            self.output_path,
+            operation="backup",
+            target=self.target_name,
+            unit=unit,
+            result=result,
+            notes=notes,
+        )
+
+
+UNIT_PATTERN = re.compile(r"/(?P<number>[0-9]+)$")
+SUMMARY_FAILURE = "❌ Failure"
+SUMMARY_SUCCESS = "✅ Success"
+SUMMARY_DEGRADED_SUCCESS = "⚠️ Success (degraded)"
+SUMMARY_DRY_RUN = "⏭️ Dry run"
+SUMMARY_DEGRADED_DRY_RUN = "⏭️ Dry run (degraded)"
+
+
+def target_from_action_inputs(values: Mapping[str, str]) -> BackupTarget:
+    """Build a normalized backup target from action environment values."""
+
+    parameters = json.loads(values.get("PARAMETERS_JSON", "{}"))
+    if not isinstance(parameters, dict):
+        raise TypeError("parameters must be a JSON object")
+    application = values.get("APPLICATION", "").strip()
+    if not application:
+        raise ValueError("application is required")
+    model = values.get("MODEL", "").strip()
+    if not model:
+        raise ValueError("model is required")
+    model_owner = values.get("MODEL_OWNER", "").strip()
+    return BackupTarget(
+        application=application,
+        model=model,
+        model_owner=model_owner,
+        action=values["ACTION"],
+        parameters=parameters,
+        unit_role=values["UNIT_ROLE"],
+        timeout=values["TIMEOUT"],
+    )
+
+
+def _model_arguments(target: BackupTarget) -> list[str]:
+    """Return the --model arguments for Juju commands."""
+
+    return ["--model", target.qualified_model]
+
+
+def select_backup_unit(
+    status: Mapping[str, Any], application: str, unit_role: str
+) -> Selection:
+    """Select an eligible application unit using workload health and role."""
+
+    if unit_role not in {"non-primary", "primary", "any"}:
+        raise SelectionError(f"invalid unit role: {unit_role}")
+    applications = status.get("applications")
+    if not isinstance(applications, dict):
+        raise SelectionError("status does not contain an applications object")
+    application_status = applications.get(application)
+    if not isinstance(application_status, dict):
+        raise SelectionError(f"application {application} is not present in status")
+    units = application_status.get("units")
+    if not isinstance(units, dict):
+        raise SelectionError(
+            f"application {application} does not contain a units object"
+        )
+    if not units:
+        raise SelectionError(f"application {application} has no units")
+
+    parsed_units = sorted(
+        (_parse_unit(name, value) for name, value in units.items()),
+        key=lambda unit: _unit_number(unit["unit"]),
+    )
+    eligible = [unit for unit in parsed_units if not unit["reasons"]]
+    if not eligible:
+        raise SelectionError(f"application {application} has no eligible units")
+
+    primaries = [unit for unit in parsed_units if unit["primary"]]
+    warning = None
+    degraded = False
+    if unit_role == "any":
+        selected = eligible[0]
+    else:
+        if len(primaries) != 1:
+            raise SelectionError(
+                f"application {application} must have exactly one unit with status message Primary"
+            )
+        eligible_primaries = [unit for unit in eligible if unit["primary"]]
+        eligible_replicas = [unit for unit in eligible if not unit["primary"]]
+        if unit_role == "primary":
+            if not eligible_primaries:
+                raise SelectionError("the primary unit is not eligible")
+            selected = eligible_primaries[0]
+        elif eligible_replicas:
+            selected = eligible_replicas[0]
+        else:
+            selected = eligible_primaries[0]
+            degraded = True
+            warning = "no eligible non-primary unit; selected the primary"
+
+    excluded = tuple(
+        Exclusion(unit["unit"], tuple(unit["reasons"]))
+        for unit in parsed_units
+        if unit["reasons"]
+    )
+    return Selection(
+        unit=selected["unit"],
+        requested_role=unit_role,
+        selected_role="primary" if selected["primary"] else "non-primary",
+        degraded=degraded,
+        warning=warning,
+        workload=selected["workload"],
+        agent=selected["agent"],
+        excluded=excluded,
+    )
+
+
+def _parse_unit(name: Any, value: Any) -> dict[str, Any]:
+    """Normalize one Juju unit and record health-based exclusion reasons."""
+
+    if not isinstance(name, str) or not UNIT_PATTERN.search(name):
+        raise SelectionError(f"invalid unit name: {name}")
+    if not isinstance(value, dict):
+        raise SelectionError(f"unit {name} has malformed status")
+    workload_status = value.get("workload-status")
+    agent_status = value.get("juju-status")
+    if not isinstance(workload_status, dict) or not isinstance(agent_status, dict):
+        raise SelectionError(f"unit {name} has malformed workload or agent status")
+    workload = workload_status.get("current")
+    agent = agent_status.get("current")
+    message = workload_status.get("message", "")
+    if (
+        not isinstance(workload, str)
+        or not isinstance(agent, str)
+        or not isinstance(message, str)
+    ):
+        raise SelectionError(f"unit {name} has malformed workload or agent status")
+
+    reasons = []
+    if workload in {"blocked", "error"}:
+        reasons.append(f"workload is {workload}")
+    if agent in {"error", "lost"}:
+        reasons.append(f"agent is {agent}")
+    return {
+        "unit": name,
+        "workload": workload,
+        "agent": agent,
+        "primary": message == "Primary",
+        "reasons": reasons,
+    }
+
+
+def _unit_number(name: str) -> int:
+    """Return the integer suffix used to order Juju unit names."""
+
+    match = UNIT_PATTERN.search(name)
+    if match is None:
+        raise SelectionError(f"invalid unit name: {name}")
+    return int(match.group("number"))
+
+
+def run_target(
+    target: BackupTarget,
+    *,
+    dry_run: bool,
+    output_path: Path | None,
+    command_runner: CommandRunner | None = None,
+    temporary_root: Path | None = None,
+) -> int:
+    """Run unit selection, backup creation, and backup verification."""
+
+    mode = "dry run" if dry_run else "live backup"
+    logger.info("Starting %s for %s/%s", mode, target.model, target.application)
+    with tempfile.TemporaryDirectory(dir=temporary_root) as temporary_directory:
+        context = RunContext(
+            target=target,
+            dry_run=dry_run,
+            output_path=output_path,
+            runner=command_runner or run_command,
+            temporary_directory=Path(temporary_directory),
+        )
+        selection = _select_target_unit(context)
+        if selection is None:
+            return 1
+        logger.info(
+            "Selected %s (%s); workload=%s, agent=%s",
+            selection.unit,
+            selection.selected_role,
+            selection.workload,
+            selection.agent,
+        )
+        if selection.degraded:
+            logger.warning(
+                "Unit selection is degraded; no eligible requested-role unit found"
+            )
+        if selection.excluded:
+            logger.info(
+                "Excluded %d unhealthy unit(s) from selection", len(selection.excluded)
+            )
+
+        notes = _selection_notes(selection)
+        if dry_run:
+            logger.info("Dry run: skipping the backup action")
+        elif not _run_backup_action(context, selection, notes):
+            return 1
+        if not _list_backups(context, selection, notes):
+            return 1
+
+        context.record_result(
+            unit=selection.unit,
+            result=_success_result(selection, dry_run=dry_run),
+            notes=join_notes(
+                notes,
+                "backup action not run" if dry_run and notes else None,
+                "Selection validated; backup action not run"
+                if dry_run and not notes
+                else None,
+            ),
+        )
+        logger.info("Completed %s for %s", mode, context.target_name)
+        return 0
+
+
+def _select_target_unit(context: RunContext) -> Selection | None:
+    """Read Juju status and select a unit, reporting failures to the summary."""
+
+    target = context.target
+    logger.info("Reading Juju status for %s", context.target_name)
+    capture = context.capture("status")
+    command = [
+        "juju",
+        "status",
+        *_model_arguments(target),
+        target.application,
+        "--format=json",
+    ]
+    if capture.run(context.runner, command) != 0:
+        logger.error(
+            "Unable to read Juju status for %s; command output was withheld",
+            context.target_name,
+        )
+        context.record_result(
+            unit=None, result=SUMMARY_FAILURE, notes="Unable to read Juju status"
+        )
+        return None
+    try:
+        status = read_json_object(capture.stdout)
+        return select_backup_unit(status, target.application, target.unit_role)
+    except (json.JSONDecodeError, OSError, ValueError) as error:
+        logger.error(
+            "Unable to select a backup unit for %s: %s", context.target_name, error
+        )
+        context.record_result(
+            unit=None,
+            result=SUMMARY_FAILURE,
+            notes=f"Unable to select a backup unit: {error}",
+        )
+        return None
+
+
+def _run_backup_action(context: RunContext, selection: Selection, notes: str) -> bool:
+    """Run the configured backup action, withholding its captured output."""
+
+    target = context.target
+    logger.info("Running the configured backup action on %s", selection.unit)
+    capture = context.capture("action")
+    command = [
+        "juju",
+        "run",
+        *_model_arguments(target),
+        selection.unit,
+        target.action,
+        f"--wait={target.timeout}",
+        "--format=json",
+        *(
+            f"{name}={parameter_value(value)}"
+            for name, value in target.parameters.items()
+        ),
+    ]
+    result = capture.run(context.runner, command)
+    if result == 0 and juju_run_succeeded(capture.stdout):
+        logger.info("Backup action completed successfully on %s", selection.unit)
+        return True
+    logger.error(
+        "Backup action failed for %s on %s; action output was withheld",
+        context.target_name,
+        selection.unit,
+    )
+    context.record_result(
+        unit=selection.unit,
+        result=SUMMARY_FAILURE,
+        notes=join_notes(notes, "Backup action failed"),
+    )
+    return False
+
+
+def _list_backups(context: RunContext, selection: Selection, notes: str) -> bool:
+    """List backups, forwarding its output and reporting verification failures."""
+
+    target = context.target
+    logger.info("Listing backups on %s; command output follows", selection.unit)
+    capture = context.capture("list-backups")
+    command = [
+        "juju",
+        "run",
+        *_model_arguments(target),
+        selection.unit,
+        "list-backups",
+        f"--wait={target.timeout}",
+        "--format=json",
+    ]
+    result = capture.run(context.runner, command)
+    capture.print()
+    if result == 0 and juju_run_succeeded(capture.stdout):
+        logger.info("Backup listing completed successfully on %s", selection.unit)
+        return True
+    logger.error(
+        "Unable to list backups for %s on %s", context.target_name, selection.unit
+    )
+    failure = (
+        "list-backups failed during dry run"
+        if context.dry_run
+        else "Backup succeeded but list-backups failed"
+    )
+    context.record_result(
+        unit=selection.unit,
+        result=SUMMARY_FAILURE,
+        notes=join_notes(notes, failure),
+    )
+    return False
+
+
+def _selection_notes(selection: Selection) -> str:
+    """Describe degraded selection and every health-based unit exclusion."""
+
+    notes = [selection.warning] if selection.warning else []
+    notes.extend(
+        f"excluded {exclusion.unit}: {', '.join(exclusion.reasons)}"
+        for exclusion in selection.excluded
+    )
+    return "; ".join(notes)
+
+
+def _success_result(selection: Selection, *, dry_run: bool) -> str:
+    """Choose the summary result for dry-run and degraded-success states."""
+
+    if dry_run:
+        return SUMMARY_DEGRADED_DRY_RUN if selection.degraded else SUMMARY_DRY_RUN
+    return SUMMARY_DEGRADED_SUCCESS if selection.degraded else SUMMARY_SUCCESS
+
+
+def main() -> int:
+    """Validate action environment inputs and run the database backup."""
+
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
+    dry_run_value = os.environ.get("DRY_RUN", "false").lower()
+    if dry_run_value not in {"true", "false"}:
+        logger.error("DRY_RUN must be true or false")
+        return 2
+    try:
+        target = target_from_action_inputs(os.environ)
+    except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
+        logger.error("Invalid action inputs: %s", error)
+        return 2
+    output_value = os.environ.get("GITHUB_OUTPUT")
+    return run_target(
+        target,
+        dry_run=dry_run_value == "true",
+        output_path=Path(output_value) if output_value else None,
+        temporary_root=(
+            Path(os.environ["RUNNER_TEMP"]) if os.environ.get("RUNNER_TEMP") else None
+        ),
+    )
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/gh-actions/infra/backup-database/backup_helpers.py b/gh-actions/infra/backup-database/backup_helpers.py
new file mode 100644
index 00000000..704117cb
--- /dev/null
+++ b/gh-actions/infra/backup-database/backup_helpers.py
@@ -0,0 +1,135 @@
+"""Shared command, parsing, and reporting helpers for database backup scripts."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+CommandRunner = Callable[[Sequence[str], Path, Path], int]
+
+
+@dataclass(frozen=True)
+class CommandCapture:
+    """Files capturing one command's stdout and stderr."""
+
+    stdout: Path
+    stderr: Path
+
+    @classmethod
+    def create(cls, directory: Path, name: str) -> CommandCapture:
+        """Create predictable capture paths for a command in ``directory``."""
+
+        return cls(directory / f"{name}.json", directory / f"{name}.err")
+
+    def run(self, runner: CommandRunner, command: Sequence[str]) -> int:
+        """Run ``command`` with this capture pair and return its exit status."""
+
+        return runner(command, self.stdout, self.stderr)
+
+    def print(self) -> None:
+        """Print both captured streams; use only for non-sensitive output."""
+
+        sys.stdout.write(self.stdout.read_text())
+        sys.stderr.write(self.stderr.read_text())
+
+
+def run_command(command: Sequence[str], stdout_path: Path, stderr_path: Path) -> int:
+    """Run a command with both streams captured, returning 127 on launch error."""
+
+    with stdout_path.open("wb") as stdout, stderr_path.open("wb") as stderr:
+        try:
+            return subprocess.run(
+                command, stdout=stdout, stderr=stderr, check=False
+            ).returncode
+        except OSError as error:
+            message = f"Unable to execute {command[0]}: {error}\n"
+            stderr.write(message.encode(errors="replace"))
+            return 127
+
+
+def read_json_object(path: Path) -> dict[str, Any]:
+    """Read a JSON object, rejecting arrays and scalar top-level values."""
+
+    value = json.loads(path.read_text())
+    if not isinstance(value, dict):
+        raise ValueError("JSON document is not an object")
+    return value
+
+
+def juju_operations(output_path: Path) -> tuple[Mapping[str, Any], ...]:
+    """Parse Juju's non-empty operation mapping into operation values."""
+
+    response = read_json_object(output_path)
+    if not response or not all(
+        isinstance(operation, dict) for operation in response.values()
+    ):
+        raise ValueError("Juju output does not contain operations")
+    return tuple(response.values())
+
+
+def juju_run_succeeded(stdout_path: Path) -> bool:
+    """Return whether Juju emitted one or more completed operations."""
+
+    try:
+        operations = juju_operations(stdout_path)
+    except (json.JSONDecodeError, OSError, ValueError):
+        return False
+    return all(operation.get("status") == "completed" for operation in operations)
+
+
+def parameter_value(value: Any) -> str:
+    """Serialize one action parameter into Juju's command-line representation."""
+
+    if isinstance(value, bool):
+        return str(value).lower()
+    if value is None or isinstance(value, (dict, list)):
+        return json.dumps(value, separators=(",", ":"))
+    return str(value)
+
+
+def join_notes(*notes: str | None) -> str:
+    """Join non-empty summary notes with consistent punctuation."""
+
+    return "; ".join(note for note in notes if note)
+
+
+def set_result_output(
+    output_path: Path | None,
+    *,
+    operation: str,
+    target: str,
+    result: str,
+    notes: str,
+    unit: str | None = None,
+) -> None:
+    """Publish one run's outcome as a JSON step output."""
+
+    record = {
+        "operation": summary_text(operation),
+        "target": summary_text(target),
+        "result": summary_text(result),
+        "notes": summary_text(notes),
+        "unit": summary_text(unit) if unit is not None else None,
+    }
+    value = json.dumps(record, separators=(",", ":"))
+    if output_path is None:
+        print(value)
+    else:
+        with output_path.open("a") as output:
+            output.write(f"result={value}\n")
+
+
+def summary_text(value: str) -> str:
+    """Flatten untrusted text so it cannot inject extra summary lines or inline code."""
+
+    return (
+        value.replace("\n", " ")
+        .replace("\r", " ")
+        .replace("`", "'")
+        .replace("|", "\\|")
+    )
diff --git a/gh-actions/infra/backup-database/tests/fixtures/agent-unhealthy.json b/gh-actions/infra/backup-database/tests/fixtures/agent-unhealthy.json
new file mode 100644
index 00000000..d7ad1e3b
--- /dev/null
+++ b/gh-actions/infra/backup-database/tests/fixtures/agent-unhealthy.json
@@ -0,0 +1,20 @@
+{
+  "applications": {
+    "database": {
+      "units": {
+        "database/0": {
+          "workload-status": {"current": "active", "message": "Primary"},
+          "juju-status": {"current": "idle"}
+        },
+        "database/1": {
+          "workload-status": {"current": "active", "message": ""},
+          "juju-status": {"current": "lost"}
+        },
+        "database/2": {
+          "workload-status": {"current": "active", "message": ""},
+          "juju-status": {"current": "idle"}
+        }
+      }
+    }
+  }
+}
diff --git a/gh-actions/infra/backup-database/tests/fixtures/ambiguous-primary.json b/gh-actions/infra/backup-database/tests/fixtures/ambiguous-primary.json
new file mode 100644
index 00000000..a2257932
--- /dev/null
+++ b/gh-actions/infra/backup-database/tests/fixtures/ambiguous-primary.json
@@ -0,0 +1,16 @@
+{
+  "applications": {
+    "database": {
+      "units": {
+        "database/0": {
+          "workload-status": {"current": "active", "message": "Primary"},
+          "juju-status": {"current": "idle"}
+        },
+        "database/1": {
+          "workload-status": {"current": "active", "message": "Primary"},
+          "juju-status": {"current": "idle"}
+        }
+      }
+    }
+  }
+}
diff --git a/gh-actions/infra/backup-database/tests/fixtures/healthy-three.json b/gh-actions/infra/backup-database/tests/fixtures/healthy-three.json
new file mode 100644
index 00000000..0df8ce53
--- /dev/null
+++ b/gh-actions/infra/backup-database/tests/fixtures/healthy-three.json
@@ -0,0 +1,20 @@
+{
+  "applications": {
+    "database": {
+      "units": {
+        "database/0": {
+          "workload-status": {"current": "active", "message": "Primary"},
+          "juju-status": {"current": "idle"}
+        },
+        "database/1": {
+          "workload-status": {"current": "active", "message": ""},
+          "juju-status": {"current": "idle"}
+        },
+        "database/2": {
+          "workload-status": {"current": "active", "message": ""},
+          "juju-status": {"current": "idle"}
+        }
+      }
+    }
+  }
+}
diff --git a/gh-actions/infra/backup-database/tests/fixtures/single-unit.json b/gh-actions/infra/backup-database/tests/fixtures/single-unit.json
new file mode 100644
index 00000000..2ddbbad8
--- /dev/null
+++ b/gh-actions/infra/backup-database/tests/fixtures/single-unit.json
@@ -0,0 +1,12 @@
+{
+  "applications": {
+    "database": {
+      "units": {
+        "database/0": {
+          "workload-status": {"current": "active", "message": "Primary"},
+          "juju-status": {"current": "idle"}
+        }
+      }
+    }
+  }
+}
diff --git a/gh-actions/infra/backup-database/tests/fixtures/unhealthy-replicas.json b/gh-actions/infra/backup-database/tests/fixtures/unhealthy-replicas.json
new file mode 100644
index 00000000..e72e2620
--- /dev/null
+++ b/gh-actions/infra/backup-database/tests/fixtures/unhealthy-replicas.json
@@ -0,0 +1,20 @@
+{
+  "applications": {
+    "database": {
+      "units": {
+        "database/0": {
+          "workload-status": {"current": "active", "message": "Primary"},
+          "juju-status": {"current": "idle"}
+        },
+        "database/1": {
+          "workload-status": {"current": "blocked", "message": "Needs attention"},
+          "juju-status": {"current": "idle"}
+        },
+        "database/2": {
+          "workload-status": {"current": "error", "message": "Hook failed"},
+          "juju-status": {"current": "error"}
+        }
+      }
+    }
+  }
+}
diff --git a/gh-actions/infra/backup-database/tests/test_backup_database.py b/gh-actions/infra/backup-database/tests/test_backup_database.py
new file mode 100644
index 00000000..4ab85994
--- /dev/null
+++ b/gh-actions/infra/backup-database/tests/test_backup_database.py
@@ -0,0 +1,530 @@
+from __future__ import annotations
+
+import importlib.util
+import io
+import json
+import sys
+import tempfile
+import unittest
+from contextlib import redirect_stderr, redirect_stdout
+from pathlib import Path
+from typing import Any
+
+ACTION_DIR = Path(__file__).parents[1]
+sys.path.insert(0, str(ACTION_DIR))
+
+
+def load_module(name: str):
+    spec = importlib.util.spec_from_file_location(name, ACTION_DIR / f"{name}.py")
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+backup_helpers = load_module("backup_helpers")
+backup = load_module("backup")
+TEST_MODEL_OWNER = "test-owner"
+DEFAULT_ACTION_INPUTS = {
+    "ACTION": "create-backup",
+    "PARAMETERS_JSON": "{}",
+    "UNIT_ROLE": "non-primary",
+    "TIMEOUT": "6h",
+}
+
+
+class FakeRunner:
+    def __init__(
+        self,
+        fixture: dict[str, Any],
+        fail_model: str | None = None,
+        fail_status: bool = False,
+        failed_action: str | None = None,
+    ):
+        self.fixture = fixture
+        self.fail_model = fail_model
+        self.fail_status = fail_status
+        self.failed_action = failed_action
+        self.commands: list[list[str]] = []
+
+    @staticmethod
+    def _model(command: list[str]) -> str | None:
+        if "--model" not in command:
+            return None
+        return command[command.index("--model") + 1]
+
+    @staticmethod
+    def _unit(command: list[str]) -> str:
+        if "--model" in command:
+            return command[command.index("--model") + 2]
+        return command[2]
+
+    @staticmethod
+    def _action(command: list[str]) -> str:
+        if "--model" in command:
+            return command[command.index("--model") + 3]
+        return command[3]
+
+    def __call__(self, command: list[str], stdout: Path, stderr: Path) -> int:
+        self.commands.append(list(command))
+        stdout.touch()
+        stderr.touch()
+        if command[1] == "status":
+            if self.fail_status:
+                stderr.write_text("simulated status failure")
+                return 1
+            stdout.write_text(json.dumps(self.fixture))
+            return 0
+        if self.fail_model and self._model(command) == self.fail_model:
+            stderr.write_text("simulated failure")
+            return 1
+        action = self._action(command)
+        unit = self._unit(command)
+        status = "failed" if action == self.failed_action else "completed"
+        if action == "list-backups":
+            stdout.write_text(
+                json.dumps(
+                    {
+                        unit: {
+                            "message": "backup-2026-07-22",
+                            "status": status,
+                        }
+                    }
+                )
+            )
+            stderr.write_text("list warning\n")
+            return 0
+        stdout.write_text(json.dumps({unit: {"status": status}}))
+        return 0
+
+
+def read_records(path: Path) -> list[dict]:
+    name, value = path.read_text().strip().split("=", maxsplit=1)
+    assert name == "result"
+    return [json.loads(value)]
+
+
+class DatabaseBackupTests(unittest.TestCase):
+    fixtures = Path(__file__).parent / "fixtures"
+
+    def fixture(self, name: str) -> dict[str, Any]:
+        return json.loads((self.fixtures / name).read_text())
+
+    def test_selects_first_healthy_replica(self):
+        selection = backup.select_backup_unit(
+            self.fixture("healthy-three.json"), "database", "non-primary"
+        )
+        self.assertEqual(selection.unit, "database/1")
+        self.assertFalse(selection.degraded)
+
+    def test_selects_primary_when_requested(self):
+        selection = backup.select_backup_unit(
+            self.fixture("healthy-three.json"), "database", "primary"
+        )
+        self.assertEqual(selection.unit, "database/0")
+
+    def test_any_selects_first_healthy_unit(self):
+        selection = backup.select_backup_unit(
+            self.fixture("healthy-three.json"), "database", "any"
+        )
+        self.assertEqual(selection.unit, "database/0")
+
+    def test_falls_back_to_primary_and_records_exclusions(self):
+        selection = backup.select_backup_unit(
+            self.fixture("unhealthy-replicas.json"), "database", "non-primary"
+        )
+        self.assertEqual(selection.unit, "database/0")
+        self.assertTrue(selection.degraded)
+        self.assertEqual(len(selection.excluded), 2)
+
+    def test_excludes_lost_agent(self):
+        selection = backup.select_backup_unit(
+            self.fixture("agent-unhealthy.json"), "database", "non-primary"
+        )
+        self.assertEqual(selection.unit, "database/2")
+        self.assertEqual(
+            selection.excluded[0],
+            backup.Exclusion("database/1", ("agent is lost",)),
+        )
+
+    def test_single_unit_is_selected(self):
+        selection = backup.select_backup_unit(
+            self.fixture("single-unit.json"), "database", "non-primary"
+        )
+        self.assertEqual(selection.unit, "database/0")
+        self.assertTrue(selection.degraded)
+        self.assertEqual(
+            selection.warning, "no eligible non-primary unit; selected the primary"
+        )
+
+    def test_rejects_ambiguous_primary(self):
+        with self.assertRaisesRegex(backup.SelectionError, "exactly one"):
+            backup.select_backup_unit(
+                self.fixture("ambiguous-primary.json"), "database", "non-primary"
+            )
+
+    def test_action_inputs_are_passed_through(self):
+        backup_target = backup.target_from_action_inputs(
+            {
+                "MODEL": "model:variant",
+                "MODEL_OWNER": "owner",
+                "APPLICATION": "database_app",
+                "ACTION": "custom_action",
+                "PARAMETERS_JSON": '{"nested":{"enabled":true}}',
+                "UNIT_ROLE": "any",
+                "TIMEOUT": "45s",
+            }
+        )
+        self.assertEqual(backup_target.qualified_model, "owner/model:variant")
+        self.assertEqual(backup_target.action, "custom_action")
+        self.assertEqual(backup_target.parameters, {"nested": {"enabled": True}})
+        self.assertEqual(backup_target.unit_role, "any")
+        self.assertEqual(backup_target.timeout, "45s")
+
+    def test_invalid_parameter_json_fails(self):
+        with self.assertRaises(json.JSONDecodeError):
+            backup.target_from_action_inputs(
+                {
+                    "MODEL": "demo",
+                    "MODEL_OWNER": TEST_MODEL_OWNER,
+                    "APPLICATION": "database",
+                    "PARAMETERS_JSON": "{",
+                }
+            )
+
+    def test_action_model_owner_qualifies_model(self):
+        backup_target = backup.target_from_action_inputs(
+            {
+                **DEFAULT_ACTION_INPUTS,
+                "MODEL": "database-model",
+                "MODEL_OWNER": TEST_MODEL_OWNER,
+                "APPLICATION": "database",
+            }
+        )
+
+        self.assertEqual(
+            backup_target.qualified_model,
+            f"{TEST_MODEL_OWNER}/database-model",
+        )
+
+    def test_model_owner_is_optional(self):
+        backup_target = backup.target_from_action_inputs(
+            {
+                **DEFAULT_ACTION_INPUTS,
+                "MODEL": "database-model",
+                "APPLICATION": "database",
+            }
+        )
+
+        self.assertEqual(backup_target.qualified_model, "database-model")
+
+    def test_model_is_required(self):
+        with self.assertRaisesRegex(ValueError, "model is required"):
+            backup.target_from_action_inputs(
+                {"MODEL_OWNER": TEST_MODEL_OWNER, "APPLICATION": "database"}
+            )
+
+    def test_application_is_required(self):
+        with self.assertRaisesRegex(ValueError, "application is required"):
+            backup.target_from_action_inputs(
+                {"MODEL": "database-model", "MODEL_OWNER": TEST_MODEL_OWNER}
+            )
+
+    def test_summary_text_flattens_markdown_and_newlines(self):
+        self.assertEqual(
+            backup_helpers.summary_text("line one\nline two\r`code`"),
+            "line one line two 'code'",
+        )
+
+    def test_selection_failure_summary_includes_reason(self):
+        runner = FakeRunner(self.fixture("ambiguous-primary.json"))
+        target = backup.BackupTarget(
+            model="ambiguous-model",
+            application="database",
+            model_owner=TEST_MODEL_OWNER,
+        )
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory)
+            result = backup.run_target(
+                target,
+                dry_run=True,
+                output_path=root / "github-output",
+                command_runner=runner,
+                temporary_root=root,
+            )
+            records = read_records(root / "github-output")
+
+        self.assertEqual(result, 1)
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0]["target"], "ambiguous-model/database")
+        self.assertEqual(records[0]["operation"], "backup")
+        self.assertEqual(records[0]["result"], "❌ Failure")
+        self.assertIsNone(records[0]["unit"])
+        self.assertIn(
+            "exactly one unit with status message Primary", records[0]["notes"]
+        )
+
+    def test_dry_run_skips_backup_and_lists_backups(self):
+        runner = FakeRunner(self.fixture("healthy-three.json"))
+        target = backup.BackupTarget(
+            model="dry-run",
+            application="database",
+            model_owner=TEST_MODEL_OWNER,
+        )
+        stdout = io.StringIO()
+        stderr = io.StringIO()
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory)
+            with redirect_stdout(stdout), redirect_stderr(stderr):
+                result = backup.run_target(
+                    target,
+                    dry_run=True,
+                    output_path=root / "github-output",
+                    command_runner=runner,
+                    temporary_root=root,
+                )
+            records = read_records(root / "github-output")
+
+        run_commands = [command for command in runner.commands if command[1] == "run"]
+        self.assertEqual(result, 0)
+        self.assertEqual(
+            [FakeRunner._action(command) for command in run_commands],
+            ["list-backups"],
+        )
+        self.assertIn("backup-2026-07-22", stdout.getvalue())
+        self.assertIn("list warning", stderr.getvalue())
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0]["target"], "dry-run/database")
+        self.assertEqual(records[0]["unit"], "database/1")
+        self.assertEqual(records[0]["result"], "⏭️ Dry run")
+        self.assertEqual(
+            records[0]["notes"], "Selection validated; backup action not run"
+        )
+
+    def test_status_failure_summary_omits_unit(self):
+        runner = FakeRunner(self.fixture("healthy-three.json"), fail_status=True)
+        target = backup.BackupTarget(
+            model="unavailable-model",
+            application="database",
+            model_owner=TEST_MODEL_OWNER,
+        )
+        stderr = io.StringIO()
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory)
+            with redirect_stderr(stderr):
+                result = backup.run_target(
+                    target,
+                    dry_run=False,
+                    output_path=root / "github-output",
+                    command_runner=runner,
+                    temporary_root=root,
+                )
+            records = read_records(root / "github-output")
+
+        self.assertEqual(result, 1)
+        self.assertIn("command output was withheld", stderr.getvalue())
+        self.assertNotIn("simulated status failure", stderr.getvalue())
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0]["target"], "unavailable-model/database")
+        self.assertEqual(records[0]["result"], "❌ Failure")
+        self.assertEqual(records[0]["notes"], "Unable to read Juju status")
+        self.assertIsNone(records[0]["unit"])
+
+    def test_command_launch_error_returns_nonzero_and_captures_error(self):
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory)
+            stdout = root / "stdout"
+            stderr = root / "stderr"
+            result = backup_helpers.run_command(
+                ["executable-that-does-not-exist"], stdout, stderr
+            )
+
+            self.assertNotEqual(result, 0)
+            self.assertEqual(stdout.read_text(), "")
+            self.assertIn("executable-that-does-not-exist", stderr.read_text())
+
+    def test_juju_run_parser_rejects_invalid_operation_envelopes(self):
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            output = Path(temporary_directory) / "output.json"
+            for response in ({}, [], {"database/0": "completed"}):
+                with self.subTest(response=response):
+                    output.write_text(json.dumps(response))
+                    self.assertFalse(backup_helpers.juju_run_succeeded(output))
+
+    def test_builds_action_argument_list(self):
+        runner = FakeRunner(self.fixture("healthy-three.json"))
+        target = backup.BackupTarget(
+            model="successful-model",
+            application="database",
+            model_owner=TEST_MODEL_OWNER,
+            parameters={
+                "force": True,
+                "label": "nightly backup",
+                "nested": {"enabled": True},
+            },
+            timeout="42m",
+        )
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory)
+            result = backup.run_target(
+                target,
+                dry_run=False,
+                output_path=root / "github-output",
+                command_runner=runner,
+                temporary_root=root,
+            )
+            records = read_records(root / "github-output")
+
+        action_command = next(
+            command for command in runner.commands if command[1] == "run"
+        )
+        self.assertEqual(result, 0)
+        self.assertEqual(
+            [FakeRunner._model(command) for command in runner.commands],
+            [f"{TEST_MODEL_OWNER}/successful-model"] * 3,
+        )
+        self.assertIn("force=true", action_command)
+        self.assertIn("label=nightly backup", action_command)
+        self.assertIn('nested={"enabled":true}', action_command)
+        self.assertIn("--wait=42m", action_command)
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0]["target"], "successful-model/database")
+        self.assertEqual(records[0]["result"], "✅ Success")
+
+    def test_lists_backups_and_prints_output_after_success(self):
+        runner = FakeRunner(self.fixture("healthy-three.json"))
+        target = backup.BackupTarget(
+            model="successful-model",
+            application="database",
+            model_owner=TEST_MODEL_OWNER,
+        )
+        stdout = io.StringIO()
+        stderr = io.StringIO()
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory)
+            with redirect_stdout(stdout), redirect_stderr(stderr):
+                result = backup.run_target(
+                    target,
+                    dry_run=False,
+                    output_path=root / "github-output",
+                    command_runner=runner,
+                    temporary_root=root,
+                )
+
+        run_commands = [command for command in runner.commands if command[1] == "run"]
+        self.assertEqual(result, 0)
+        self.assertEqual(
+            [FakeRunner._action(command) for command in run_commands],
+            ["create-backup", "list-backups"],
+        )
+        self.assertEqual(FakeRunner._unit(run_commands[1]), "database/1")
+        self.assertIn("backup-2026-07-22", stdout.getvalue())
+        self.assertIn("list warning", stderr.getvalue())
+
+    def test_degraded_success_uses_warning_result(self):
+        runner = FakeRunner(self.fixture("unhealthy-replicas.json"))
+        target = backup.BackupTarget(
+            model="degraded-model",
+            application="database",
+            model_owner=TEST_MODEL_OWNER,
+        )
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory)
+            result = backup.run_target(
+                target,
+                dry_run=False,
+                output_path=root / "github-output",
+                command_runner=runner,
+                temporary_root=root,
+            )
+            records = read_records(root / "github-output")
+
+        self.assertEqual(result, 0)
+        self.assertEqual(records[0]["result"], "⚠️ Success (degraded)")
+
+    def test_failed_action_returns_failure(self):
+        runner = FakeRunner(
+            self.fixture("healthy-three.json"),
+            fail_model=f"{TEST_MODEL_OWNER}/failed-model",
+        )
+        target = backup.BackupTarget(
+            model="failed-model",
+            application="database",
+            model_owner=TEST_MODEL_OWNER,
+            unit_role="primary",
+        )
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory)
+            result = backup.run_target(
+                target,
+                dry_run=False,
+                output_path=root / "github-output",
+                command_runner=runner,
+                temporary_root=root,
+            )
+            records = read_records(root / "github-output")
+
+        self.assertEqual(result, 1)
+        self.assertEqual(records[0]["target"], "failed-model/database")
+        self.assertEqual(records[0]["unit"], "database/0")
+        self.assertEqual(records[0]["result"], "❌ Failure")
+        self.assertIn("Backup action failed", records[0]["notes"])
+
+    def test_failed_action_status_returns_failure_when_cli_succeeds(self):
+        runner = FakeRunner(
+            self.fixture("healthy-three.json"), failed_action="create-backup"
+        )
+        target = backup.BackupTarget(
+            model="failed-model",
+            application="database",
+            model_owner=TEST_MODEL_OWNER,
+            unit_role="primary",
+        )
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory)
+            result = backup.run_target(
+                target,
+                dry_run=False,
+                output_path=root / "github-output",
+                command_runner=runner,
+                temporary_root=root,
+            )
+            records = read_records(root / "github-output")
+
+        run_commands = [command for command in runner.commands if command[1] == "run"]
+        self.assertEqual(result, 1)
+        self.assertEqual(
+            [FakeRunner._action(command) for command in run_commands],
+            ["create-backup"],
+        )
+        self.assertEqual(records[0]["result"], "❌ Failure")
+        self.assertIn("Backup action failed", records[0]["notes"])
+
+    def test_failed_list_status_returns_failure_when_cli_succeeds(self):
+        runner = FakeRunner(
+            self.fixture("healthy-three.json"), failed_action="list-backups"
+        )
+        target = backup.BackupTarget(
+            model="dry-run",
+            application="database",
+            model_owner=TEST_MODEL_OWNER,
+        )
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory)
+            result = backup.run_target(
+                target,
+                dry_run=True,
+                output_path=root / "github-output",
+                command_runner=runner,
+                temporary_root=root,
+            )
+            records = read_records(root / "github-output")
+
+        self.assertEqual(result, 1)
+        self.assertEqual(records[0]["result"], "❌ Failure")
+        self.assertIn("list-backups failed during dry run", records[0]["notes"])
+
+
+if __name__ == "__main__":
+    unittest.main()