diff --git a/.github/workflows/node-recovery.yml b/.github/workflows/node-recovery.yml
new file mode 100644
index 0000000..3661d3b
--- /dev/null
+++ b/.github/workflows/node-recovery.yml
@@ -0,0 +1,44 @@
+name: Node recovery
+
+# Runs the ARC role's node recovery watcher in a three-node kind cluster (tests/node_recovery/run.sh): the role's rendered Deployment and RBAC, the watcher image built from this checkout, and a stopped worker container standing in for a node that goes down. Only on changes that can affect the watcher.
+#
+# GitHub-hosted ubuntu-latest has Docker, kind, kubectl and jq already.
+
+on:
+  pull_request:
+    paths:
+      - roles/github_runner_arc/**
+      - scripts/node-recovery.sh
+      - node-recovery/**
+      - tests/node_recovery/**
+      - .github/workflows/node-recovery.yml
+  workflow_dispatch:
+
+concurrency:
+  group: node-recovery-${{ github.ref }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  recovery:
+    name: Recovery in kind
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Install Ansible
+        run: |
+          set -euo pipefail
+          if ! python3 -m venv "${RUNNER_TEMP}/venv" 2>/dev/null; then
+            sudo apt-get update -qq
+            sudo apt-get install -y -qq python3-venv
+            python3 -m venv "${RUNNER_TEMP}/venv"
+          fi
+          "${RUNNER_TEMP}/venv/bin/pip" install --quiet --prefer-binary ansible-core
+          echo "${RUNNER_TEMP}/venv/bin" >> "$GITHUB_PATH"
+
+      - name: Run the node recovery test
+        run: tests/node_recovery/run.sh
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 2dfbe86..5c1f7d4 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -1,6 +1,6 @@
 name: Release
 
-# Runs semantic-release against every push to main. Almost every conventional-commit type triggers at least a patch release (see .releaserc.json's commit-analyzer releaseRules), so in practice this fires on nearly every push, stamping the collection version into galaxy.yml and the arc role's heartbeat, autoscaler and pull-secret-renewer image defaults, building the collection tarball, and, if a release was actually published, building and pushing all four container images (runner, heartbeat, autoscaler, pull-secret-renewer) tagged with the exact new version alongside floating major and latest tags on the same manifest.
+# Runs semantic-release against every push to main. Almost every conventional-commit type triggers at least a patch release (see .releaserc.json's commit-analyzer releaseRules), so in practice this fires on nearly every push, stamping the collection version into galaxy.yml and the arc role's heartbeat, autoscaler, pull-secret-renewer and node-recovery image defaults, building the collection tarball, and, if a release was actually published, building and pushing all five container images (runner, heartbeat, autoscaler, pull-secret-renewer, node-recovery) tagged with the exact new version alongside floating major and latest tags on the same manifest.
 #
 # The release commit and tag are pushed over SSH with a write-enabled deploy key (secret RELEASE_DEPLOY_KEY): actions/checkout loads it for every later git command, and .releaserc.json's repositoryUrl is the SSH remote, so @semantic-release/git pushes with it. A deploy key is scoped to this one repository and can be listed as a ruleset bypass actor. It can't call the GitHub API, so the semantic-release process uses the ordinary secrets.GITHUB_TOKEN for creating the Release and uploading the collection tarball, which need no bypass.
 #
@@ -76,6 +76,9 @@ jobs:
           - name: pull-secret-renewer
             dockerfile: pull-secret-renewer/Dockerfile
             image: ghcr.io/exadev/github-runner-pull-secret-renewer
+          - name: node-recovery
+            dockerfile: node-recovery/Dockerfile
+            image: ghcr.io/exadev/github-runner-node-recovery
     steps:
       - uses: actions/checkout@v7
         with:
diff --git a/README.md b/README.md
index 669a75c..b5a5a9f 100644
--- a/README.md
+++ b/README.md
@@ -43,8 +43,8 @@ Which hosts are k3s servers is worked out from the inventory, not set per host.
 
 ## Build, test, and smoke-test
 
-- **Build the runner image locally (the only working path right now):** `docker buildx build --platform linux/arm64,linux/amd64 --push -t ghcr.io/exadev/github-runner:latest .` from a machine with buildx/QEMU cross-platform support (Docker Desktop has this built in). A plain `docker build` with no `--platform` only produces an image for the building machine's own architecture, which silently breaks scheduling on the fleet's other architecture. The fleet-health platform's own two images need the identical treatment, for the identical reason (they're genuinely unpinned pods now too - see [Heartbeat](#heartbeat)/[Autoscaler](#autoscaler)): `docker buildx build --platform linux/arm64,linux/amd64 --push -f heartbeat/Dockerfile -t ghcr.io/exadev/github-runner-heartbeat:latest .` and the same with `autoscaler/Dockerfile` → `ghcr.io/exadev/github-runner-autoscaler:latest`, and `pull-secret-renewer/Dockerfile` → `ghcr.io/exadev/github-runner-pull-secret-renewer:latest` (the image of the ARC role's App-sourced pull Secret renewal).
-- **Build and push via CI:** `.github/workflows/release.yml`'s `build-images` job builds and pushes all four images (this one, `heartbeat`, `autoscaler` and `pull-secret-renewer`) multi-arch on GitHub-hosted `ubuntu-latest` whenever a release is published - see [Releases](#releases) below.
+- **Build the runner image locally (the only working path right now):** `docker buildx build --platform linux/arm64,linux/amd64 --push -t ghcr.io/exadev/github-runner:latest .` from a machine with buildx/QEMU cross-platform support (Docker Desktop has this built in). A plain `docker build` with no `--platform` only produces an image for the building machine's own architecture, which silently breaks scheduling on the fleet's other architecture. The fleet-health platform's own two images need the identical treatment, for the identical reason (they're genuinely unpinned pods now too - see [Heartbeat](#heartbeat)/[Autoscaler](#autoscaler)): `docker buildx build --platform linux/arm64,linux/amd64 --push -f heartbeat/Dockerfile -t ghcr.io/exadev/github-runner-heartbeat:latest .` and the same with `autoscaler/Dockerfile` → `ghcr.io/exadev/github-runner-autoscaler:latest`, `pull-secret-renewer/Dockerfile` → `ghcr.io/exadev/github-runner-pull-secret-renewer:latest` (the image of the ARC role's App-sourced pull Secret renewal), and `node-recovery/Dockerfile` → `ghcr.io/exadev/github-runner-node-recovery:latest` (see [Node recovery](#node-recovery)).
+- **Build and push via CI:** `.github/workflows/release.yml`'s `build-images` job builds and pushes all five images (this one, `heartbeat`, `autoscaler`, `pull-secret-renewer` and `node-recovery`) multi-arch on GitHub-hosted `ubuntu-latest` whenever a release is published - see [Releases](#releases) below.
 - **Smoke-test the ARC scale set** (confirms a job gets a real ephemeral pod): `gh workflow run test-arc-runner.yml`, then watch `kubectl get pods -n arc-runners-<org> -w`. A pod must appear only once the job is queued, run to completion, and be deleted within seconds.
 - **Smoke-test `ubuntu-latest` routing** (confirms `runner-fallback-action` still reaches GitHub-hosted runners when needed): `gh workflow run test-ubuntu-latest.yml`.
 - **Generate load for the autoscaler** (see [Autoscaler](#autoscaler) below): `gh workflow run test-autoscaler.yml`.
@@ -65,10 +65,10 @@ Every push to `main` runs `.github/workflows/release.yml`, driven by [semantic-r
 
 A release:
 
-1. Stamps the new version into `galaxy.yml`'s `version:` field (which stays `0.0.0` in source between releases) and into the published-image tag defaults in `roles/github_runner_arc/defaults/main.yml` (`github_runner_arc_heartbeat_image`, `github_runner_arc_autoscaler_image`, `github_runner_arc_image_pull_secret_renewer_image`) - `scripts/release/stamp_version.py` does this by matching each variable's own name, not a line number, so it stays correct even while those files are edited concurrently by unrelated work.
+1. Stamps the new version into `galaxy.yml`'s `version:` field (which stays `0.0.0` in source between releases) and into the published-image tag defaults in `roles/github_runner_arc/defaults/main.yml` (`github_runner_arc_heartbeat_image`, `github_runner_arc_autoscaler_image`, `github_runner_arc_image_pull_secret_renewer_image`, `github_runner_arc_node_recovery_image`) - `scripts/release/stamp_version.py` does this by matching each variable's own name, not a line number, so it stays correct even while those files are edited concurrently by unrelated work.
 2. Builds the collection tarball (`ansible-galaxy collection build --force`) and attaches it to the GitHub Release.
 3. Updates `CHANGELOG.md` and commits it, `galaxy.yml`, and the stamped defaults file back to `main` as `chore(release): <version> [skip ci]`.
-4. Builds and pushes all four container images (`Dockerfile`, `heartbeat/Dockerfile`, `autoscaler/Dockerfile`, `pull-secret-renewer/Dockerfile`) multi-arch (`linux/arm64,linux/amd64`) to their `ghcr.io/exadev/github-runner*` repositories, each tagged with the exact version, the floating major version, and `latest` - all three tags on the same manifest, via `docker/metadata-action`'s `type=semver` patterns. To pin a host to a specific release rather than floating on `latest`, set `roles/github_runner_arc/defaults/main.yml`'s image variables (or a host's own override) to an exact version tag instead.
+4. Builds and pushes all five container images (`Dockerfile`, `heartbeat/Dockerfile`, `autoscaler/Dockerfile`, `pull-secret-renewer/Dockerfile`, `node-recovery/Dockerfile`) multi-arch (`linux/arm64,linux/amd64`) to their `ghcr.io/exadev/github-runner*` repositories, each tagged with the exact version, the floating major version, and `latest` - all three tags on the same manifest, via `docker/metadata-action`'s `type=semver` patterns. To pin a host to a specific release rather than floating on `latest`, set `roles/github_runner_arc/defaults/main.yml`'s image variables (or a host's own override) to an exact version tag instead.
 5. Optionally publishes the collection to Ansible Galaxy, only if a `GALAXY_API_KEY` secret is present - the `[skip ci]` release commit and the image build both happen unconditionally, in this same run, regardless of whether Galaxy publishing is configured.
 
 The `[skip ci]` in the release commit message is deliberate: the image build already happens in this same run using the exact version semantic-release just decided, so there is nothing useful for a second, separately-triggered run of this workflow (or of `ci.yml`) to do against a commit that only changed a version stamp and a changelog - `ci.yml`'s own lint jobs already ran against every commit before it reached `main`.
@@ -134,6 +134,18 @@ Ships with `AUTOSCALER_DRY_RUN=true` by default: it computes and logs the target
 
 Every Helm upgrade above reverts `maxRunners` to the values file's safe floor. The role's `install_org.yml` triggers one immediate autoscaler poll after every such upgrade (`kubectl exec deploy/autoscaler -- ...`, reaching the pod through the API server regardless of which node it's on), so the safe-floor window after a redeploy is seconds, not a full poll interval.
 
+### Node recovery
+
+When a node stops (its host is switched off or asleep, Docker quits, or the k3s container is stopped), the node controller marks it `Unknown` once its kubelet has missed heartbeats for the node-monitor grace period, and after the pods' five-minute unreachable toleration marks them for deletion. Only the node's kubelet can confirm a deletion, so they then stay `Terminating` indefinitely. A Deployment's pods are replaced regardless, which is why the ARC controller comes back on its own, but ARC creates each scale set's listener as a single pod and waits for the old one to go, so a listener on the stopped node is never replaced and every job for that scale set queues until someone force-deletes it.
+
+`node-recovery` (an in-cluster Deployment in the platform namespace, see `roles/github_runner_arc/templates/node-recovery.yaml.j2` and `scripts/node-recovery.sh`) applies Kubernetes' [non-graceful node shutdown](https://kubernetes.io/docs/concepts/cluster-administration/node-shutdown/#non-graceful-node-shutdown): every `NODE_RECOVERY_POLL_SECONDS` (default 15s) it gives each node whose `Ready` condition has been `Unknown` for `github_runner_arc_node_recovery_after_seconds` (default 120s) the `node.kubernetes.io/out-of-service=nodeshutdown:NoExecute` taint. The control plane then evicts every pod on that node at once, unless it tolerates that taint, and the pod garbage collector force-deletes the ones already terminating there, so ARC starts the listener again on a healthy node. When the node reports `Ready` again the watcher removes the taint, and the node's kubelet stops the containers of the pods that were deleted while it was away. With the defaults a stopped node stops holding up job pickup within roughly four minutes: the grace period (50s on current Kubernetes), the threshold, one poll, the garbage collector's 20-second sweep, and the listener's own restart. `github_runner_arc_node_recovery_enabled: false` removes the watcher and any taint it applied.
+
+The watcher only acts on a node whose kubelet has gone silent (`Unknown`), never on one reporting `NotReady` (`False`), whose kubelet is alive and finishes its pods' deletion itself. It never taints the node it runs on, removes only taints it applied (recognised by its `github-runner.exadev/out-of-service-applied-at` annotation), and sends each change as a JSON patch that first tests the node's `resourceVersion`, so a node that came back since it was read is left for the next poll. RBAC is `get`, `list` and `patch` on nodes, with no `delete`: the watcher never removes a Node object, and the taint does not touch k3s's etcd membership, which only deleting a server's Node object changes. Its own pod tolerates an unreachable or not-ready node for only 10 seconds, so if it was running on the node that stopped, its ReplicaSet starts a replacement elsewhere well before the threshold passes.
+
+The threshold is also where a stopped node's runner pods are given up, earlier than the five-minute default eviction would. Their containers stopped with the node, so their jobs are lost either way unless the node returns within that time; lower the threshold for faster recovery, or raise it to wait longer for a node that is only asleep.
+
+Its image, `github_runner_arc_node_recovery_image`, is pulled without a pull Secret, so the published package must be public.
+
 ## Conventions
 
 Adding a second org is additive, never a change to anything existing:
diff --git a/galaxy.yml b/galaxy.yml
index fad4dbf..732e1b8 100644
--- a/galaxy.yml
+++ b/galaxy.yml
@@ -38,6 +38,7 @@ build_ignore:
   - autoscaler
   - heartbeat
   - pull-secret-renewer
+  - node-recovery
   - scripts
   - Dockerfile
   - bootstrap.sh
diff --git a/node-recovery/Dockerfile b/node-recovery/Dockerfile
new file mode 100644
index 0000000..896d05d
--- /dev/null
+++ b/node-recovery/Dockerfile
@@ -0,0 +1,13 @@
+# Image for the node recovery Deployment (see roles/github_runner_arc/templates/node-recovery.yaml.j2 and scripts/node-recovery.sh): bash, jq and kubectl, all from Alpine's apk, so it is small and builds for either architecture. Build context is the repository root so it can COPY the script. The Deployment pulls it without a pull Secret, so the published package must be public.
+FROM alpine:3.20
+
+RUN apk add --no-cache bash jq kubectl
+
+COPY node-recovery/loop.sh /app/loop.sh
+COPY scripts/node-recovery.sh /app/scripts/node-recovery.sh
+RUN chmod +x /app/loop.sh /app/scripts/node-recovery.sh
+
+# A non-root user by number, so the Deployment's runAsNonRoot check can verify it without reading /etc/passwd.
+USER 65532:65532
+WORKDIR /app
+ENTRYPOINT ["/app/loop.sh"]
diff --git a/node-recovery/loop.sh b/node-recovery/loop.sh
new file mode 100755
index 0000000..dbb5b5a
--- /dev/null
+++ b/node-recovery/loop.sh
@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+# Container entrypoint for the node recovery Deployment (see roles/github_runner_arc/templates/node-recovery.yaml.j2). Runs scripts/node-recovery.sh every NODE_RECOVERY_POLL_SECONDS forever; a failed pass is logged and retried on the next one. kubectl finds the in-cluster config from the pod's mounted ServiceAccount token.
+set -euo pipefail
+
+NODE_RECOVERY_POLL_SECONDS="${NODE_RECOVERY_POLL_SECONDS:-15}"
+
+while true; do
+  /app/scripts/node-recovery.sh || true
+  sleep "$NODE_RECOVERY_POLL_SECONDS"
+done
diff --git a/roles/github_runner_arc/README.md b/roles/github_runner_arc/README.md
index 182f6e4..7cdb49e 100644
--- a/roles/github_runner_arc/README.md
+++ b/roles/github_runner_arc/README.md
@@ -37,6 +37,7 @@ Before touching the cluster the role then checks the secrets it is about to writ
 - `github_runner_arc_image_pull_registry`, `github_runner_arc_image_pull_secret_name`, `github_runner_arc_verify_image_pull`: the pull Secret's registry and name, and whether to prove the credential before writing it.
 - `github_runner_arc_heartbeat_gist_id`: install the fleet-health platform from this host, refreshing this gist. `github_runner_arc_heartbeat_bootstrap_gist`, `github_runner_arc_heartbeat_gist_description` and `github_runner_arc_heartbeat_gist_consumer` control the bootstrap described above.
 - `github_runner_arc_autoscaler_usable_budget_gi`, `github_runner_arc_autoscaler_max_ceiling`, `github_runner_arc_autoscaler_floor`: required when a profile sets `autoscale: true` and the platform is installed. The floor must equal the autoscaled profile's `maxRunners`, which every Helm upgrade reverts to. The other `github_runner_arc_autoscaler_*` and `github_runner_arc_heartbeat_*` settings have defaults; see `defaults/main.yml`.
+- `github_runner_arc_node_recovery_enabled` (default `true`), `github_runner_arc_node_recovery_after_seconds` (default 120), `github_runner_arc_node_recovery_poll_seconds` (default 15), `github_runner_arc_node_recovery_image`: the node recovery watcher (see [Node recovery](#node-recovery)).
 - `github_runner_arc_app_setup_*`, `github_runner_arc_app_manifest_code`, `github_runner_arc_app_private_key_path`: inputs to `playbooks/github_app_setup.yml` (see below), including `github_runner_arc_app_setup_write_secret`, `github_runner_arc_app_setup_secret_namespaces`, `github_runner_arc_app_setup_secret_name`, `github_runner_arc_app_setup_keep_private_key` and `github_runner_arc_app_setup_command`.
 
 ## Sizing
@@ -96,6 +97,10 @@ github_runner_arc_orgs:
       - max_runners: 4
 ```
 
+## Node recovery
+
+Every host that installs anything also installs a small watcher, the `node-recovery` Deployment in `github_runner_arc_platform_namespace`, so that a node that stops does not leave its pods `Terminating` for ever. ARC waits for a listener's old pod to go before starting a new one, so a listener on a stopped node otherwise stalls its scale set until the pod is force-deleted by hand. Once a node's `Ready` condition has been `Unknown` for `github_runner_arc_node_recovery_after_seconds`, the watcher gives it the `node.kubernetes.io/out-of-service=nodeshutdown:NoExecute` taint of Kubernetes' non-graceful node shutdown, which makes the control plane evict the node's pods and force-delete the terminating ones; it removes the taint when the node reports `Ready` again. It never acts on a node reporting `NotReady`, never on its own node, and never removes a taint it did not apply. Its ClusterRole allows `get`, `list` and `patch` on nodes and nothing else. `github_runner_arc_node_recovery_enabled: false` removes the watcher, its RBAC and any taint it left behind. Its image is pulled without a pull Secret, so it must be public. The repository README's Node recovery section explains the default threshold.
+
 ## GitHub App setup
 
 `playbooks/github_app_setup.yml` creates the App the scale sets authenticate as, through GitHub's manifest flow. The first run renders a local form that posts the manifest (organisation self-hosted runner write access, no webhook, private) to GitHub; an organisation owner submits it and copies the one-time code GitHub returns. The second run, with that code and a path for the key, exchanges the code, writes the private key there, waits while the App is installed on the organisation, and prints the `app_id` and `installation_id` to record.
diff --git a/roles/github_runner_arc/defaults/main.yml b/roles/github_runner_arc/defaults/main.yml
index 5ac01ed..ab25b5d 100644
--- a/roles/github_runner_arc/defaults/main.yml
+++ b/roles/github_runner_arc/defaults/main.yml
@@ -80,6 +80,15 @@ github_runner_arc_autoscaler_poll_seconds: 45
 github_runner_arc_autoscaler_raise_confirm_polls: 2
 github_runner_arc_autoscaler_mem_available_pressure_pct: 15
 
+# Node recovery (see tasks/install_node_recovery.yml and scripts/node-recovery.sh). When a node's kubelet stops reporting, the control plane marks the node's pods for deletion but only the kubelet can confirm it, so they stay Terminating, and an ARC listener on that node is never replaced. The watcher gives such a node the node.kubernetes.io/out-of-service=nodeshutdown:NoExecute taint once its Ready condition has been Unknown for github_runner_arc_node_recovery_after_seconds, which makes the control plane evict its pods and force-delete the terminating ones, and removes the taint when the node reports Ready again. false removes the watcher and any taint it left.
+github_runner_arc_node_recovery_enabled: true
+# Counted from when the node controller marks the node Unknown, which it does only after the kubelet has missed heartbeats for the controller manager's node-monitor-grace-period (50 seconds on current Kubernetes). 120 seconds more leaves room for a k3s container restart or image upgrade to report Ready again before anything is given up, while bounding how long a stopped node stalls its listeners to roughly four minutes at worst (the grace period, this threshold, one poll, the pod garbage collector's 20-second sweep and the listener's own restart) instead of indefinitely. It is also where the node's own runner pods are given up, earlier than the five-minute default eviction would; their containers stopped with the node, so a job on it is lost either way unless the node returns within that time. Lower it for faster recovery at the cost of losing the jobs of a node that comes back sooner; raise it to keep them for longer outages.
+github_runner_arc_node_recovery_after_seconds: 120
+# How often the watcher checks the nodes. A taint lands at most this long after the threshold passes.
+github_runner_arc_node_recovery_poll_seconds: 15
+# Image the watcher runs. Pulled without a pull Secret, so it must be public.
+github_runner_arc_node_recovery_image: "ghcr.io/exadev/github-runner-node-recovery:1.2.2"
+
 # GitHub App creation through the manifest flow (tasks/app_setup.yml, run by playbooks/github_app_setup.yml). The App needs only organisation self-hosted runner write access; add packages: read to the permissions if its installation token should also pull private packages.
 github_runner_arc_app_setup_org: ""
 github_runner_arc_app_setup_name: ""
diff --git a/roles/github_runner_arc/meta/argument_specs.yml b/roles/github_runner_arc/meta/argument_specs.yml
index 92fe49b..8c736df 100644
--- a/roles/github_runner_arc/meta/argument_specs.yml
+++ b/roles/github_runner_arc/meta/argument_specs.yml
@@ -176,6 +176,23 @@ argument_specs:
         type: int
         default: 15
         description: Available memory percentage below which the autoscaler treats a node as under pressure.
+      github_runner_arc_node_recovery_enabled:
+        type: bool
+        default: true
+        description:
+          - Install the node recovery watcher, which gives a node the out-of-service taint once it has stopped reporting for github_runner_arc_node_recovery_after_seconds, so the control plane clears its stuck pods, and removes the taint when the node is Ready again.
+          - false removes the watcher and any taint it applied.
+      github_runner_arc_node_recovery_after_seconds:
+        type: int
+        default: 120
+        description: Seconds a node's Ready condition must have been Unknown before the watcher taints it out of service.
+      github_runner_arc_node_recovery_poll_seconds:
+        type: int
+        default: 15
+        description: How often the node recovery watcher checks the nodes.
+      github_runner_arc_node_recovery_image:
+        type: str
+        description: Image the node recovery watcher runs, pulled without a pull Secret.
       github_runner_arc_autoscaler_usable_budget_gi:
         type: int
         description: Memory runner pods may use across every node. Required for an autoscaled profile.
diff --git a/roles/github_runner_arc/tasks/install_node_recovery.yml b/roles/github_runner_arc/tasks/install_node_recovery.yml
new file mode 100644
index 0000000..ca40946
--- /dev/null
+++ b/roles/github_runner_arc/tasks/install_node_recovery.yml
@@ -0,0 +1,59 @@
+---
+# Installs the node recovery watcher (templates/node-recovery.yaml.j2, running scripts/node-recovery.sh) when github_runner_arc_node_recovery_enabled is on, and otherwise removes it along with any out-of-service taint it left behind, so that turning it off never leaves a node that returns unable to run anything. Included from main.yml on any host that installs something.
+
+- name: "Node recovery: describe its resources"
+  ansible.builtin.set_fact:
+    github_runner_arc_node_recovery_resources: "{{ lookup('ansible.builtin.template', 'node-recovery.yaml.j2') | from_yaml_all | list }}"
+
+- name: "Node recovery: create the namespace it runs in"
+  kubernetes.core.k8s:
+    definition:
+      apiVersion: v1
+      kind: Namespace
+      metadata:
+        name: "{{ github_runner_arc_platform_namespace }}"
+  when: github_runner_arc_node_recovery_enabled | bool
+
+- name: "Node recovery: install the watcher"
+  kubernetes.core.k8s:
+    definition: "{{ item }}"
+  loop: "{{ github_runner_arc_node_recovery_resources }}"
+  loop_control:
+    label: "{{ item.kind }}/{{ item.metadata.name }}"
+  when: github_runner_arc_node_recovery_enabled | bool
+
+- name: "Node recovery: remove the watcher and its taints"
+  when: not (github_runner_arc_node_recovery_enabled | bool)
+  block:
+    # The Deployment goes before the permissions it runs with, so nothing is left running that can no longer patch nodes.
+    - name: "Node recovery: remove the watcher"
+      kubernetes.core.k8s:
+        state: absent
+        api_version: "{{ item.apiVersion }}"
+        kind: "{{ item.kind }}"
+        namespace: "{{ item.metadata.namespace | default(omit) }}"
+        name: "{{ item.metadata.name }}"
+      loop: "{{ github_runner_arc_node_recovery_resources | reverse | list }}"
+      loop_control:
+        label: "{{ item.kind }}/{{ item.metadata.name }}"
+
+    - name: "Node recovery: find the nodes it tainted"
+      kubernetes.core.k8s_info:
+        kind: Node
+      register: github_runner_arc_node_recovery_nodes
+
+    # Only nodes carrying the watcher's own annotation, and only the out-of-service taint; every other taint and annotation is kept.
+    - name: "Node recovery: remove the taint it applied"
+      kubernetes.core.k8s_json_patch:
+        kind: Node
+        name: "{{ item.metadata.name }}"
+        patch:
+          - op: add
+            path: /spec/taints
+            value: "{{ item.spec.taints | default([]) | rejectattr('key', 'equalto', 'node.kubernetes.io/out-of-service') | list }}"
+          - op: add
+            path: /metadata/annotations
+            value: "{{ item.metadata.annotations | dict2items | rejectattr('key', 'equalto', github_runner_arc_node_recovery_annotation) | items2dict }}"
+      loop: "{{ github_runner_arc_node_recovery_nodes.resources | selectattr('metadata.annotations', 'defined') | selectattr('metadata.annotations', 'contains', github_runner_arc_node_recovery_annotation) | list }}"
+      loop_control:
+        label: "{{ item.metadata.name }}"
diff --git a/roles/github_runner_arc/tasks/main.yml b/roles/github_runner_arc/tasks/main.yml
index 33ab727..02ab53a 100644
--- a/roles/github_runner_arc/tasks/main.yml
+++ b/roles/github_runner_arc/tasks/main.yml
@@ -53,6 +53,9 @@
       ansible.builtin.include_tasks: label_nodes.yml
       when: github_runner_arc_node_label_key | length > 0 and github_runner_arc_labelled_nodes | length > 0
 
+    - name: Install the node recovery watcher, or remove it when turned off
+      ansible.builtin.include_tasks: install_node_recovery.yml
+
     - name: Ensure the fleet-health platform (heartbeat and autoscaler) is installed
       ansible.builtin.include_tasks: install_platform.yml
       when: github_runner_arc_heartbeat_gist_id | length > 0
diff --git a/roles/github_runner_arc/tasks/validate.yml b/roles/github_runner_arc/tasks/validate.yml
index d9115de..e35e4e1 100644
--- a/roles/github_runner_arc/tasks/validate.yml
+++ b/roles/github_runner_arc/tasks/validate.yml
@@ -30,7 +30,7 @@
     github_runner_arc_values_path: "{{ item.values_file if item.values_file is abs else github_runner_arc_values_dir ~ '/' ~ item.values_file }}"
   when: github_runner_arc_values_path is not file or (item.values_file is not abs and github_runner_arc_values_dir | length == 0)
 
-- name: Fail on invalid controller, metrics, probe, pull Secret or node label settings
+- name: Fail on invalid controller, metrics, probe, pull Secret, node recovery or node label settings
   ansible.builtin.fail:
     msg: "{{ item.msg }}"
   loop:
@@ -62,6 +62,13 @@
           and not (github_runner_arc_image_pull_secret_renewal_schedule | string is match('^@[a-z]+$')
                    or github_runner_arc_image_pull_secret_renewal_schedule | string | split | length == 5)
         }}
+    - msg: github_runner_arc_node_recovery_after_seconds and github_runner_arc_node_recovery_poll_seconds must be whole numbers of seconds, at least 1.
+      failed: >-
+        {{
+          github_runner_arc_node_recovery_enabled | bool
+          and ([github_runner_arc_node_recovery_after_seconds, github_runner_arc_node_recovery_poll_seconds]
+               | map('string') | reject('match', '^[1-9][0-9]*$') | list | length > 0)
+        }}
     - msg: github_runner_arc_labelled_nodes needs github_runner_arc_node_label_key.
       failed: "{{ github_runner_arc_labelled_nodes | length > 0 and github_runner_arc_node_label_key | length == 0 }}"
     - msg: github_runner_arc_node_label_value must be set when github_runner_arc_node_label_key is.
diff --git a/roles/github_runner_arc/templates/node-recovery.yaml.j2 b/roles/github_runner_arc/templates/node-recovery.yaml.j2
new file mode 100644
index 0000000..bba1a60
--- /dev/null
+++ b/roles/github_runner_arc/templates/node-recovery.yaml.j2
@@ -0,0 +1,105 @@
+{# The node recovery watcher (see tasks/install_node_recovery.yml and scripts/node-recovery.sh): a ServiceAccount, a ClusterRole allowing only get, list and patch on nodes, its binding, and the Deployment. The namespace is ensured separately, since the platform's other Deployments share it. #}
+{% set name = 'node-recovery' %}
+{% set cluster_name = 'github-runner-node-recovery' %}
+{% set labels = {'app.kubernetes.io/name': name, 'app.kubernetes.io/part-of': 'github-runner'} %}
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: {{ name | to_json }}
+  namespace: {{ github_runner_arc_platform_namespace | to_json }}
+  labels: {{ labels | to_json }}
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: {{ cluster_name | to_json }}
+  labels: {{ labels | to_json }}
+rules:
+  # Nodes are cluster-scoped, and a taint is part of the Node object, so patch cannot be narrower than this. No delete: the watcher never removes a Node, which on a k3s server would remove its etcd member.
+  - apiGroups: [""]
+    resources: ["nodes"]
+    verbs: ["get", "list", "patch"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: {{ cluster_name | to_json }}
+  labels: {{ labels | to_json }}
+subjects:
+  - kind: ServiceAccount
+    name: {{ name | to_json }}
+    namespace: {{ github_runner_arc_platform_namespace | to_json }}
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: {{ cluster_name | to_json }}
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ name | to_json }}
+  namespace: {{ github_runner_arc_platform_namespace | to_json }}
+  labels: {{ labels | to_json }}
+spec:
+  replicas: 1
+  selector:
+    matchLabels: {{ {'app.kubernetes.io/name': name} | to_json }}
+  template:
+    metadata:
+      labels: {{ labels | to_json }}
+    spec:
+      serviceAccountName: {{ name | to_json }}
+      # Unpinned, like the rest of the platform. When the watcher's own node stops, a short toleration gets it evicted within seconds of the node being marked unreachable, so its ReplicaSet starts a replacement elsewhere long before the threshold passes; the default five minutes would leave no watcher running for longer than the threshold it enforces. The old pod stays Terminating until the new watcher taints its node.
+      tolerations:
+        - key: node.kubernetes.io/unreachable
+          operator: Exists
+          effect: NoExecute
+          tolerationSeconds: 10
+        - key: node.kubernetes.io/not-ready
+          operator: Exists
+          effect: NoExecute
+          tolerationSeconds: 10
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 65532
+        runAsGroup: 65532
+        seccompProfile:
+          type: RuntimeDefault
+      containers:
+        - name: node-recovery
+          image: {{ github_runner_arc_node_recovery_image | to_json }}
+          env:
+            - name: NODE_RECOVERY_AFTER_SECONDS
+              value: {{ github_runner_arc_node_recovery_after_seconds | string | to_json }}
+            - name: NODE_RECOVERY_POLL_SECONDS
+              value: {{ github_runner_arc_node_recovery_poll_seconds | string | to_json }}
+            - name: NODE_RECOVERY_ANNOTATION
+              value: {{ github_runner_arc_node_recovery_annotation | to_json }}
+            # The watcher never taints the node it runs on.
+            - name: NODE_RECOVERY_SELF_NODE
+              valueFrom:
+                fieldRef:
+                  fieldPath: spec.nodeName
+            # kubectl's cache goes to the in-memory volume, since the root filesystem is read-only.
+            - name: HOME
+              value: /tmp
+          securityContext:
+            allowPrivilegeEscalation: false
+            readOnlyRootFilesystem: true
+            capabilities:
+              drop: ["ALL"]
+          resources:
+            requests:
+              cpu: 10m
+              memory: 32Mi
+            limits:
+              cpu: 200m
+              memory: 64Mi
+          volumeMounts:
+            - name: tmp
+              mountPath: /tmp
+      volumes:
+        - name: tmp
+          emptyDir:
+            medium: Memory
+            sizeLimit: 8Mi
diff --git a/roles/github_runner_arc/vars/main.yml b/roles/github_runner_arc/vars/main.yml
index b723ec1..ea48352 100644
--- a/roles/github_runner_arc/vars/main.yml
+++ b/roles/github_runner_arc/vars/main.yml
@@ -9,3 +9,6 @@ github_runner_arc_pull_token_expiry_annotation: github-runner.exadev/pull-token-
 
 # Server-side apply field manager the renewal job writes the pull Secret with.
 github_runner_arc_pull_secret_field_manager: github-runner-pull-secret-renewer
+
+# Annotation the node recovery watcher writes alongside the out-of-service taint it applies, recording when it applied it. The watcher removes only a taint carrying this annotation, so an operator's own out-of-service taint is left alone.
+github_runner_arc_node_recovery_annotation: github-runner.exadev/out-of-service-applied-at
diff --git a/scripts/node-recovery.sh b/scripts/node-recovery.sh
new file mode 100755
index 0000000..e7fae3d
--- /dev/null
+++ b/scripts/node-recovery.sh
@@ -0,0 +1,87 @@
+#!/usr/bin/env bash
+# One pass of the node recovery watcher (see roles/github_runner_arc/templates/node-recovery.yaml.j2 and node-recovery/loop.sh, which runs this on an interval).
+#
+# A node whose kubelet has stopped reporting (Ready condition Unknown) keeps its pods forever: the control plane marks them for deletion once their unreachable toleration runs out, but only the kubelet can confirm a deletion, so they stay Terminating. A replacement that has to wait for the old pod to go, such as an ARC listener or a StatefulSet pod, then never starts. Kubernetes' answer is non-graceful node shutdown: the node.kubernetes.io/out-of-service=nodeshutdown:NoExecute taint makes the control plane evict every pod on the node at once and force-delete the ones already terminating there.
+#
+# This pass applies that taint to each node whose Ready condition has been Unknown for at least NODE_RECOVERY_AFTER_SECONDS, and removes it again once the node reports Ready. It only removes a taint it applied itself, recognised by the NODE_RECOVERY_ANNOTATION annotation it writes with it, so an operator's own out-of-service taint is left alone. A node reporting Ready False is never tainted: its kubelet is alive and still finishes its pods' deletion itself. The node this pass runs on (NODE_RECOVERY_SELF_NODE) is never tainted either. Each change is one JSON patch that first tests the node's resourceVersion, so a node whose status changed after it was read (for example because it came back) is left for the next pass instead of being tainted on stale information. It never deletes a Node object, and needs only get, list and patch on nodes.
+set -euo pipefail
+
+after="${NODE_RECOVERY_AFTER_SECONDS:?NODE_RECOVERY_AFTER_SECONDS must be set}"
+annotation="${NODE_RECOVERY_ANNOTATION:-github-runner.exadev/out-of-service-applied-at}"
+self_node="${NODE_RECOVERY_SELF_NODE:-}"
+taint_key="node.kubernetes.io/out-of-service"
+taint_value="nodeshutdown"
+
+if ! [[ "$after" =~ ^[0-9]+$ ]]; then
+  echo "NODE_RECOVERY_AFTER_SECONDS must be a whole number of seconds, not '${after}'" >&2
+  exit 1
+fi
+
+log() { echo "[$(date -u +%Y-%m-%dT%H:%M:%SZ)] $*"; }
+
+nodes="$(kubectl get nodes -o json)"
+now="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+
+# One JSON object per node that needs a change: its name, the action, how long it has been Unknown, and the patch to send.
+plan="$(jq -c \
+  --argjson after "$after" \
+  --arg annotation "$annotation" \
+  --arg self "$self_node" \
+  --arg key "$taint_key" \
+  --arg value "$taint_value" \
+  --arg now "$now" '
+  ($now | fromdateiso8601) as $now_s
+  | .items[]
+  | .metadata.name as $name
+  | ((.status.conditions // []) | map(select(.type == "Ready")) | first) as $ready
+  | (.spec.taints // []) as $taints
+  | ([$taints[] | select(.key == $key)] | length > 0) as $tainted
+  | (.metadata.annotations // {}) as $annotations
+  | ($annotations | has($annotation)) as $ours
+  | {op: "test", path: "/metadata/resourceVersion", value: .metadata.resourceVersion} as $test
+  | if $name == $self or $ready == null then empty
+    elif $ready.status == "Unknown" and ($tainted | not)
+         and ($now_s - ($ready.lastTransitionTime | fromdateiso8601)) >= $after then
+      {
+        name: $name,
+        action: "taint",
+        unknown_for: ($now_s - ($ready.lastTransitionTime | fromdateiso8601)),
+        patch: [
+          $test,
+          {op: "add", path: "/metadata/annotations", value: ($annotations + {($annotation): $now})},
+          {op: "add", path: "/spec/taints", value: ($taints + [{key: $key, value: $value, effect: "NoExecute", timeAdded: $now}])}
+        ]
+      }
+    elif $ready.status == "True" and $ours then
+      {
+        name: $name,
+        action: "clear",
+        patch: [
+          $test,
+          {op: "add", path: "/metadata/annotations", value: ($annotations | del(.[$annotation]))},
+          {op: "add", path: "/spec/taints", value: [$taints[] | select(.key != $key)]}
+        ]
+      }
+    else empty
+    end
+' <<< "$nodes")"
+
+exit_code=0
+while IFS= read -r change; do
+  [ -n "$change" ] || continue
+  name="$(jq -r '.name' <<< "$change")"
+  action="$(jq -r '.action' <<< "$change")"
+  patch="$(jq -c '.patch' <<< "$change")"
+  if kubectl patch node "$name" --type=json -p "$patch" >/dev/null; then
+    if [ "$action" = taint ]; then
+      log "Tainted ${name} ${taint_key}=${taint_value}:NoExecute: not reporting for $(jq -r '.unknown_for' <<< "$change")s, so its pods are evicted and force-deleted"
+    else
+      log "Removed ${taint_key} from ${name}: it reports Ready again"
+    fi
+  else
+    log "Could not ${action} ${name}; retrying on the next pass" >&2
+    exit_code=1
+  fi
+done <<< "$plan"
+
+exit "$exit_code"
diff --git a/scripts/release/stamp_version.py b/scripts/release/stamp_version.py
index 6974fb8..6d0aae6 100755
--- a/scripts/release/stamp_version.py
+++ b/scripts/release/stamp_version.py
@@ -21,6 +21,7 @@
     "github_runner_arc_heartbeat_image",
     "github_runner_arc_autoscaler_image",
     "github_runner_arc_image_pull_secret_renewer_image",
+    "github_runner_arc_node_recovery_image",
 )
 
 
diff --git a/tests/node_recovery/render.yml b/tests/node_recovery/render.yml
new file mode 100644
index 0000000..a14ed63
--- /dev/null
+++ b/tests/node_recovery/render.yml
@@ -0,0 +1,14 @@
+# Renders the github_runner_arc role's node recovery manifests (templates/node-recovery.yaml.j2) with the role's defaults and constants, for tests/node_recovery/run.sh to apply to a test cluster. Takes recovery_output as an extra variable; a role variable such as the image or the threshold is overridden with an extra variable of its own name, since the defaults file loaded here outranks play variables.
+- name: Render the node recovery manifests
+  hosts: localhost
+  connection: local
+  gather_facts: false
+  vars_files:
+    - ../../roles/github_runner_arc/defaults/main.yml
+    - ../../roles/github_runner_arc/vars/main.yml
+  tasks:
+    - name: Write the manifests
+      ansible.builtin.copy:
+        content: "{{ lookup('ansible.builtin.template', playbook_dir ~ '/../../roles/github_runner_arc/templates/node-recovery.yaml.j2') }}"
+        dest: "{{ recovery_output }}"
+        mode: "0644"
diff --git a/tests/node_recovery/run.sh b/tests/node_recovery/run.sh
new file mode 100755
index 0000000..01aab5e
--- /dev/null
+++ b/tests/node_recovery/run.sh
@@ -0,0 +1,222 @@
+#!/usr/bin/env bash
+# Integration test for the github_runner_arc role's node recovery watcher: renders the role's manifests, applies them to a three-node kind cluster with the watcher image built from this checkout, and stops one worker's container to stand in for a node that goes down. A single-replica StatefulSet stands in for an ARC listener: like the listener, its controller waits for the old pod to go before it creates a new one, so a pod stuck Terminating on a stopped node stalls it the same way.
+#
+# First, with the watcher scaled to zero, it shows the failure: the stand-in's pod on the stopped node stays Terminating and is never replaced. Then the watcher taints that node out of service and the pod is replaced on a healthy node; the worker is started again and the taint is removed. Finally, with the watcher running throughout and the stand-in on default tolerations, it stops the worker again and asserts the pod is replaced within the node monitor grace period plus the threshold plus a margin, well before Kubernetes' own five-minute eviction, and not before the threshold. Also checks the watcher's permissions, that it never deletes a Node and that it leaves healthy nodes alone.
+#
+# Usage: tests/node_recovery/run.sh. Needs Docker, kind, kubectl, jq and ansible-playbook (ANSIBLE_PLAYBOOK overrides which). Creates a kind cluster named grtest-recovery and removes it on exit unless GRTEST_KEEP=1.
+set -euo pipefail
+
+repo_root="$(cd "$(dirname "$0")/../.." && pwd)"
+ansible_playbook="${ANSIBLE_PLAYBOOK:-ansible-playbook}"
+cluster=grtest-recovery
+platform_namespace=github-runner-platform
+namespace=grtest-listeners
+image=grtest/node-recovery:test
+pod_image=busybox:1.36
+watcher_node="${cluster}-worker"
+doomed_node="${cluster}-worker2"
+after_seconds=30
+poll_seconds=5
+# kube-controller-manager's default node-monitor-grace-period: 50s from Kubernetes 1.32, 40s before.
+grace_seconds=50
+# The pod garbage collector's sweep interval, how long the StatefulSet takes to schedule and start the replacement, and slack for a busy CI runner.
+margin_seconds=60
+recovery_bound=$((grace_seconds + after_seconds + poll_seconds + margin_seconds))
+# How long the failure is watched for, with the watcher off, before it is taken as not recovering on its own.
+stuck_watch_seconds=60
+out_of_service=node.kubernetes.io/out-of-service
+annotation=github-runner.exadev/out-of-service-applied-at
+work="$(mktemp -d "${TMPDIR:-/tmp}/grtest-recovery.XXXXXX")"
+export KUBECONFIG="$work/kubeconfig"
+
+log() { echo "==> $*"; }
+fail() {
+  echo "FAIL: $*" >&2
+  kubectl get nodes -o wide >&2 || true
+  kubectl get nodes -o json | jq -r '.items[] | "\(.metadata.name) taints=\(.spec.taints // [] | map(.key) | join(",")) annotations=\(.metadata.annotations // {} | keys | join(","))"' >&2 || true
+  kubectl get pods -A -o wide >&2 || true
+  kubectl -n "$platform_namespace" logs deploy/node-recovery --tail=40 >&2 || true
+  exit 1
+}
+cleanup() {
+  if [ "${GRTEST_KEEP:-0}" != 1 ]; then
+    kind delete cluster --name "$cluster" >/dev/null 2>&1 || true
+    rm -rf "$work"
+  fi
+}
+trap cleanup EXIT
+
+# Polls a condition command every two seconds until it succeeds or the timeout passes.
+wait_for() {
+  local timeout="$1" description="$2"
+  shift 2
+  local deadline=$(( $(date +%s) + timeout ))
+  until "$@"; do
+    [ "$(date +%s)" -lt "$deadline" ] || fail "timed out after ${timeout}s waiting for ${description}"
+    sleep 2
+  done
+}
+
+node_ready() { kubectl get node "$1" -o jsonpath='{.status.conditions[?(@.type=="Ready")].status}'; }
+node_is() { [ "$(node_ready "$1")" = "$2" ]; }
+has_taint() { kubectl get node "$1" -o json | jq -e --arg key "$out_of_service" '[.spec.taints // [] | .[] | select(.key == $key)] | length > 0' >/dev/null; }
+lacks_taint() { ! has_taint "$1"; }
+lacks_annotation() { kubectl get node "$1" -o json | jq -e --arg a "$annotation" '(.metadata.annotations // {}) | has($a) | not' >/dev/null; }
+pod_json() { kubectl -n "$namespace" get pod "$1" -o json 2>/dev/null; }
+pod_field() { pod_json "$1" | jq -r "$2"; }
+pod_terminating() { [ "$(pod_field "$1" '.metadata.deletionTimestamp // ""')" != "" ]; }
+# Whether the pod exists as a different pod from the given uid, running on a node other than the doomed one.
+pod_replaced() {
+  local json
+  json="$(pod_json "$1")" || return 1
+  [ "$(jq -r '.metadata.uid' <<< "$json")" != "$2" ] \
+    && [ "$(jq -r '.spec.nodeName // ""' <<< "$json")" != "$doomed_node" ] \
+    && [ "$(jq -r '.status.phase' <<< "$json")" = Running ] \
+    && [ "$(jq -r '.metadata.deletionTimestamp // ""' <<< "$json")" = "" ]
+}
+
+# A single-replica StatefulSet preferring the doomed node, standing in for an ARC listener. Extra tolerations are passed as a JSON list.
+listener() {
+  local name="$1" tolerations="$2"
+  kubectl apply -f - >/dev/null <<EOF
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: ${name}
+  namespace: ${namespace}
+spec:
+  replicas: 1
+  serviceName: ${name}
+  selector:
+    matchLabels: {app: ${name}}
+  template:
+    metadata:
+      labels: {app: ${name}}
+    spec:
+      terminationGracePeriodSeconds: 5
+      tolerations: ${tolerations}
+      affinity:
+        nodeAffinity:
+          preferredDuringSchedulingIgnoredDuringExecution:
+            - weight: 100
+              preference:
+                matchExpressions:
+                  - {key: kubernetes.io/hostname, operator: In, values: [${doomed_node}]}
+      containers:
+        - name: listener
+          image: ${pod_image}
+          imagePullPolicy: IfNotPresent
+          command: [sleep, "86400"]
+EOF
+  wait_for 120 "${name}-0 to run on ${doomed_node}" pod_running_on "${name}-0" "$doomed_node"
+}
+pod_running_on() { [ "$(pod_field "$1" '.status.phase + " " + (.spec.nodeName // "")' 2>/dev/null)" = "Running $2" ]; }
+
+stop_doomed() {
+  docker stop "$doomed_node" >/dev/null
+  stopped_at="$(date +%s)"
+}
+
+log "Creating kind cluster $cluster"
+cat > "$work/kind.yaml" <<EOF
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+nodes:
+  - role: control-plane
+  - role: worker
+  - role: worker
+EOF
+kind create cluster --name "$cluster" --config "$work/kind.yaml" --kubeconfig "$KUBECONFIG" --wait 180s >/dev/null
+server_version="$(kubectl version -o json | jq -r '.serverVersion.gitVersion')"
+log "Server version ${server_version}"
+doomed_uid="$(kubectl get node "$doomed_node" -o jsonpath='{.metadata.uid}')"
+
+log "Building and loading the watcher image and the stand-in's image"
+docker build -q -f "$repo_root/node-recovery/Dockerfile" -t "$image" "$repo_root" >/dev/null
+docker pull -q "$pod_image" >/dev/null
+kind load docker-image "$image" "$pod_image" --name "$cluster" >/dev/null
+
+log "Rendering and applying the role's node recovery manifests"
+ANSIBLE_COLLECTIONS_PATH="$repo_root/playbooks/collections${ANSIBLE_COLLECTIONS_PATH:+:$ANSIBLE_COLLECTIONS_PATH}" \
+  "$ansible_playbook" "$repo_root/tests/node_recovery/render.yml" \
+  -e github_runner_arc_node_recovery_image="$image" \
+  -e github_runner_arc_node_recovery_after_seconds="$after_seconds" \
+  -e github_runner_arc_node_recovery_poll_seconds="$poll_seconds" \
+  -e recovery_output="$work/recovery.yaml" >/dev/null
+kubectl create namespace "$platform_namespace" >/dev/null
+kubectl create namespace "$namespace" >/dev/null
+kubectl apply -f "$work/recovery.yaml" >/dev/null
+# The test keeps the watcher off the node it stops, so each phase measures the taint rather than the watcher's own rescheduling; everything else is as the role renders it.
+kubectl -n "$platform_namespace" patch deployment node-recovery --type=merge \
+  -p "{\"spec\": {\"template\": {\"spec\": {\"nodeSelector\": {\"kubernetes.io/hostname\": \"${watcher_node}\"}}}}}" >/dev/null
+kubectl -n "$platform_namespace" rollout status deployment/node-recovery --timeout=120s >/dev/null
+
+log "Checking the watcher's permissions"
+as="system:serviceaccount:${platform_namespace}:node-recovery"
+check_can() {
+  local expected="$1"
+  shift
+  local answer
+  answer="$(kubectl auth can-i --as="$as" "$@" 2>/dev/null || true)"
+  [ "$answer" = "$expected" ] || fail "can-i $* answered '$answer', expected '$expected'"
+}
+check_can yes get nodes
+check_can yes list nodes
+check_can yes patch nodes
+check_can no delete nodes
+check_can no create nodes
+check_can no delete pods --all-namespaces
+check_can no get secrets --all-namespaces
+
+log "Phase 1: with the watcher off, a pod on a stopped node stays Terminating and is never replaced"
+kubectl -n "$platform_namespace" scale deployment node-recovery --replicas=0 >/dev/null
+wait_for 60 "the watcher to stop" bash -c "[ -z \"\$(kubectl -n $platform_namespace get pods -l app.kubernetes.io/name=node-recovery -o name)\" ]"
+# A short toleration gets the pod marked for deletion soon after the node is marked unreachable, the state a listener reaches after the default five minutes, without waiting them out.
+listener stuck '[{"key": "node.kubernetes.io/unreachable", "operator": "Exists", "effect": "NoExecute", "tolerationSeconds": 5}, {"key": "node.kubernetes.io/not-ready", "operator": "Exists", "effect": "NoExecute", "tolerationSeconds": 5}]'
+stuck_uid="$(pod_field stuck-0 '.metadata.uid')"
+stop_doomed
+wait_for $((grace_seconds + margin_seconds)) "${doomed_node} to be marked Unknown" node_is "$doomed_node" Unknown
+wait_for 60 "stuck-0 to be marked for deletion" pod_terminating stuck-0
+log "stuck-0 is Terminating; watching it for ${stuck_watch_seconds}s"
+sleep "$stuck_watch_seconds"
+[ "$(pod_field stuck-0 '.metadata.uid')" = "$stuck_uid" ] || fail "stuck-0 was replaced without the watcher, so the test does not reproduce the failure"
+pod_terminating stuck-0 || fail "stuck-0 is no longer Terminating"
+log "stuck-0 is still Terminating on ${doomed_node} and has not been replaced"
+
+log "Phase 1: starting the watcher clears it"
+started_at="$(date +%s)"
+kubectl -n "$platform_namespace" scale deployment node-recovery --replicas=1 >/dev/null
+wait_for $((poll_seconds + margin_seconds)) "${doomed_node} to be tainted out of service" has_taint "$doomed_node"
+wait_for $((poll_seconds + margin_seconds)) "stuck-0 to be replaced on a healthy node" pod_replaced stuck-0 "$stuck_uid"
+log "stuck-0 replaced on $(pod_field stuck-0 '.spec.nodeName') $(( $(date +%s) - started_at ))s after the watcher started"
+
+log "Phase 2: the taint is removed once the node is back"
+docker start "$doomed_node" >/dev/null
+wait_for 240 "${doomed_node} to report Ready" node_is "$doomed_node" True
+wait_for $((poll_seconds * 4 + 10)) "the taint to be removed from ${doomed_node}" lacks_taint "$doomed_node"
+wait_for 20 "the watcher's annotation to be removed from ${doomed_node}" lacks_annotation "$doomed_node"
+[ "$(kubectl get node "$doomed_node" -o jsonpath='{.metadata.uid}')" = "$doomed_uid" ] || fail "the Node object of ${doomed_node} was replaced"
+log "Taint and annotation removed; the Node object is the original one"
+
+log "Phase 3: with the watcher running and default tolerations, a stopped node is recovered within ${recovery_bound}s"
+listener bounded '[]'
+bounded_uid="$(pod_field bounded-0 '.metadata.uid')"
+stop_doomed
+wait_for "$recovery_bound" "bounded-0 to be replaced on a healthy node" pod_replaced bounded-0 "$bounded_uid"
+elapsed=$(( $(date +%s) - stopped_at ))
+log "bounded-0 replaced on $(pod_field bounded-0 '.spec.nodeName') ${elapsed}s after ${doomed_node} stopped (bound ${recovery_bound}s; Kubernetes alone would only mark it for deletion after ${grace_seconds}s + 300s)"
+node_json="$(kubectl get node "$doomed_node" -o json)"
+unknown_since="$(jq -r '.status.conditions[] | select(.type == "Ready") | .lastTransitionTime' <<< "$node_json")"
+tainted_at="$(jq -r --arg a "$annotation" '.metadata.annotations[$a]' <<< "$node_json")"
+waited=$(( $(jq -rn --arg t "$tainted_at" '$t | fromdateiso8601') - $(jq -rn --arg t "$unknown_since" '$t | fromdateiso8601') ))
+[ "$waited" -ge "$after_seconds" ] || fail "the watcher tainted ${doomed_node} ${waited}s after it went Unknown, before the ${after_seconds}s threshold"
+log "The watcher waited ${waited}s after ${doomed_node} went Unknown before tainting it (threshold ${after_seconds}s)"
+
+log "Checking the healthy nodes were left alone"
+for healthy in "${cluster}-control-plane" "$watcher_node"; do
+  lacks_taint "$healthy" || fail "${healthy} was tainted out of service"
+  lacks_annotation "$healthy" || fail "${healthy} carries the watcher's annotation"
+done
+[ "$(kubectl get node "$doomed_node" -o jsonpath='{.metadata.uid}')" = "$doomed_uid" ] || fail "the Node object of ${doomed_node} was replaced"
+
+log "PASS"
diff --git a/tests/unit/test_node_recovery.py b/tests/unit/test_node_recovery.py
new file mode 100644
index 0000000..771427d
--- /dev/null
+++ b/tests/unit/test_node_recovery.py
@@ -0,0 +1,186 @@
+"""Tests for scripts/node-recovery.sh, one pass of the node recovery watcher, run against a scripted stand-in for kubectl with the real bash and jq."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import tempfile
+import textwrap
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+SCRIPT = Path(__file__).resolve().parents[2] / "scripts" / "node-recovery.sh"
+ANNOTATION = "github-runner.exadev/out-of-service-applied-at"
+OUT_OF_SERVICE = "node.kubernetes.io/out-of-service"
+AFTER_SECONDS = 120
+
+# Answers `kubectl get nodes -o json` from a file and records every `kubectl patch`; FAKE_PATCH_EXIT sets the patch's exit status.
+FAKE_KUBECTL = textwrap.dedent(
+    """\
+    #!/usr/bin/env python3
+    import json, os, sys
+    args = sys.argv[1:]
+    if args[:2] == ["get", "nodes"]:
+        sys.stdout.write(open(os.environ["FAKE_NODES"]).read())
+        sys.exit(0)
+    if args[:2] == ["patch", "node"]:
+        with open(os.environ["FAKE_PATCHES"], "a") as log:
+            log.write(json.dumps({"name": args[2], "argv": args}) + "\\n")
+        sys.exit(int(os.environ.get("FAKE_PATCH_EXIT", "0")))
+    sys.stderr.write("unexpected kubectl call: %s\\n" % args)
+    sys.exit(2)
+    """
+)
+
+
+def timestamp(seconds_ago: int) -> str:
+    return (datetime.now(timezone.utc) - timedelta(seconds=seconds_ago)).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def node(
+    name: str,
+    ready: str,
+    seconds_ago: int,
+    taints: list[dict[str, str]] | None = None,
+    annotations: dict[str, str] | None = None,
+) -> dict[str, object]:
+    return {
+        "metadata": {"name": name, "resourceVersion": f"rv-{name}", "annotations": annotations or {}},
+        "spec": {"taints": taints or []},
+        "status": {"conditions": [{"type": "Ready", "status": ready, "lastTransitionTime": timestamp(seconds_ago)}]},
+    }
+
+
+UNREACHABLE = {"key": "node.kubernetes.io/unreachable", "effect": "NoExecute"}
+OURS = {"key": OUT_OF_SERVICE, "value": "nodeshutdown", "effect": "NoExecute"}
+
+
+class NodeRecoveryTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        for tool in ("bash", "jq"):
+            if shutil.which(tool) is None:
+                raise unittest.SkipTest(f"{tool} is not installed")
+
+    def setUp(self) -> None:
+        self.work = Path(tempfile.mkdtemp())
+        self.addCleanup(shutil.rmtree, self.work)
+        bin_dir = self.work / "bin"
+        bin_dir.mkdir()
+        kubectl = bin_dir / "kubectl"
+        kubectl.write_text(FAKE_KUBECTL)
+        kubectl.chmod(0o755)
+        self.nodes_file = self.work / "nodes.json"
+        self.patches_file = self.work / "patches.jsonl"
+        self.env = {
+            **os.environ,
+            "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}",
+            "FAKE_NODES": str(self.nodes_file),
+            "FAKE_PATCHES": str(self.patches_file),
+            "NODE_RECOVERY_AFTER_SECONDS": str(AFTER_SECONDS),
+            "NODE_RECOVERY_ANNOTATION": ANNOTATION,
+            "NODE_RECOVERY_SELF_NODE": "watcher-node",
+        }
+
+    def run_pass(self, nodes: list[dict[str, object]], **env: str) -> tuple[subprocess.CompletedProcess[str], list[dict[str, object]]]:
+        self.nodes_file.write_text(json.dumps({"items": nodes}))
+        result = subprocess.run(["bash", str(SCRIPT)], env={**self.env, **env}, capture_output=True, text=True, check=False)
+        patches = []
+        if self.patches_file.exists():
+            for line in self.patches_file.read_text().splitlines():
+                call = json.loads(line)
+                argv = call["argv"]
+                self.assertEqual(argv[3], "--type=json")
+                self.assertEqual(argv[4], "-p")
+                patches.append({"name": call["name"], "patch": json.loads(argv[5])})
+        return result, patches
+
+    @staticmethod
+    def value_at(patch: list[dict[str, object]], path: str) -> object:
+        return next(op["value"] for op in patch if op["path"] == path)
+
+    def test_taints_a_node_unknown_for_longer_than_the_threshold(self) -> None:
+        result, patches = self.run_pass([node("dead", "Unknown", AFTER_SECONDS + 30, taints=[UNREACHABLE], annotations={"keep": "me"})])
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual([p["name"] for p in patches], ["dead"])
+        patch = patches[0]["patch"]
+        self.assertEqual(patch[0], {"op": "test", "path": "/metadata/resourceVersion", "value": "rv-dead"})
+        taints = self.value_at(patch, "/spec/taints")
+        self.assertIn(UNREACHABLE, taints)
+        added = [t for t in taints if t["key"] == OUT_OF_SERVICE]
+        self.assertEqual(len(added), 1)
+        self.assertEqual((added[0]["value"], added[0]["effect"]), ("nodeshutdown", "NoExecute"))
+        annotations = self.value_at(patch, "/metadata/annotations")
+        self.assertEqual(annotations["keep"], "me")
+        self.assertIn(ANNOTATION, annotations)
+        self.assertIn("Tainted dead", result.stdout)
+
+    def test_leaves_a_node_unknown_for_less_than_the_threshold(self) -> None:
+        result, patches = self.run_pass([node("blip", "Unknown", AFTER_SECONDS - 30, taints=[UNREACHABLE])])
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(patches, [])
+
+    def test_leaves_a_node_whose_kubelet_reports_not_ready(self) -> None:
+        _, patches = self.run_pass([node("sick", "False", AFTER_SECONDS * 10)])
+        self.assertEqual(patches, [])
+
+    def test_never_taints_its_own_node(self) -> None:
+        _, patches = self.run_pass([node("watcher-node", "Unknown", AFTER_SECONDS * 10)])
+        self.assertEqual(patches, [])
+
+    def test_leaves_an_out_of_service_taint_it_did_not_apply(self) -> None:
+        operator_taint = {"key": OUT_OF_SERVICE, "value": "maintenance", "effect": "NoExecute"}
+        _, patches = self.run_pass(
+            [
+                node("operator-dead", "Unknown", AFTER_SECONDS * 10, taints=[operator_taint]),
+                node("operator-back", "True", 5, taints=[operator_taint]),
+            ]
+        )
+        self.assertEqual(patches, [])
+
+    def test_does_not_taint_a_node_twice(self) -> None:
+        _, patches = self.run_pass([node("dead", "Unknown", AFTER_SECONDS * 10, taints=[UNREACHABLE, OURS], annotations={ANNOTATION: timestamp(60)})])
+        self.assertEqual(patches, [])
+
+    def test_clears_its_taint_once_the_node_is_ready(self) -> None:
+        other = {"key": "dedicated", "value": "builds", "effect": "NoSchedule"}
+        result, patches = self.run_pass([node("back", "True", 5, taints=[other, OURS], annotations={ANNOTATION: timestamp(600), "keep": "me"})])
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual([p["name"] for p in patches], ["back"])
+        patch = patches[0]["patch"]
+        self.assertEqual(patch[0], {"op": "test", "path": "/metadata/resourceVersion", "value": "rv-back"})
+        self.assertEqual(self.value_at(patch, "/spec/taints"), [other])
+        self.assertEqual(self.value_at(patch, "/metadata/annotations"), {"keep": "me"})
+        self.assertIn("Removed", result.stdout)
+
+    def test_leaves_healthy_nodes_alone(self) -> None:
+        _, patches = self.run_pass([node("healthy", "True", 3600), node("fresh", "True", 1)])
+        self.assertEqual(patches, [])
+
+    def test_patches_every_node_that_needs_it(self) -> None:
+        _, patches = self.run_pass(
+            [
+                node("dead", "Unknown", AFTER_SECONDS + 1),
+                node("healthy", "True", 3600),
+                node("back", "True", 5, taints=[OURS], annotations={ANNOTATION: timestamp(600)}),
+            ]
+        )
+        self.assertEqual(sorted(p["name"] for p in patches), ["back", "dead"])
+
+    def test_a_refused_patch_fails_the_pass(self) -> None:
+        result, patches = self.run_pass([node("dead", "Unknown", AFTER_SECONDS * 10)], FAKE_PATCH_EXIT="1")
+        self.assertEqual(len(patches), 1)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("retrying on the next pass", result.stderr)
+
+    def test_rejects_a_threshold_that_is_not_a_number(self) -> None:
+        result, patches = self.run_pass([node("dead", "Unknown", AFTER_SECONDS * 10)], NODE_RECOVERY_AFTER_SECONDS="two minutes")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(patches, [])
+
+
+if __name__ == "__main__":
+    unittest.main()