Add borg prune and compact jobs

Run repository retention either after a successful backup or from a
dedicated systemd timer. Clean up script generation with templates and
expand molecule test coverage.

BREAKING CHANGE: Aggregate backup scripts are no longer managed, and
state=preset now requires at least one readable included directory.

10 files changed, 844 insertions, 87 deletions

diff --git a/molecule/default/converge.yml b/molecule/default/converge.yml
index 90b3258..0e9fbf0 100644
--- a/molecule/default/converge.yml
+++ b/molecule/default/converge.yml
@@ -6,6 +6,8 @@
     - borg-client-multi
     - borg-client-nonroot
     - borg-client-multi-keys
+    - borg-client-transition
+    - borg-client-validation
 
   vars:
     borg_server_host: borg-server
@@ -64,6 +66,11 @@
           - /opt
           - /var
           - /reee reeee
+        borg_prune_enabled: true
+        borg_prune_keep_last: "2"
+        borg_prune_glob_archives: "{hostname}-*"
+        borg_prune_compact_enabled: true
+        borg_compact_threshold: 1
 
 - name: Converge - borg-client-2 (custom server user)
   hosts: borg-client-2
@@ -93,6 +100,17 @@
           - /opt
           - /var
           - /reee reeee
+        borg_prune_enabled: true
+        borg_prune_trigger: timer
+        borg_prune_keep_daily: "7"
+        borg_prune_keep_weekly: "4"
+        borg_prune_glob_archives: "{hostname}-*"
+        borg_prune_compact_enabled: true
+        borg_compact_threshold: 10
+        borg_prune_service_successful_exit_status:
+          - 1
+          - TEMPFAIL
+        borg_prune_systemd_oncalendar: "*-*-* 05:00:00"
 
 - name: Converge - Multi-instance backup (same host, different repos)
   hosts: borg-client-multi
