diff --git a/.gitignore b/.gitignore
index d2a60eff..24c54270 100644
--- a/.gitignore
+++ b/.gitignore
@@ -14,6 +14,7 @@ tests/integration/*
 !tests/integration/elasticsearch_template_contract.yml
 !tests/integration/elasticsearch_template_inventory.ini
 !tests/integration/elasticsearch_upgrade_detection_contract.yml
+!tests/integration/node_maintenance_contract.yml
 !tests/integration/rolling_restart_contract.yml
 !tests/integration/rolling_restart_inventory.ini
 site/
diff --git a/docs/reference/elasticsearch.md b/docs/reference/elasticsearch.md
index 8c642b45..f9f74078 100644
--- a/docs/reference/elasticsearch.md
+++ b/docs/reference/elasticsearch.md
@@ -439,6 +439,33 @@ elasticsearch_config_restart_node_delay: 3
 
 `elasticsearch_config_restart_node_retries` and `elasticsearch_config_restart_node_delay` control how long the role waits for the node it just restarted to rejoin the cluster. Defaults give ~10 minutes per node (200 × 3s).
 
+### Node maintenance entry points
+
+For taking a node down outside the role's own upgrade and restart flows (OS updates, reboots, storage work), the role exposes two task entry points that follow the same drain procedure Elastic prescribes:
+
+```yaml
+- name: Drain this node
+  ansible.builtin.include_role:
+    name: oddly.elasticstack.elasticsearch
+    tasks_from: node_maintenance_start
+  vars:
+    elasticsearch_maintenance_password: "{{ elastic_password }}"
+
+# ... stop the service, patch, reboot, start the service ...
+
+- name: Restore cluster state
+  ansible.builtin.include_role:
+    name: oddly.elasticstack.elasticsearch
+    tasks_from: node_maintenance_end
+  vars:
+    elasticsearch_maintenance_password: "{{ elastic_password }}"
+    # Wait matches on node.name, which the role sets from elasticsearch_nodename
+    # (defaults to the short hostname, may differ from inventory_hostname).
+    elasticsearch_maintenance_wait_for_node: "{{ elasticsearch_nodename | default(inventory_hostname) }}"
+```
+
+`node_maintenance_start` waits for cluster health, excludes the node from voting, sets allocation to primaries-only, enables ML upgrade mode, optionally applies `elasticsearch_drain_cluster_settings` (a recovery throughput boost for the drain window) and flushes. `node_maintenance_end` reverses all of it — restoring every boosted key to its baseline in `elasticsearch_cluster_settings` — waits for the node to rejoin when `elasticsearch_maintenance_wait_for_node` is set, and gates on `elasticsearch_maintenance_wait_status`. Restore steps are best-effort, so `node_maintenance_end` belongs in an `always` block and doubles as a defensive state reset (`elasticsearch_maintenance_wait_health: false` skips the health gate for that use).
+
 ### Rolling Upgrades
 
 The role validates the upgrade path before any work begins. When `elasticstack_release` is 9 or higher and Elasticsearch is currently installed, the role checks that the installed version is at least 8.19.0. If it finds an older 8.x version, the play fails immediately -- you must step through 8.19.x first. This matches [Elastic's official upgrade requirements](https://www.elastic.co/docs/deploy-manage/upgrade/deployment-or-cluster).
diff --git a/roles/elasticsearch/defaults/main.yml b/roles/elasticsearch/defaults/main.yml
index 3bd09746..e536e98a 100644
--- a/roles/elasticsearch/defaults/main.yml
+++ b/roles/elasticsearch/defaults/main.yml
@@ -98,6 +98,23 @@ elasticsearch_upgrade_wait_status: green
 elasticsearch_upgrade_health_retries: 100
 # @var elasticsearch_upgrade_health_delay:description: Delay in seconds between cluster health polling attempts during a rolling upgrade
 elasticsearch_upgrade_health_delay: 30
+# @var elasticsearch_maintenance_wait_status:description: Minimum cluster health status the node maintenance entry points wait for. Use yellow or green
+elasticsearch_maintenance_wait_status: green
+# @var elasticsearch_maintenance_health_retries:description: Number of cluster health polling attempts in the node maintenance entry points
+elasticsearch_maintenance_health_retries: 60
+# @var elasticsearch_maintenance_health_delay:description: Delay in seconds between cluster health polling attempts in the node maintenance entry points
+elasticsearch_maintenance_health_delay: 30
+# @var elasticsearch_maintenance_wait_health:description: Wait for cluster health at the end of node maintenance. Disable for a defensive state reset at the start of a run
+elasticsearch_maintenance_wait_health: true
+# @var elasticsearch_maintenance_require_green:description: Fail node_maintenance_end unless the cluster returns to green. Default accepts yellow
+elasticsearch_maintenance_require_green: false
+# @var elasticsearch_drain_cluster_settings:description: >
+#   Persistent cluster settings applied while a node is drained for maintenance
+#   (typically a recovery throughput boost). node_maintenance_end restores every
+#   key listed here to its value in elasticsearch_cluster_settings, or removes
+#   it when no baseline is declared there.
+# @end
+elasticsearch_drain_cluster_settings: {}
 
 # @var elasticsearch_jvm_custom_parameters:description: Additional JVM parameters appended to jvm.options.d. Use for GC tuning, debug flags, etc
 # @var elasticsearch_jvm_custom_parameters:example: >
diff --git a/roles/elasticsearch/meta/argument_specs.yml b/roles/elasticsearch/meta/argument_specs.yml
index baf02814..53b27658 100644
--- a/roles/elasticsearch/meta/argument_specs.yml
+++ b/roles/elasticsearch/meta/argument_specs.yml
@@ -340,3 +340,76 @@ argument_specs:
           set to true from a playbook to force renewal regardless of buffer.
         type: bool
         default: false
+  node_maintenance_start:
+    short_description: Prepare the cluster for taking this node down
+    description: Health gate, voting exclusion, primaries-only allocation, ML upgrade mode, optional recovery boost and
+      flush. Pair with node_maintenance_end in an always block so cluster state is restored even when the maintenance
+      itself fails.
+    options:
+      elasticsearch_maintenance_password:
+        description: Password for the elastic user, used for all API calls in the entry point.
+        type: str
+        required: true
+        no_log: true
+      elasticsearch_maintenance_api_url:
+        description: Base URL of the cluster API. Defaults to the role's protocol, API host and HTTP port.
+        type: str
+      elasticsearch_maintenance_wait_status:
+        description: Minimum cluster health status to wait for before draining. Use yellow or green.
+        type: str
+        default: green
+        choices: [green, yellow]
+      elasticsearch_maintenance_health_retries:
+        description: Number of cluster health polling attempts.
+        type: int
+        default: 60
+      elasticsearch_maintenance_health_delay:
+        description: Delay in seconds between cluster health polling attempts.
+        type: int
+        default: 30
+      elasticsearch_drain_cluster_settings:
+        description: Persistent cluster settings applied while the node is drained, typically a recovery throughput boost.
+        type: dict
+        default: {}
+  node_maintenance_end:
+    short_description: Restore cluster state after node maintenance
+    description: Re-enables allocation, disables ML upgrade mode, clears voting exclusions, restores boosted recovery
+      settings to their baseline from elasticsearch_cluster_settings, optionally waits for the node to rejoin, and gates
+      on cluster health. All restore steps are best-effort so the entry point is safe to run defensively.
+    options:
+      elasticsearch_maintenance_password:
+        description: Password for the elastic user, used for all API calls in the entry point.
+        type: str
+        required: true
+        no_log: true
+      elasticsearch_maintenance_api_url:
+        description: Base URL of the cluster API. Defaults to the role's protocol, API host and HTTP port.
+        type: str
+      elasticsearch_maintenance_wait_for_node:
+        description: Node name to wait for in _cat/nodes before the health gate. Empty skips the check.
+        type: str
+      elasticsearch_maintenance_wait_status:
+        description: Minimum cluster health status to wait for after maintenance. Use yellow or green.
+        type: str
+        default: green
+        choices: [green, yellow]
+      elasticsearch_maintenance_health_retries:
+        description: Number of cluster health polling attempts.
+        type: int
+        default: 60
+      elasticsearch_maintenance_health_delay:
+        description: Delay in seconds between cluster health polling attempts.
+        type: int
+        default: 30
+      elasticsearch_maintenance_wait_health:
+        description: Wait for cluster health at the end. Disable for a defensive state reset at the start of a run.
+        type: bool
+        default: true
+      elasticsearch_maintenance_require_green:
+        description: Fail unless the cluster returns to green. The default accepts yellow.
+        type: bool
+        default: false
+      elasticsearch_drain_cluster_settings:
+        description: Persistent cluster settings that were applied during the drain; every key is restored to its baseline.
+        type: dict
+        default: {}
diff --git a/roles/elasticsearch/tasks/node_maintenance_end.yml b/roles/elasticsearch/tasks/node_maintenance_end.yml
new file mode 100644
index 00000000..cb312015
--- /dev/null
+++ b/roles/elasticsearch/tasks/node_maintenance_end.yml
@@ -0,0 +1,163 @@
+---
+# Entry point: restore cluster state after node maintenance. Counterpart of
+# node_maintenance_start; safe to run defensively (all restore steps are
+# best-effort) and from an always block after a failed maintenance.
+#
+# Optional: set elasticsearch_maintenance_wait_for_node to a node name to wait
+# for that node to rejoin the cluster before the health gate.
+
+- name: node_maintenance_end | Validate credentials
+  ansible.builtin.assert:
+    that:
+      - elasticsearch_maintenance_password is defined
+      - elasticsearch_maintenance_password | length > 0
+    fail_msg: elasticsearch_maintenance_password must be set to the elastic user password.
+    quiet: true
+
+- name: node_maintenance_end | Resolve API URL
+  ansible.builtin.set_fact:
+    _elasticsearch_maintenance_url: >-
+      {{ elasticsearch_maintenance_api_url
+         if elasticsearch_maintenance_api_url is defined
+         else elasticsearch_http_protocol ~ '://' ~ elasticsearch_api_host
+              ~ ':' ~ elasticstack_elasticsearch_http_port }}
+
+- name: node_maintenance_end | Wait for cluster API
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_cluster/health"
+    method: GET
+    status_code: [200, 503]
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  register: _elasticsearch_maintenance_api
+  until: (_elasticsearch_maintenance_api.status | default(0)) == 200
+  retries: 12
+  delay: 10
+  changed_when: false
+  failed_when: false
+  no_log: "{{ elasticstack_no_log }}"
+
+- name: node_maintenance_end | Re-enable shard allocation
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_cluster/settings"
+    method: PUT
+    body: '{ "persistent": { "cluster.routing.allocation.enable": null } }'
+    body_format: json
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  register: _elasticsearch_maintenance_alloc
+  until: (_elasticsearch_maintenance_alloc.json | default({})).acknowledged | default(false)
+  retries: 10
+  delay: 30
+  failed_when: false
+  no_log: "{{ elasticstack_no_log }}"
+
+- name: node_maintenance_end | Disable ML upgrade mode
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_ml/set_upgrade_mode?enabled=false"
+    method: POST
+    status_code: [200]
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  failed_when: false
+  no_log: "{{ elasticstack_no_log }}"
+  when: elasticsearch_ml_enabled | bool
+
+- name: node_maintenance_end | Clear voting config exclusions
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_cluster/voting_config_exclusions?wait_for_removal=false"
+    method: DELETE
+    status_code: [200]
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  failed_when: false
+  no_log: "{{ elasticstack_no_log }}"
+
+# Every key the drain boosted goes back to its declared baseline from
+# elasticsearch_cluster_settings, or to null when it has no baseline there.
+- name: node_maintenance_end | Restore recovery settings to the baseline
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_cluster/settings"
+    method: PUT
+    body: "{{ {'persistent': _elasticsearch_maintenance_baseline} | to_json }}"
+    body_format: json
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  vars:
+    _elasticsearch_maintenance_drain_keys: "{{ elasticsearch_drain_cluster_settings | list }}"
+    _elasticsearch_maintenance_baseline: >-
+      {{ dict(_elasticsearch_maintenance_drain_keys
+              | zip([None] * (_elasticsearch_maintenance_drain_keys | length)))
+         | combine(elasticsearch_cluster_settings | default({}) | dict2items
+                   | selectattr('key', 'in', _elasticsearch_maintenance_drain_keys) | items2dict) }}
+  register: _elasticsearch_maintenance_restore
+  until: (_elasticsearch_maintenance_restore.json | default({})).acknowledged | default(false)
+  retries: 5
+  delay: 10
+  failed_when: false
+  no_log: "{{ elasticstack_no_log }}"
+  when: elasticsearch_drain_cluster_settings | length > 0
+
+- name: node_maintenance_end | Confirm the node rejoined the cluster
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_cat/nodes?h=name"
+    method: GET
+    return_content: true
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  register: _elasticsearch_maintenance_nodes
+  until: elasticsearch_maintenance_wait_for_node in (_elasticsearch_maintenance_nodes.content | default('')).split()
+  retries: 60
+  delay: 10
+  changed_when: false
+  no_log: "{{ elasticstack_no_log }}"
+  when: elasticsearch_maintenance_wait_for_node | default('') | length > 0
+
+- name: node_maintenance_end | Wait for cluster health
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_cluster/health"
+    method: GET
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  register: _elasticsearch_maintenance_health
+  until: >-
+    ((_elasticsearch_maintenance_health.json | default({})).status | default(''))
+    in (['green'] if elasticsearch_maintenance_wait_status == 'green' else ['green', 'yellow'])
+  retries: "{{ elasticsearch_maintenance_health_retries }}"
+  delay: "{{ elasticsearch_maintenance_health_delay }}"
+  changed_when: false
+  failed_when: false
+  no_log: "{{ elasticstack_no_log }}"
+  when: elasticsearch_maintenance_wait_health | bool
+
+# Hard stop for serialised maintenance loops: with any_errors_fatal in the
+# calling play this keeps the next node untouched on an unhealthy cluster.
+- name: node_maintenance_end | Require a workable cluster before continuing
+  ansible.builtin.assert:
+    that:
+      - _elasticsearch_maintenance_status in ['green', 'yellow']
+      - >-
+        not (elasticsearch_maintenance_require_green | bool)
+        or _elasticsearch_maintenance_status == 'green'
+    fail_msg: >-
+      Cluster status is '{{ _elasticsearch_maintenance_status }}' after maintenance;
+      not proceeding to the next node.
+    quiet: true
+  vars:
+    _elasticsearch_maintenance_status: >-
+      {{ (_elasticsearch_maintenance_health.json | default({})).status | default('unreachable') }}
+  when: elasticsearch_maintenance_wait_health | bool
diff --git a/roles/elasticsearch/tasks/node_maintenance_start.yml b/roles/elasticsearch/tasks/node_maintenance_start.yml
new file mode 100644
index 00000000..f30824e8
--- /dev/null
+++ b/roles/elasticsearch/tasks/node_maintenance_start.yml
@@ -0,0 +1,141 @@
+---
+# Entry point: prepare the cluster for taking this node down (OS updates,
+# reboots, storage work — anything that stops Elasticsearch outside the role's
+# own upgrade/restart flows). Mirrors Elastic's rolling-upgrade procedure:
+# health gate -> voting exclusion -> allocation primaries -> ML upgrade mode ->
+# optional recovery boost -> flush.
+#
+# Call via:
+#   ansible.builtin.include_role:
+#     name: oddly.elasticstack.elasticsearch
+#     tasks_from: node_maintenance_start
+#   vars:
+#     elasticsearch_maintenance_password: "{{ your_elastic_password }}"
+#
+# Pair with node_maintenance_end (put it in an always block so cluster state
+# is restored even when the maintenance itself fails).
+
+- name: node_maintenance_start | Validate credentials
+  ansible.builtin.assert:
+    that:
+      - elasticsearch_maintenance_password is defined
+      - elasticsearch_maintenance_password | length > 0
+    fail_msg: elasticsearch_maintenance_password must be set to the elastic user password.
+    quiet: true
+
+- name: node_maintenance_start | Resolve API URL
+  ansible.builtin.set_fact:
+    _elasticsearch_maintenance_url: >-
+      {{ elasticsearch_maintenance_api_url
+         if elasticsearch_maintenance_api_url is defined
+         else elasticsearch_http_protocol ~ '://' ~ elasticsearch_api_host
+              ~ ':' ~ elasticstack_elasticsearch_http_port }}
+
+- name: node_maintenance_start | Wait for cluster API
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_cluster/health"
+    method: GET
+    status_code: [200, 503]
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  register: _elasticsearch_maintenance_api
+  until: (_elasticsearch_maintenance_api.status | default(0)) == 200
+  retries: 12
+  delay: 10
+  changed_when: false
+  no_log: "{{ elasticstack_no_log }}"
+
+- name: node_maintenance_start | Wait for cluster health before draining this node
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_cluster/health"
+    method: GET
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  register: _elasticsearch_maintenance_health
+  until: >-
+    ((_elasticsearch_maintenance_health.json | default({})).status | default(''))
+    in (['green'] if elasticsearch_maintenance_wait_status == 'green' else ['green', 'yellow'])
+  retries: "{{ elasticsearch_maintenance_health_retries }}"
+  delay: "{{ elasticsearch_maintenance_health_delay }}"
+  changed_when: false
+  no_log: "{{ elasticstack_no_log }}"
+
+# Voting exclusion is Elastic's documented mechanism for *permanent* master
+# removal, not for reboots/OS patches. Using it here anyway is defensive:
+# a long-running maintenance (kernel upgrade with a slow reboot, disk swap
+# with resilvering) is fine either way, and node_maintenance_end always
+# clears the exclusion. Rolling-upgrade skips this step on purpose because
+# the maintenance window there is short by design.
+- name: node_maintenance_start | Exclude this node from voting
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_cluster/voting_config_exclusions?node_names={{ elasticsearch_nodename | default(inventory_hostname) }}&timeout=60s"
+    method: POST
+    status_code: [200]
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  no_log: "{{ elasticstack_no_log }}"
+
+- name: node_maintenance_start | Disable replica allocation
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_cluster/settings"
+    method: PUT
+    body: '{ "persistent": { "cluster.routing.allocation.enable": "primaries" } }'
+    body_format: json
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  register: _elasticsearch_maintenance_alloc
+  until: (_elasticsearch_maintenance_alloc.json | default({})).acknowledged | default(false)
+  retries: 5
+  delay: 10
+  no_log: "{{ elasticstack_no_log }}"
+
+- name: node_maintenance_start | Enable ML upgrade mode
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_ml/set_upgrade_mode?enabled=true"
+    method: POST
+    status_code: [200]
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  failed_when: false
+  no_log: "{{ elasticstack_no_log }}"
+  when: elasticsearch_ml_enabled | bool
+
+- name: node_maintenance_start | Boost recovery settings for the drain window
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_cluster/settings"
+    method: PUT
+    body: "{{ {'persistent': elasticsearch_drain_cluster_settings} | to_json }}"
+    body_format: json
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  register: _elasticsearch_maintenance_boost
+  until: (_elasticsearch_maintenance_boost.json | default({})).acknowledged | default(false)
+  retries: 5
+  delay: 10
+  no_log: "{{ elasticstack_no_log }}"
+  when: elasticsearch_drain_cluster_settings | length > 0
+
+- name: node_maintenance_start | Flush indices
+  ansible.builtin.uri:
+    url: "{{ _elasticsearch_maintenance_url }}/_flush"
+    method: POST
+    status_code: [200, 409, 503]
+    user: elastic
+    password: "{{ elasticsearch_maintenance_password }}"
+    force_basic_auth: true
+    validate_certs: "{{ elasticsearch_validate_api_certs }}"
+  failed_when: false
+  changed_when: false
+  no_log: "{{ elasticstack_no_log }}"
diff --git a/tests/fakes/fake_es_rolling_api.py b/tests/fakes/fake_es_rolling_api.py
index 4ecc7a6c..13c999dc 100644
--- a/tests/fakes/fake_es_rolling_api.py
+++ b/tests/fakes/fake_es_rolling_api.py
@@ -84,6 +84,19 @@ def do_POST(self):
             if self.path.startswith("/_flush"):
                 self._send_json({"_shards": {"failed": 0}})
                 return
+            if self.path.startswith("/_cluster/voting_config_exclusions"):
+                self._send_json({})
+                return
+            if self.path.startswith("/_ml/set_upgrade_mode"):
+                self._send_json({"acknowledged": True})
+                return
+            self._send_json({"error": "not found"}, status=404)
+
+        def do_DELETE(self):
+            state.log(self.server.server_port, "DELETE", self.path, "")
+            if self.path.startswith("/_cluster/voting_config_exclusions"):
+                self._send_json({})
+                return
             self._send_json({"error": "not found"}, status=404)
 
         def do_PUT(self):
diff --git a/tests/integration/node_maintenance_contract.yml b/tests/integration/node_maintenance_contract.yml
new file mode 100644
index 00000000..dfaadaae
--- /dev/null
+++ b/tests/integration/node_maintenance_contract.yml
@@ -0,0 +1,225 @@
+---
+# Contract test for the node_maintenance_start / node_maintenance_end
+# entry points. Runs against the fake ES API from tests/fakes/ so it can
+# execute on a plain ubuntu-latest runner in CI without any real cluster.
+#
+# What it proves:
+#   - _start posts a voting exclusion for the node, disables replica
+#     allocation, enables ML upgrade mode, applies the drain boost, and
+#     flushes.
+#   - _end reverses all of it (allocation reset, ML disabled, exclusion
+#     deleted) AND restores each key in elasticsearch_drain_cluster_settings
+#     to its declared baseline from elasticsearch_cluster_settings (not
+#     null, which would silently drop declared settings back to defaults).
+#   - When elasticsearch_ml_enabled: false, both _start and _end skip the
+#     ML upgrade-mode calls entirely.
+
+- name: Start fake Elasticsearch API for node maintenance
+  hosts: localhost
+  gather_facts: false
+  vars:
+    _fake_es_log: /tmp/elasticstack-node-maintenance-api.jsonl
+  tasks:
+    - name: Stop previous fake API if present
+      ansible.builtin.shell: |
+        set -o pipefail
+        if [ -f /tmp/elasticstack-node-maintenance-api.pid ]; then
+          kill "$(cat /tmp/elasticstack-node-maintenance-api.pid)" 2>/dev/null || true
+        fi
+      args:
+        executable: /bin/bash
+      changed_when: false
+
+    - name: Reset fake API log
+      ansible.builtin.copy:
+        dest: "{{ _fake_es_log }}"
+        content: ""
+        mode: "0644"
+
+    - name: Start fake API server
+      ansible.builtin.shell: |
+        set -o pipefail
+        nohup python3 {{ playbook_dir }}/../fakes/fake_es_rolling_api.py \
+          --ports 19210 \
+          --nodes esA \
+          --persistent-settings '{"indices.recovery.max_bytes_per_sec": "50mb"}' \
+          --log {{ _fake_es_log }} >/tmp/elasticstack-node-maintenance-api.out 2>&1 &
+        echo $!
+      args:
+        executable: /bin/bash
+      register: _fake_es_api
+      changed_when: true
+
+    - name: Store fake API PID
+      ansible.builtin.copy:
+        dest: /tmp/elasticstack-node-maintenance-api.pid
+        content: "{{ _fake_es_api.stdout }}"
+        mode: "0644"
+
+    - name: Wait for fake API
+      ansible.builtin.uri:
+        url: http://127.0.0.1:19210/_cluster/health
+        method: GET
+      register: _fake_api_health
+      until: _fake_api_health.status | default(0) == 200
+      retries: 10
+      delay: 1
+
+- name: Contract - node_maintenance_start drains, node_maintenance_end restores
+  hosts: localhost
+  gather_facts: false
+  vars:
+    elasticstack_elasticsearch_http_port: 19210
+    elasticsearch_api_host: 127.0.0.1
+    elasticsearch_http_protocol: http
+    elasticsearch_validate_api_certs: false
+    elasticstack_no_log: false
+    elasticsearch_ml_enabled: true
+    elasticsearch_nodename: esA
+    # Restore semantics: the boost bumps max_bytes_per_sec to 500mb; the
+    # baseline in elasticsearch_cluster_settings is 200mb. After _end the
+    # cluster setting must be 200mb (not null), otherwise a caller who
+    # declared a persistent recovery ceiling would silently lose it.
+    elasticsearch_cluster_settings:
+      indices.recovery.max_bytes_per_sec: "200mb"
+    elasticsearch_drain_cluster_settings:
+      indices.recovery.max_bytes_per_sec: "500mb"
+      cluster.routing.allocation.node_concurrent_recoveries: 8
+    elasticsearch_maintenance_password: fake-elastic-pw
+    elasticsearch_maintenance_health_retries: 1
+    elasticsearch_maintenance_health_delay: 0
+    elasticsearch_maintenance_wait_status: yellow
+    elasticsearch_maintenance_wait_health: true
+    elasticsearch_maintenance_require_green: false
+  tasks:
+    - name: Drain the node
+      ansible.builtin.include_role:
+        name: oddly.elasticstack.elasticsearch
+        tasks_from: node_maintenance_start
+
+    - name: Restore the node
+      ansible.builtin.include_role:
+        name: oddly.elasticstack.elasticsearch
+        tasks_from: node_maintenance_end
+
+- name: Verify drain/restore conversation
+  hosts: localhost
+  gather_facts: false
+  vars:
+    _fake_es_log: /tmp/elasticstack-node-maintenance-api.jsonl
+  tasks:
+    - name: Read fake API log
+      ansible.builtin.command:
+        cmd: cat {{ _fake_es_log }}
+      register: _log
+      changed_when: false
+
+    # The fake API dumps one JSON object per request line, with the keys
+    # emitted alphabetically: {"body":…, "method":…, "path":…, "port":…}.
+    # `search()` regex '.' does not cross newlines by default, so each of
+    # these patterns matches within a single request line, and content
+    # that lives in the request body (primaries, 500mb, null) is placed
+    # before the "method"/"path" markers rather than after.
+    - name: Assert drain sequence hit every expected endpoint
+      ansible.builtin.assert:
+        that:
+          # Voting exclusion posted for the correct node
+          - "_log.stdout is search('voting_config_exclusions.*node_names=esA')"
+          # Allocation dropped to primaries
+          - "_log.stdout is search('primaries.*\"method\": \"PUT\".*/_cluster/settings')"
+          # ML upgrade mode enabled
+          - "_log.stdout is search('_ml/set_upgrade_mode.*enabled=true')"
+          # Boost applied (500mb from drain_cluster_settings)
+          - "_log.stdout is search('500mb.*\"method\": \"PUT\".*/_cluster/settings')"
+          # Flush called
+          - "_log.stdout is search('\"method\": \"POST\".*/_flush')"
+        fail_msg: |
+          Drain sequence missing an endpoint in the log:
+          {{ _log.stdout }}
+
+    - name: Assert restore sequence hit every expected endpoint
+      ansible.builtin.assert:
+        that:
+          # Allocation reset (null)
+          - "_log.stdout is search('allocation.enable.*null.*\"method\": \"PUT\".*/_cluster/settings')"
+          # ML upgrade mode disabled
+          - "_log.stdout is search('_ml/set_upgrade_mode.*enabled=false')"
+          # Voting exclusion cleared (DELETE)
+          - "_log.stdout is search('\"method\": \"DELETE\".*voting_config_exclusions')"
+          # Baseline restored: 200mb from elasticsearch_cluster_settings
+          - "_log.stdout is search('200mb.*\"method\": \"PUT\".*/_cluster/settings')"
+          # Boosted key that has no baseline (node_concurrent_recoveries) is
+          # explicitly nulled — not silently left at the boosted value
+          - "_log.stdout is search('node_concurrent_recoveries.*null.*\"method\": \"PUT\".*/_cluster/settings')"
+        fail_msg: |
+          Restore sequence missing an endpoint in the log:
+          {{ _log.stdout }}
+
+    - name: Reset fake API log for the ML-disabled run
+      ansible.builtin.copy:
+        dest: "{{ _fake_es_log }}"
+        content: ""
+        mode: "0644"
+
+- name: Contract - ML-disabled nodes never hit ML upgrade mode
+  hosts: localhost
+  gather_facts: false
+  vars:
+    elasticstack_elasticsearch_http_port: 19210
+    elasticsearch_api_host: 127.0.0.1
+    elasticsearch_http_protocol: http
+    elasticsearch_validate_api_certs: false
+    elasticstack_no_log: false
+    elasticsearch_ml_enabled: false
+    elasticsearch_nodename: esA
+    elasticsearch_cluster_settings: {}
+    elasticsearch_drain_cluster_settings: {}
+    elasticsearch_maintenance_password: fake-elastic-pw
+    elasticsearch_maintenance_health_retries: 1
+    elasticsearch_maintenance_health_delay: 0
+    elasticsearch_maintenance_wait_status: yellow
+    elasticsearch_maintenance_wait_health: false
+    elasticsearch_maintenance_require_green: false
+  tasks:
+    - name: Drain the node (ML disabled)
+      ansible.builtin.include_role:
+        name: oddly.elasticstack.elasticsearch
+        tasks_from: node_maintenance_start
+
+    - name: Restore the node (ML disabled)
+      ansible.builtin.include_role:
+        name: oddly.elasticstack.elasticsearch
+        tasks_from: node_maintenance_end
+
+- name: Verify ML calls were skipped
+  hosts: localhost
+  gather_facts: false
+  tasks:
+    - name: Read fake API log
+      ansible.builtin.command:
+        cmd: cat /tmp/elasticstack-node-maintenance-api.jsonl
+      register: _log
+      changed_when: false
+
+    - name: Assert _ml/set_upgrade_mode was never called
+      ansible.builtin.assert:
+        that:
+          - _log.stdout is not search('_ml/set_upgrade_mode')
+        fail_msg: |
+          ML upgrade mode call fired despite elasticsearch_ml_enabled: false.
+          Log:
+          {{ _log.stdout }}
+
+- name: Stop fake API
+  hosts: localhost
+  gather_facts: false
+  tasks:
+    - name: Kill fake API
+      ansible.builtin.shell: |
+        set -o pipefail
+        if [ -f /tmp/elasticstack-node-maintenance-api.pid ]; then
+          kill "$(cat /tmp/elasticstack-node-maintenance-api.pid)" 2>/dev/null || true
+        fi
+      args:
+        executable: /bin/bash
+      changed_when: false