@@ -190,3 +208,149 @@
         borg_excluded_dirs:
           - /home/*/.cache
         borg_systemd_oncalendar: "*-*-* 04:00:00"
+
+- name: Converge - Transition host in stable after_backup state
+  hosts: borg-client-transition
+
+  pre_tasks:
+    - name: Seed legacy aggregate backup script
+      ansible.builtin.copy:
+        dest: /usr/local/bin/run_borg_backup
+        content: |
+          #!/bin/bash
+          ## BEGIN ANSIBLE MANAGED BLOCK for borg-server/transition-repo
+          echo legacy-backup-block
+          ## END ANSIBLE MANAGED BLOCK for borg-server/transition-repo
+        owner: root
+        group: root
+        mode: "0711"
+        force: false
+      become: true
+
+  roles:
+    - role: kliwniloc.borgbackup
+      vars:
+        borg_server_host: borg-server
+        borg_server_user_home: /opt/borg
+        borg_decryption_keys_yaml_path: "{{ playbook_dir }}/decryption_keys.yml"
+        borg_repo_name: transition-repo
+        borg_backup_argument: transition-repo
+        borg_included_dirs:
+          - /etc
+        borg_excluded_dirs: []
+        borg_prune_enabled: true
+        borg_prune_trigger: after_backup
+        borg_prune_keep_last: "2"
+        borg_prune_glob_archives: "{hostname}-*"
+
+- name: Converge - Validation guard coverage
+  hosts: borg-client-validation
+
+  tasks:
+    - name: Verify prune with append-only fails validation
+      block:
+        - name: Run role with incompatible append-only pruning
+          ansible.builtin.include_role:
+            name: kliwniloc.borgbackup
+          vars:
+            borg_server_host: borg-server
+            borg_server_user_home: /opt/borg
+            borg_decryption_keys_yaml_path: "{{ playbook_dir }}/decryption_keys.yml"
+            borg_repo_name: validation-append-only
+            borg_backup_argument: validation-append-only
+            borg_included_dirs:
+              - /etc
+            borg_prune_enabled: true
+            borg_mode_append_only: true
+            borg_prune_keep_last: "2"
+            borg_prune_glob_archives: "{hostname}-*"
+
+        - name: Fail when append-only prune validation unexpectedly passes
+          ansible.builtin.fail:
+            msg: Append-only prune validation unexpectedly passed
+      rescue:
+        - name: Assert append-only prune validation failed as expected
+          ansible.builtin.assert:
+            that:
+              - >-
+                'incompatible with borg_mode_append_only'
+                in (ansible_failed_result.msg | default(''))
+
+    - name: Verify prune without a retention policy fails validation
+      block:
+        - name: Run role without a prune retention policy
+          ansible.builtin.include_role:
+            name: kliwniloc.borgbackup
+          vars:
+            borg_server_host: borg-server
+            borg_server_user_home: /opt/borg
+            borg_decryption_keys_yaml_path: "{{ playbook_dir }}/decryption_keys.yml"
+            borg_repo_name: validation-no-retention
+            borg_backup_argument: validation-no-retention
+            borg_included_dirs:
+              - /etc
+            borg_prune_enabled: true
+            borg_prune_glob_archives: "{hostname}-*"
+
+        - name: Fail when missing retention validation unexpectedly passes
+          ansible.builtin.fail:
+            msg: Missing retention validation unexpectedly passed
+      rescue:
+        - name: Assert missing retention validation failed as expected
+          ansible.builtin.assert:
+            that:
+              - >-
+                'Prune requires at least one retention rule'
+                in (ansible_failed_result.msg | default(''))
+
+    - name: Prune with retention only in additional arguments should succeed
+      ansible.builtin.include_role:
+        name: kliwniloc.borgbackup
+      vars:
+        borg_server_host: borg-server
+        borg_server_user_home: /opt/borg
+        borg_decryption_keys_yaml_path: "{{ playbook_dir }}/decryption_keys.yml"
+        borg_repo_name: validation-additional-args
+        borg_backup_argument: validation-additional-args
+        borg_included_dirs:
+          - /etc
+        borg_prune_enabled: true
+        borg_prune_glob_archives: "{hostname}-*"
+        borg_prune_additional_arguments: --keep-last 2
+      register: prune_additional_args_result
+
+    - name: Assert additional-arguments prune validation passes
+      ansible.builtin.assert:
+        that:
+          - prune_additional_args_result is succeeded
+
+    - name: Converge disabled pruning baseline
+      ansible.builtin.include_role:
+        name: kliwniloc.borgbackup
+      vars:
+        borg_server_host: borg-server
+        borg_server_user_home: /opt/borg
+        borg_decryption_keys_yaml_path: "{{ playbook_dir }}/decryption_keys.yml"
+        borg_repo_name: validation-disable-prune
+        borg_backup_argument: validation-disable-prune
+        borg_included_dirs:
+          - /etc
+        borg_prune_enabled: false
+
+    - name: Absent state with empty prune script path should succeed
+      ansible.builtin.include_role:
+        name: kliwniloc.borgbackup
+      vars:
+        state: absent
+        borg_server_host: borg-server
+        borg_server_user_home: /opt/borg
+        borg_decryption_keys_yaml_path: "{{ playbook_dir }}/decryption_keys.yml"
+        borg_repo_name: validation-absent-empty-prune-path
+        borg_backup_argument: validation-absent-empty-prune-path
+        borg_prune_script_location: ""
+      register: absent_empty_prune_path_result
+
+    - name: Assert absent with empty prune script path passes
+      ansible.builtin.assert:
+        that:
+          - absent_empty_prune_path_result is succeeded
diff --git a/molecule/default/molecule.yml b/molecule/default/molecule.yml
index 0ea5c96..68dcf4a 100644
--- a/molecule/default/molecule.yml
+++ b/molecule/default/molecule.yml
@@ -51,6 +51,22 @@ platforms:
     networks:
       - name: molecule-container-net
 
+  - name: borg-client-transition
+    image: ${MOLECULE_DISTRO_CLIENT:-debian:12}
+    dockerfile: ../Dockerfile.j2
+    pre_build_image: false
+    privileged: true
+    networks:
+      - name: molecule-container-net
+
+  - name: borg-client-validation
+    image: ${MOLECULE_DISTRO_CLIENT:-debian:12}
+    dockerfile: ../Dockerfile.j2
+    pre_build_image: false
+    privileged: true
+    networks:
+      - name: molecule-container-net
+
   - name: borg-server
     image: ${MOLECULE_DISTRO_SERVER:-debian:12}
     dockerfile: ../Dockerfile.j2
@@ -71,6 +87,7 @@ provisioner:
   name: ansible
   playbooks:
     converge: ${MOLECULE_PLAYBOOK:-converge.yml}
+    side_effect: side_effect.yml
 
 verifier:
   name: testinfra
diff --git a/molecule/default/side_effect.yml b/molecule/default/side_effect.yml
new file mode 100644
index 0000000..63a10e5
--- /dev/null
+++ b/molecule/default/side_effect.yml
@@ -0,0 +1,167 @@
+---
+- name: Side effect - Configure SSH access for transition tests
+  hosts:
+    - borg-client-transition
+    - borg-client-validation
+
+  tasks:
+    - name: Start ssh on borg-server
+      ansible.builtin.systemd:
+        name: sshd
+        state: started
+      become: true
+      delegate_to: borg-server
+
+    - name: Fetch ssh key for borg-server
+      ansible.builtin.command: >-
+        ssh-keyscan -t rsa borg-server | sed "s/^[^ ]* //"
+      register: borg_server_ssh_keyscan
+      changed_when: false
+
+    - name: Set ssh key for borg-server
+      ansible.builtin.set_fact:
+        borg_server_host_ssh_key: >-
+          {{ borg_server_ssh_keyscan.stdout
+          | split(" ")
+          | reject("search", "borg-server")
+          | join(" ") }}
+
+- name: Side effect - Transition prune from timer to after_backup
+  hosts: borg-client-transition
+
+  tasks:
+    - name: Configure timer-triggered pruning
+      ansible.builtin.include_role:
+        name: kliwniloc.borgbackup
+      vars:
+        borg_server_host: borg-server
+        borg_server_user_home: /opt/borg
+        borg_decryption_keys_yaml_path: "{{ playbook_dir }}/decryption_keys.yml"
+        borg_repo_name: transition-repo
+        borg_backup_argument: transition-repo
+        borg_included_dirs:
+          - /etc
+        borg_excluded_dirs: []
+        borg_prune_enabled: true
+        borg_prune_trigger: timer
+        borg_prune_keep_daily: "2"
+        borg_prune_glob_archives: "{hostname}-*"
+        borg_prune_systemd_oncalendar: "*-*-* 05:15:00"
+
+    - name: Check timer-triggered prune resources
+      ansible.builtin.stat:
+        path: "{{ item }}"
+      loop:
+        - /usr/local/bin/run_borg_prune@transition-repo
+        - /etc/systemd/system/borg_prune@transition-repo.service
+        - /etc/systemd/system/borg_prune@transition-repo.timer
+      register: transition_timer_resources
+
+    - name: Check timer-triggered prune timer is enabled
+      ansible.builtin.systemd:
+        name: borg_prune@transition-repo.timer
+      register: transition_timer_status
+      become: true
+
+    - name: Assert timer-triggered pruning was configured
+      ansible.builtin.assert:
+        that:
+          - transition_timer_resources.results | map(attribute='stat.exists') | list == [true, true, true]
+          - transition_timer_status.status.UnitFileState == 'enabled'
+
+    - name: Reconfigure pruning to run after backup
+      ansible.builtin.include_role:
+        name: kliwniloc.borgbackup
+      vars:
+        borg_server_host: borg-server
+        borg_server_user_home: /opt/borg
+        borg_decryption_keys_yaml_path: "{{ playbook_dir }}/decryption_keys.yml"
+        borg_repo_name: transition-repo
+        borg_backup_argument: transition-repo
+        borg_included_dirs:
+          - /etc
+        borg_excluded_dirs: []
+        borg_prune_enabled: true
+        borg_prune_trigger: after_backup
+        borg_prune_keep_last: "2"
+        borg_prune_glob_archives: "{hostname}-*"
+
+    - name: Check stale prune timer was removed
+      ansible.builtin.stat:
+        path: /etc/systemd/system/borg_prune@transition-repo.timer
+      register: stale_transition_timer
+
+    - name: Read transition backup service
+      ansible.builtin.slurp:
+        src: /etc/systemd/system/borg_backup@transition-repo.service
+      register: transition_backup_service
+      become: true
+
+    - name: Assert after_backup transition completed
+      ansible.builtin.assert:
+        that:
+          - not stale_transition_timer.stat.exists
+          - >-
+            'OnSuccess=borg_prune@transition-repo.service'
+            in (transition_backup_service.content | b64decode)
+
+- name: Side effect - Disable timer-triggered pruning
+  hosts: borg-client-validation
+
+  tasks:
+    - name: Configure timer-triggered pruning before disabling it
+      ansible.builtin.include_role:
+        name: kliwniloc.borgbackup
+      vars:
+        borg_server_host: borg-server
+        borg_server_user_home: /opt/borg
+        borg_decryption_keys_yaml_path: "{{ playbook_dir }}/decryption_keys.yml"
+        borg_repo_name: validation-disable-prune
+        borg_backup_argument: validation-disable-prune
+        borg_included_dirs:
+          - /etc
+        borg_prune_enabled: true
+        borg_prune_trigger: timer
+        borg_prune_keep_last: "2"
+
+    - name: Check prune resources before disabling
+      ansible.builtin.stat:
+        path: "{{ item }}"
+      loop:
+        - /usr/local/bin/run_borg_prune@validation-disable-prune
+        - /etc/systemd/system/borg_prune@validation-disable-prune.service
+        - /etc/systemd/system/borg_prune@validation-disable-prune.timer
+      register: enabled_prune_resources
+
+    - name: Assert prune resources were created
+      ansible.builtin.assert:
+        that:
+          - enabled_prune_resources.results | map(attribute='stat.exists') | list == [true, true, true]
+
+    - name: Disable timer-triggered pruning
+      ansible.builtin.include_role:
+        name: kliwniloc.borgbackup
+      vars:
+        borg_server_host: borg-server
+        borg_server_user_home: /opt/borg
+        borg_decryption_keys_yaml_path: "{{ playbook_dir }}/decryption_keys.yml"
+        borg_repo_name: validation-disable-prune
+        borg_backup_argument: validation-disable-prune
+        borg_included_dirs:
+          - /etc
+        borg_prune_enabled: false
+
+    - name: Check that disabled prune resources were removed
+      ansible.builtin.stat:
+        path: "{{ item }}"
+      loop:
+        - /usr/local/bin/run_borg_prune
+        - /usr/local/bin/run_borg_prune@validation-disable-prune
+        - /etc/systemd/system/borg_prune@validation-disable-prune.service
+        - /etc/systemd/system/borg_prune@validation-disable-prune.timer
+      register: disabled_prune_resources
+
+    - name: Assert disabled prune resources were removed
+      ansible.builtin.assert:
+        that:
+          - disabled_prune_resources.results | map(attribute='stat.exists') | select | list | length == 0
diff --git a/molecule/default/tests/test_client_setup.py b/molecule/default/tests/test_client_setup.py
index 78e751a..231b946 100644
--- a/molecule/default/tests/test_client_setup.py
+++ b/molecule/default/tests/test_client_setup.py
@@ -1,5 +1,7 @@
 """Tests for client setup configuration"""
 
+import shlex
+
 import pytest
 
 testinfra_hosts = [
@@ -8,6 +10,7 @@ testinfra_hosts = [
     "borg-client-multi",
     "borg-client-nonroot",
     "borg-client-multi-keys",
+    "borg-client-transition",
 ]
 
 
@@ -17,6 +20,7 @@ CLIENT_USER_MAP = {
     "borg-client-multi": "root",
     "borg-client-nonroot": "backupuser",
     "borg-client-multi-keys": "root",
+    "borg-client-transition": "root",
 }
 
 CLIENT_SSH_KEY_TYPE_MAP = {
@@ -25,6 +29,7 @@ CLIENT_SSH_KEY_TYPE_MAP = {
     "borg-client-multi": "rsa",
     "borg-client-nonroot": "rsa",
     "borg-client-multi-keys": "ed25519",
+    "borg-client-transition": "rsa",
 }
 
 CLIENT_SSH_KEY_PER_REPO_MAP = {
@@ -33,6 +38,7 @@ CLIENT_SSH_KEY_PER_REPO_MAP = {
     "borg-client-multi": False,
     "borg-client-nonroot": False,
     "borg-client-multi-keys": True,
+    "borg-client-transition": False,
 }
 
 
@@ -56,6 +62,76 @@ def get_client_home(host):
     return f"/home/{user}" if user != "root" else "/root"
 
 
+def get_backup_script_paths(hostname):
+    if hostname == "borg-client-multi":
+        return [
+            "/usr/local/bin/run_borg_backup@configs",
+            "/usr/local/bin/run_borg_backup@home-data",
+        ]
+    if hostname == "borg-client-multi-keys":
+        return [
+            "/usr/local/bin/run_borg_backup@configs-keys",
+            "/usr/local/bin/run_borg_backup@home-data-keys",
+        ]
+    if hostname == "borg-client-nonroot":
+        return ["/usr/local/bin/run_borg_backup@borg-server"]
+    if hostname == "borg-client":
+        return ["/usr/local/bin/run_borg_backup@borg-server"]
+    if hostname == "borg-client-2":
+        return ["/usr/local/bin/run_borg_backup@borg-server-2"]
+    if hostname == "borg-client-transition":
+        return ["/usr/local/bin/run_borg_backup@transition-repo"]
+    return []
+
+
+def get_prune_script_paths(hostname):
+    if hostname == "borg-client":
+        return ["/usr/local/bin/run_borg_prune@borg-server"]
+    if hostname == "borg-client-2":
+        return ["/usr/local/bin/run_borg_prune@borg-server-2"]
+    if hostname == "borg-client-transition":
+        return ["/usr/local/bin/run_borg_prune@transition-repo"]
+    return []
+
+
+def run_script_with_fake_borg(host, script_path, failed_command, failed_status):
+    """Run a generated script with a fake borg and return its result and calls."""
+    temp_dir_result = host.run("mktemp -d /tmp/borg-exit-test.XXXXXX")
+    assert temp_dir_result.rc == 0
+    temp_dir = temp_dir_result.stdout.strip()
+    fake_borg_path = f"{temp_dir}/borg"
+    log_path = f"{temp_dir}/calls"
+
+    fake_borg = """#!/bin/bash
+printf '%s\\n' "$1" >> "${BORG_TEST_LOG}"
+if [ "$1" = "${BORG_FAIL_COMMAND}" ]; then
+  exit "${BORG_FAIL_STATUS}"
+fi
+exit 0
+"""
+    setup = host.run(
+        f"cat > {shlex.quote(fake_borg_path)} <<'EOF'\n"
+        f"{fake_borg}"
+        "EOF\n"
+        f"chmod 0755 {shlex.quote(fake_borg_path)}"
+    )
+    assert setup.rc == 0
+
+    try:
+        result = host.run(
+            "env "
+            f"BORG_FAIL_COMMAND={shlex.quote(failed_command)} "
+            f"BORG_FAIL_STATUS={failed_status} "
+            f"BORG_TEST_LOG={shlex.quote(log_path)} "
+            f"PATH={shlex.quote(temp_dir)}:$PATH "
+            f"{shlex.quote(script_path)}"
+        )
+        calls = host.file(log_path).content_string.splitlines()
+        return result, calls
+    finally:
+        host.run(f"rm -rf {shlex.quote(temp_dir)}")
+
+
 class TestSSHSetup:
     def test_ssh_directory_exists(self, host):
         client_home = get_client_home(host)
@@ -162,8 +238,8 @@ class TestBackupScript:
             assert script.user == client_user
             assert script.group == client_user
             assert script.mode == 0o711
-        elif hostname in ("borg-client", "borg-client-2"):
-            script = host.file("/usr/local/bin/run_borg_backup")
+        elif hostname in ("borg-client", "borg-client-2", "borg-client-transition"):
+            script = host.file(get_backup_script_paths(hostname)[0])
             assert script.exists
             assert script.user == client_user
             assert script.group == client_user
@@ -187,12 +263,156 @@ class TestBackupScript:
         elif hostname == "borg-client-nonroot":
             script = host.file("/usr/local/bin/run_borg_backup@borg-server")
             assert script.contains("borg create")
-        elif hostname in ("borg-client", "borg-client-2"):
-            script = host.file("/usr/local/bin/run_borg_backup")
+        elif hostname in ("borg-client", "borg-client-2", "borg-client-transition"):
+            script = host.file(get_backup_script_paths(hostname)[0])
             assert script.contains("borg create")
+            assert not script.contains("borg prune")
+            assert not script.contains("borg compact")
         else:
             pytest.fail(f"Unexpected hostname: {hostname}")
 
+
+class TestPruneScript:
+    def test_prune_script_created_when_pruning_enabled(self, host):
+        hostname = host.backend.get_hostname()
+        if hostname not in ("borg-client", "borg-client-2", "borg-client-transition"):
+            return
+
+        for script_path in get_prune_script_paths(hostname):
+            script = host.file(script_path)
+            assert script.exists
+            assert script.mode == 0o711
+            assert script.contains("borg prune")
+            assert script.contains("borg compact")
+
+    def test_scripts_have_valid_shell_syntax(self, host):
+        hostname = host.backend.get_hostname()
+
+        for script_path in get_backup_script_paths(hostname) + get_prune_script_paths(
+            hostname
+        ):
+            result = host.run(f"bash -n {script_path}")
+            assert result.rc == 0
+            assert "ANSIBLE MANAGED BLOCK" not in host.file(script_path).content_string
+
+    def test_after_backup_prune_script_flow_and_options(self, host):
+        hostname = host.backend.get_hostname()
+        if hostname != "borg-client":
+            return
+
+        backup = host.file("/usr/local/bin/run_borg_backup@borg-server")
+        prune = host.file("/usr/local/bin/run_borg_prune@borg-server")
+        content = prune.content_string
+
+        assert "borg create" in backup.content_string
+        assert "borg prune" not in backup.content_string
+        assert "borg compact" not in backup.content_string
+
+        prune_index = content.index("borg prune")
+        prune_status_index = content.index("borg_prune_exit=$?")
+        prune_guard_index = content.index('if [ "${borg_prune_exit}" -ne 0 ]; then')
+        compact_index = content.index("borg compact")
+
+        assert prune_index < prune_status_index < prune_guard_index < compact_index
+        assert "--checkpoint-interval 1800" in content
+        assert "--glob-archives '{hostname}-*'" in content
+        assert "--keep-last 2" in content
+        assert "--stats" in content
+        assert "--threshold 1" in content
+
+    def test_timer_prune_script_flow_and_options(self, host):
+        hostname = host.backend.get_hostname()
+        if hostname != "borg-client-2":
+            return
+
+        backup_script = host.file("/usr/local/bin/run_borg_backup@borg-server-2")
+        prune_script = host.file("/usr/local/bin/run_borg_prune@borg-server-2")
+        prune_content = prune_script.content_string
+
+        assert "borg prune" not in backup_script.content_string
+        assert "borg compact" not in backup_script.content_string
+
+        prune_index = prune_content.index("borg prune")
+        prune_status_index = prune_content.index("borg_prune_exit=$?")
+        prune_guard_index = prune_content.index(
+            'if [ "${borg_prune_exit}" -ne 0 ]; then'
+        )
+        compact_index = prune_content.index("borg compact")
+
+        assert prune_index < prune_status_index < prune_guard_index < compact_index
+        assert "--checkpoint-interval 1800" in prune_content
+        assert "--glob-archives '{hostname}-*'" in prune_content
+        assert "--keep-daily 7" in prune_content
+        assert "--keep-weekly 4" in prune_content
+        assert "--stats" in prune_content
+        assert "--threshold 10" in prune_content
+
+    def test_transition_host_uses_separate_prune_script(self, host):
+        hostname = host.backend.get_hostname()
+        if hostname != "borg-client-transition":
+            return
+
+        backup = host.file("/usr/local/bin/run_borg_backup@transition-repo")
+        prune = host.file("/usr/local/bin/run_borg_prune@transition-repo")
+
+        assert "borg create" in backup.content_string
+        assert "borg prune" not in backup.content_string
+        assert "borg prune" in prune.content_string
+        assert "borg compact" in prune.content_string
+        assert "--keep-last 2" in prune.content_string
+        assert "--glob-archives '{hostname}-*'" in prune.content_string
+
+    def test_backup_propagates_create_failure_status(self, host):
+        if host.backend.get_hostname() != "borg-client":
+            return
+
+        result, calls = run_script_with_fake_borg(
+            host,
+            "/usr/local/bin/run_borg_backup@borg-server",
+            "create",
+            42,
+        )
+
+        assert result.rc == 42
+        assert calls == ["create"]
+
+    def test_after_backup_preserves_borg_warning_status(self, host):
+        if host.backend.get_hostname() != "borg-client":
+            return
+
+        result, calls = run_script_with_fake_borg(
+            host,
+            "/usr/local/bin/run_borg_backup@borg-server",
+            "create",
+            1,
+        )
+
+        assert result.rc == 1
+        assert calls == ["create"]
+
+    @pytest.mark.parametrize(
+        ("failed_command", "failed_status", "expected_calls"),
+        [
+            ("prune", 43, ["prune"]),
+            ("compact", 44, ["prune", "compact"]),
+        ],
+    )
+    def test_timer_prune_propagates_failure_status(
+        self, host, failed_command, failed_status, expected_calls
+    ):
+        if host.backend.get_hostname() != "borg-client-2":
+            return
+
+        result, calls = run_script_with_fake_borg(
+            host,
+            "/usr/local/bin/run_borg_prune@borg-server-2",
+            failed_command,
+            failed_status,
+        )
+
+        assert result.rc == failed_status
+        assert calls == expected_calls
+
     def test_backup_script_contains_compression(self, host):
         hostname = host.backend.get_hostname()
 
@@ -206,18 +426,15 @@ class TestBackupScript:
             script2 = host.file("/usr/local/bin/run_borg_backup@home-data-keys")
             assert script1.contains("-C zstd")
             assert script2.contains("-C lz4")
-        elif hostname == "borg-client-2":
-            script = host.file("/usr/local/bin/run_borg_backup")
-            assert script.contains("-C")
-            assert script.contains("lz4")
-        elif hostname in ("borg-client", "borg-client-nonroot"):
-            script = (
-                host.file("/usr/local/bin/run_borg_backup")
-                if hostname == "borg-client"
-                else host.file("/usr/local/bin/run_borg_backup@borg-server")
-            )
+        elif hostname in (
+            "borg-client",
+            "borg-client-2",
+            "borg-client-nonroot",
+            "borg-client-transition",
+        ):
+            script = host.file(get_backup_script_paths(hostname)[0])
             assert script.contains("-C")
-            assert script.contains("zstd")
+            assert script.contains("lz4" if hostname == "borg-client-2" else "zstd")
         else:
             pytest.fail(f"Unexpected hostname: {hostname}")
 
@@ -243,13 +460,17 @@ class TestBackupScript:
             assert script.contains("borg@borg-server")
             assert script.contains("/opt/borg")
         elif hostname == "borg-client":
-            script = host.file("/usr/local/bin/run_borg_backup")
+            script = host.file(get_backup_script_paths(hostname)[0])
             assert script.contains("borg@borg-server")
             assert script.contains("/opt/borg")
         elif hostname == "borg-client-2":
-            script = host.file("/usr/local/bin/run_borg_backup")
+            script = host.file(get_backup_script_paths(hostname)[0])
             assert script.contains("backupserver@borg-server-2")
             assert script.contains("/var/backups")
+        elif hostname == "borg-client-transition":
+            script = host.file("/usr/local/bin/run_borg_backup@transition-repo")
+            assert script.contains("borg@borg-server")
+            assert script.contains("/opt/borg/transition-repo")
         else:
             pytest.fail(f"Unexpected hostname: {hostname}")
 
@@ -272,8 +493,8 @@ class TestBackupScript:
             script = host.file("/usr/local/bin/run_borg_backup@borg-server")
             content = script.content_string
             assert "/etc" in content
-        elif hostname in ("borg-client", "borg-client-2"):
-            script = host.file("/usr/local/bin/run_borg_backup")
+        elif hostname in ("borg-client", "borg-client-2", "borg-client-transition"):
+            script = host.file(get_backup_script_paths(hostname)[0])
             content = script.content_string
             assert "/etc" in content or "/home" in content
         else:
@@ -295,8 +516,8 @@ class TestBackupScript:
         elif hostname == "borg-client-nonroot":
             script = host.file("/usr/local/bin/run_borg_backup@borg-server")
             assert script.mode == 0o711
-        elif hostname in ("borg-client", "borg-client-2"):
-            script = host.file("/usr/local/bin/run_borg_backup")
+        elif hostname in ("borg-client", "borg-client-2", "borg-client-transition"):
+            script = host.file(get_backup_script_paths(hostname)[0])
             assert script.mode == 0o711
         else:
             pytest.fail(f"Unexpected hostname: {hostname}")
@@ -313,7 +534,7 @@ class TestBackupScript:
             assert "--one-file-system" in script2.content_string
             assert "--exclude-caches" in script2.content_string
         elif hostname == "borg-client-2":
-            script = host.file("/usr/local/bin/run_borg_backup")
+            script = host.file(get_backup_script_paths(hostname)[0])
             content = script.content_string
             assert "-C zlib,6" in content
 
@@ -322,7 +543,7 @@ class TestBackupScript:
         if hostname != "borg-client-2":
             return
 
-        script = host.file("/usr/local/bin/run_borg_backup")
+        script = host.file(get_backup_script_paths(hostname)[0])
         content = script.content_string
 
         assert "-C lz4" in content
@@ -344,57 +565,15 @@ class TestBackupScript:
             assert "ssh -i" in script2.content_string
 
 
-class TestMultiInstanceBaseScript:
-    def test_base_script_exists(self, host):
-        hostname = host.backend.get_hostname()
-        if hostname != "borg-client-multi":
-            return
-
-        client_user = get_client_user(host)
-        base_script = host.file("/usr/local/bin/run_borg_backup")
-        assert base_script.exists
-        assert base_script.user == client_user
-        assert base_script.mode == 0o711
-
-    def test_base_script_contains_both_blocks(self, host):
-        hostname = host.backend.get_hostname()
-        if hostname != "borg-client-multi":
-            return
-
-        base_script = host.file("/usr/local/bin/run_borg_backup")
-        content = base_script.content_string
+class TestAggregateScripts:
+    def test_legacy_aggregate_backup_script_is_preserved(self, host):
+        aggregate = host.file("/usr/local/bin/run_borg_backup")
 
-        assert "borg-server/configs" in content
-        assert "borg-server/home-data" in content
-
-    def test_base_script_contains_both_repos(self, host):
-        hostname = host.backend.get_hostname()
-        if hostname != "borg-client-multi":
-            return
-
-        base_script = host.file("/usr/local/bin/run_borg_backup")
-        content = base_script.content_string
-
-        assert "/opt/borg/configs" in content
-        assert "/opt/borg/home-data" in content
-
-    def test_base_script_contains_both_compressions(self, host):
-        hostname = host.backend.get_hostname()
-        if hostname != "borg-client-multi":
-            return
-
-        base_script = host.file("/usr/local/bin/run_borg_backup")
-        content = base_script.content_string
-
-        assert "-C zstd" in content
-        assert "-C lz4" in content
-
-    def test_base_script_two_borg_create_commands(self, host):
-        hostname = host.backend.get_hostname()
-        if hostname != "borg-client-multi":
-            return
-
-        base_script = host.file("/usr/local/bin/run_borg_backup")
-        content = base_script.content_string
+        if host.backend.get_hostname() == "borg-client-transition":
+            assert aggregate.exists
+            assert "legacy-backup-block" in aggregate.content_string
+        else:
+            assert not aggregate.exists
 
-        assert content.count("borg create") == 2
+    def test_aggregate_prune_script_is_not_created(self, host):
+        assert not host.file("/usr/local/bin/run_borg_prune").exists
diff --git a/molecule/default/tests/test_manual_backup.py b/molecule/default/tests/test_manual_backup.py
index 8cd044c..ea48655 100644
--- a/molecule/default/tests/test_manual_backup.py
+++ b/molecule/default/tests/test_manual_backup.py
@@ -84,3 +84,125 @@ def test_backup_restore(host, compression):
     assert c.rc == 0
     assert c.stdout == ""
     assert c.stderr == ""
+
+
+def test_after_backup_scripts_prune_old_archives(host):
+    hostname = host.backend.get_hostname()
+    if hostname != "borg-client":
+        return
+
+    server_user, server_host, server_path = get_server_info(hostname)
+    backup_script_path = "/usr/local/bin/run_borg_backup@borg-server"
+    prune_script_path = "/usr/local/bin/run_borg_prune@borg-server"
+
+    original_backup_script = host.file(backup_script_path).content_string
+    original_prune_script = host.file(prune_script_path).content_string
+    prefix = f"testprune-{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
+
+    modified_backup_script = original_backup_script.replace(
+        "::{hostname}-{now:%Y-%m-%dT%H:%M:%S}",
+        f"::{prefix}-${{BORG_TEST_ITERATION}}-{{now:%Y-%m-%dT%H:%M:%S}}",
+    )
+    modified_prune_script = original_prune_script.replace(
+        "{hostname}-*",
+        f"{prefix}-*",
+    )
+
+    rewrite = host.run(
+        "python3 - <<'PY'\n"
+        "from pathlib import Path\n"
+        f"Path({backup_script_path!r}).write_text({modified_backup_script!r})\n"
+        f"Path({prune_script_path!r}).write_text({modified_prune_script!r})\n"
+        "PY"
+    )
+    assert rewrite.rc == 0
+    assert host.run(f"bash -n {backup_script_path}").rc == 0
+    assert host.run(f"bash -n {prune_script_path}").rc == 0
+
+    try:
+        for iteration in range(4):
+            backup = host.run(f"BORG_TEST_ITERATION={iteration} {backup_script_path}")
+            assert backup.rc == 0
+            prune = host.run(prune_script_path)
+            assert prune.rc == 0
+
+        archives = host.run(
+            f"borg list {server_user}@{server_host}:{server_path}/{hostname} --glob-archives '{prefix}-*'"
+        )
+        assert archives.rc == 0
+
+        archive_lines = [line for line in archives.stdout.splitlines() if line.strip()]
+        assert len(archive_lines) == 2
+    finally:
+        restore = host.run(
+            "python3 - <<'PY'\n"
+            "from pathlib import Path\n"
+            f"Path({backup_script_path!r}).write_text({original_backup_script!r})\n"
+            f"Path({prune_script_path!r}).write_text({original_prune_script!r})\n"
+            "PY"
+        )
+        assert restore.rc == 0
+
+
+def test_manual_prune_script_prunes_old_archives(host):
+    hostname = host.backend.get_hostname()
+    if hostname != "borg-client-2":
+        return
+
+    server_user, server_host, server_path = get_server_info(hostname)
+    prefix = f"testprune-timer-{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
+    script_path = "/usr/local/bin/run_borg_prune@borg-server-2"
+
+    for iteration in range(4):
+        create = host.run(
+            f"borg create -C lz4 {server_user}@{server_host}:{server_path}/{hostname}::{prefix}-{iteration}-{{now:%Y-%m-%dT%H:%M:%S}} /etc"
+        )
+        assert create.rc == 0
+
+    original_script = host.file(script_path).content_string
+    assert "--keep-daily 7" in original_script
+    assert "--keep-weekly 4" in original_script
+
+    modified_script = (
+        original_script.replace("{hostname}-*", f"{prefix}-*")
+        .replace("--keep-daily 7", "--keep-last 2")
+        .replace("--keep-weekly 4", "")
+    )
+    assert "--keep-last 2" in modified_script
+    assert "--keep-daily 7" not in modified_script
+    assert "--keep-weekly 4" not in modified_script
+
+    rewrite = host.run(
+        "python3 - <<'PY'\n"
+        "from pathlib import Path\n"
+        f"Path({script_path!r}).write_text({modified_script!r})\n"
+        "PY"
+    )
+    assert rewrite.rc == 0
+    syntax_check = host.run(f"bash -n {script_path}")
+    assert syntax_check.rc == 0
+
+    try:
+        result = host.run(script_path)
+        assert result.rc == 0
+
+        archives = host.run(
+            f"borg list {server_user}@{server_host}:{server_path}/{hostname} --glob-archives '{prefix}-*'"
+        )
+        assert archives.rc == 0
+
+        archive_lines = [line for line in archives.stdout.splitlines() if line.strip()]
+        assert len(archive_lines) == 2
+    finally:
+        restore = host.run(
+            "python3 - <<'PY'\n"
+            "from pathlib import Path\n"
+            f"Path({script_path!r}).write_text({original_script!r})\n"
+            "PY"
+        )
+        assert restore.rc == 0
+
+        cleanup = host.run(
+            f"borg delete --glob-archives '{prefix}-*' {server_user}@{server_host}:{server_path}/{hostname}"
+        )
+        assert cleanup.rc == 0
diff --git a/molecule/default/tests/test_server_setup.py b/molecule/default/tests/test_server_setup.py
index 93c319d..9a5156b 100644
--- a/molecule/default/tests/test_server_setup.py
+++ b/molecule/default/tests/test_server_setup.py
@@ -73,6 +73,8 @@ class TestBorgSSHSetup:
             "borg-client-multi",
             "borg-client-nonroot",
             "borg-client-multi-keys",
+            "borg-client-transition",
+            "borg-client-validation",
         )
         for line in content.split("\n"):
             if not line.strip():
diff --git a/molecule/default/tests/test_systemd.py b/molecule/default/tests/test_systemd.py
index e68e232..297ae0d 100644
--- a/molecule/default/tests/test_systemd.py
+++ b/molecule/default/tests/test_systemd.py
@@ -8,6 +8,7 @@ testinfra_hosts = [
     "borg-client-multi",
     "borg-client-nonroot",
     "borg-client-multi-keys",
+    "borg-client-transition",
 ]
 
 CLIENT_CONFIGS = {
@@ -16,18 +17,33 @@ CLIENT_CONFIGS = {
         "server": "borg-server",
         "schedule": "*-*-* 02:00:00",
         "success_exit_status": False,
+        "prune_enabled": True,
+        "prune_timer": False,
     },
     "borg-client-2": {
         "user": "root",
         "server": "borg-server-2",
         "schedule": "*-*-* 03:00:00",
         "success_exit_status": True,
+        "prune_enabled": True,
+        "prune_timer": True,
+        "prune_schedule": "*-*-* 05:00:00",
     },
     "borg-client-nonroot": {
         "user": "backupuser",
         "server": "borg-server",
         "schedule": "*-*-* 02:00:00",
         "success_exit_status": False,
+        "prune_enabled": False,
+        "prune_timer": False,
+    },
+    "borg-client-transition": {
+        "user": "root",
+        "server": "transition-repo",
+        "schedule": "*-*-* 02:00:00",
+        "success_exit_status": False,
+        "prune_enabled": True,
+        "prune_timer": False,
     },
 }
 
@@ -85,7 +101,19 @@ class TestSystemdServiceFile:
             assert service.contains("[Service]")
             assert service.contains("[Install]")
             assert service.contains("Type=oneshot")
-            assert service.contains("ExecStart=/usr/local/bin/run_borg_backup")
+            assert service.contains(
+                f"ExecStart=/usr/local/bin/run_borg_backup@{config['server']}"
+            )
+            assert (
+                "ExecStart=/usr/local/bin/run_borg_backup\n"
+                not in service.content_string
+            )
+            if config["prune_enabled"] and not config["prune_timer"]:
+                assert service.contains(
+                    f"OnSuccess=borg_prune@{config['server']}.service"
+                )
+            else:
+                assert not service.contains("OnSuccess=")
         else:
             for repo in config["repos"]:
                 service = host.file(f"/etc/systemd/system/borg_backup@{repo}.service")
@@ -93,6 +121,9 @@ class TestSystemdServiceFile:
                 assert service.contains("[Service]")
                 assert service.contains("[Install]")
                 assert service.contains("Type=oneshot")
+                assert service.contains(
+                    f"ExecStart=/usr/local/bin/run_borg_backup@{repo}"
+                )
 
     def test_service_user(self, host, config):
         if config["type"] == "single":
@@ -193,3 +224,54 @@ class TestSystemdState:
     def test_daemon_reload_ok(self, host, config):
         c = host.run("systemctl daemon-reload")
         assert c.rc == 0
+
+
+class TestPruneSystemd:
+    def test_prune_service_created_when_pruning_enabled(self, host, config):
+        if config["type"] != "single":
+            return
+
+        service = host.file(
+            f"/etc/systemd/system/borg_prune@{config['server']}.service"
+        )
+        assert service.exists == config["prune_enabled"]
+        if config["prune_enabled"]:
+            assert (
+                f"ExecStart=/usr/local/bin/run_borg_prune@{config['server']}"
+                in service.content_string
+            )
+
+    def test_prune_timer_created_when_configured(self, host, config):
+        if config["type"] != "single" or not config.get("prune_timer"):
+            return
+
+        timer = host.file(f"/etc/systemd/system/borg_prune@{config['server']}.timer")
+        service = host.file(
+            f"/etc/systemd/system/borg_prune@{config['server']}.service"
+        )
+        assert timer.exists
+        assert service.exists
+        assert f"OnCalendar={config['prune_schedule']}" in timer.content_string
+        assert (
+            f"ExecStart=/usr/local/bin/run_borg_prune@{config['server']}"
+            in service.content_string
+        )
+        assert "ExecStart=/usr/local/bin/run_borg_prune\n" not in service.content_string
+        assert "SuccessExitStatus=1 TEMPFAIL" in service.content_string
+
+    def test_prune_timer_not_created_for_after_backup(self, host, config):
+        if config["type"] != "single" or config.get("prune_timer"):
+            return
+
+        timer = host.file(f"/etc/systemd/system/borg_prune@{config['server']}.timer")
+        assert not timer.exists
+
+    def test_transition_host_stale_prune_timer_removed(self, host):
+        hostname = host.backend.get_hostname()
+        if hostname != "borg-client-transition":
+            return
+
+        timer = host.file("/etc/systemd/system/borg_prune@transition-repo.timer")
+        service = host.file("/etc/systemd/system/borg_prune@transition-repo.service")
+        assert not timer.exists
+        assert service.exists
diff --git a/molecule/delete/converge.yml b/molecule/delete/converge.yml
index 00de93d..037e56a 100644
--- a/molecule/delete/converge.yml
+++ b/molecule/delete/converge.yml
@@ -35,6 +35,10 @@
         borg_backup_argument: single-backup
         borg_dangerously_delete_backups: true
         borg_decryption_keys_yaml_path: "{{ playbook_dir }}/decryption_keys_delete.yml"
+        borg_prune_enabled: true
+        borg_prune_trigger: timer
+        borg_prune_keep_last: "2"
+        borg_prune_glob_archives: "{hostname}-*"
 
 - name: Delete one repo from multi-instance (keep data)
   hosts: borg-client-multi-delete
@@ -135,6 +139,10 @@
         borg_ssh_key_type: "{{ 'ed25519' if inventory_hostname == 'borg-client-per-repo-delete' else 'rsa' }}"
         borg_dangerously_delete_backups: >-
           {{ inventory_hostname in ['borg-client-single-delete', 'borg-client-per-repo-delete'] }}
+        borg_prune_enabled: "{{ inventory_hostname == 'borg-client-single-delete' }}"
+        borg_prune_trigger: "{{ 'timer' if inventory_hostname == 'borg-client-single-delete' else 'after_backup' }}"
+        borg_prune_keep_last: "{{ '2' if inventory_hostname == 'borg-client-single-delete' else '' }}"
+        borg_prune_glob_archives: "{{ '{hostname}-*' if inventory_hostname == 'borg-client-single-delete' else '{hostname}-*' }}"
         borg_decryption_keys_yaml_path: >-
           {{
             (playbook_dir ~ '/decryption_keys_empty_delete.yml')
diff --git a/molecule/delete/prepare.yml b/molecule/delete/prepare.yml
index 98a94a1..430387d 100644
--- a/molecule/delete/prepare.yml
+++ b/molecule/delete/prepare.yml
@@ -44,6 +44,10 @@
         borg_included_dirs:
           - /etc
         borg_excluded_dirs: []
+        borg_prune_enabled: true
+        borg_prune_trigger: timer
+        borg_prune_keep_last: "2"
+        borg_prune_glob_archives: "{hostname}-*"
 
 - name: Create multi-instance backup (shared key)
   hosts: borg-client-multi-delete
diff --git a/molecule/delete/tests/test_delete.py b/molecule/delete/tests/test_delete.py
index 4c0d859..3fc3189 100644
--- a/molecule/delete/tests/test_delete.py
+++ b/molecule/delete/tests/test_delete.py
@@ -37,6 +37,16 @@ class TestSingleRepoDelete:
         assert not timer.exists
         assert not service.exists
 
+    def test_prune_systemd_units_removed(self, host):
+        hostname = host.backend.get_hostname()
+        if hostname != "borg-client-single-delete":
+            return
+
+        timer = host.file("/etc/systemd/system/borg_prune@single-backup.timer")
+        service = host.file("/etc/systemd/system/borg_prune@single-backup.service")
+        assert not timer.exists
+        assert not service.exists
+
     def test_backup_scripts_removed(self, host):
         hostname = host.backend.get_hostname()
         if hostname != "borg-client-single-delete":
@@ -47,6 +57,16 @@ class TestSingleRepoDelete:
         assert not repo_script.exists
         assert not base_script.exists
 
+    def test_prune_scripts_removed(self, host):
+        hostname = host.backend.get_hostname()
+        if hostname != "borg-client-single-delete":
+            return
+
+        repo_script = host.file("/usr/local/bin/run_borg_prune@single-backup")
+        base_script = host.file("/usr/local/bin/run_borg_prune")
+        assert not repo_script.exists
+        assert not base_script.exists
+
     def test_shared_ssh_key_kept(self, host):
         hostname = host.backend.get_hostname()
         if hostname != "borg-client-single-delete":
@@ -98,17 +118,12 @@ class TestMultiInstanceDelete:
         assert service.exists
         assert script.exists
 
-    def test_base_script_keeps_only_remaining_block(self, host):
+    def test_aggregate_script_is_absent(self, host):
         hostname = host.backend.get_hostname()
         if hostname != "borg-client-multi-delete":
             return
 
-        base_script = host.file("/usr/local/bin/run_borg_backup")
-        assert base_script.exists
-        content = base_script.content_string
-        assert "/opt/borg/multi-repo-b" in content
-        assert "/opt/borg/multi-repo-a" not in content
-        assert content.count("borg create") == 1
+        assert not host.file("/usr/local/bin/run_borg_backup").exists
 
     def test_shared_ssh_key_kept(self, host):
         hostname = host.backend.get_hostname()
@@ -168,13 +183,10 @@ class TestPerRepoKeyDelete:
         timer = host.file("/etc/systemd/system/borg_backup@per-repo-b.timer")
         service = host.file("/etc/systemd/system/borg_backup@per-repo-b.service")
         script = host.file("/usr/local/bin/run_borg_backup@per-repo-b")
-        base_script = host.file("/usr/local/bin/run_borg_backup")
         assert timer.exists
         assert service.exists
         assert script.exists
-        assert base_script.exists
-        assert "/opt/borg/per-repo-b" in base_script.content_string
-        assert "/opt/borg/per-repo-a" not in base_script.content_string
+        assert not host.file("/usr/local/bin/run_borg_backup").exists
 
     def test_per_repo_ssh_key_removed(self, host):
         hostname = host.backend.get_hostname()