diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml
deleted file mode 100644
index 3f014e19107..00000000000
--- a/.github/workflows/codeql-analysis.yml
+++ /dev/null
@@ -1,71 +0,0 @@
-# For most projects, this workflow file will not need changing; you simply need
-# to commit it to your repository.
-#
-# You may wish to alter this file to override the set of languages analyzed,
-# or to provide custom queries or build logic.
-#
-# ******** NOTE ********
-# We have attempted to detect the languages in your repository. Please check
-# the `language` matrix defined below to confirm you have the correct set of
-# supported CodeQL languages.
-#
-name: "CodeQL"
-
-on:
-  push:
-    branches: [ main ]
-  pull_request:
-    # The branches below must be a subset of the branches above
-    branches: [ main ]
-  workflow_dispatch:
-
-
-
-jobs:
-  analyze:
-    name: Analyze
-    runs-on: blacksmith-4vcpu-ubuntu-2204
-    permissions:
-      actions: read
-      contents: read
-      security-events: write
-
-    strategy:
-      fail-fast: false
-      matrix:
-        language: [ 'python' ]
-        # CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python', 'ruby' ]
-        # Learn more about CodeQL language support at https://git.io/codeql-language-support
-
-    steps:
-    - name: Checkout repository
-      uses: actions/checkout@v6
-
-    # Initializes the CodeQL tools for scanning.
-    - name: Initialize CodeQL
-      uses: github/codeql-action/init@v4
-      with:
-        languages: ${{ matrix.language }}
-        # If you wish to specify custom queries, you can do so here or in a config file.
-        # By default, queries listed here will override any specified in a config file.
-        # Prefix the list here with "+" to use these queries and those in the config file.
-        # queries: ./path/to/local/query, your-org/your-repo/queries@main
-
-    # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
-    # If this step fails, then you should remove it and run the build manually (see below)
-    - name: Autobuild
-      uses: github/codeql-action/autobuild@v4
-
-    # ℹ️ Command-line programs to run using the OS shell.
-    # 📚 https://git.io/JvXDl
-
-    # ✏️ If the Autobuild fails above, remove it and uncomment the following three lines
-    #    and modify them (or add more) to build your code if your project
-    #    uses a compiled language
-
-    #- run: |
-    #   make bootstrap
-    #   make release
-
-    - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v4
diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml
index 43fde7bb735..9ecc90d008e 100644
--- a/.github/workflows/docker.yml
+++ b/.github/workflows/docker.yml
@@ -26,51 +26,51 @@ on:
 
 jobs:
   docker-build:
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     timeout-minutes: 60
     steps:
     - uses: actions/checkout@v6
     - name: Setup Docker Builder
-      uses: useblacksmith/setup-docker-builder@v1
+      uses: docker/setup-buildx-action@v3
     - name: Build Docker container
       run: make docker-build
 
   smoke-tests_dev:
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     timeout-minutes: 10
     steps:
     - uses: actions/checkout@v6
     - name: Setup Docker Builder
-      uses: useblacksmith/setup-docker-builder@v1
+      uses: docker/setup-buildx-action@v3
     - name: "Build smoke tests container: dev"
       run: docker build -f t/smoke/workers/docker/dev .
 
   smoke-tests_latest:
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     timeout-minutes: 10
     steps:
     - uses: actions/checkout@v6
     - name: Setup Docker Builder
-      uses: useblacksmith/setup-docker-builder@v1
+      uses: docker/setup-buildx-action@v3
     - name: "Build smoke tests container: latest"
       run: docker build -f t/smoke/workers/docker/pypi .
 
   smoke-tests_pypi:
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     timeout-minutes: 10
     steps:
     - uses: actions/checkout@v6
     - name: Setup Docker Builder
-      uses: useblacksmith/setup-docker-builder@v1
+      uses: docker/setup-buildx-action@v3
     - name: "Build smoke tests container: pypi"
       run: docker build -f t/smoke/workers/docker/pypi --build-arg CELERY_VERSION="5" .
 
   smoke-tests_legacy:
-    runs-on: blacksmith-4vcpu-ubuntu-2204
+    runs-on: ubuntu-22.04
     timeout-minutes: 10
     steps:
     - uses: actions/checkout@v6
     - name: Setup Docker Builder
-      uses: useblacksmith/setup-docker-builder@v1
+      uses: docker/setup-buildx-action@v3
     - name: "Build smoke tests container: legacy"
       run: docker build -f t/smoke/workers/docker/pypi --build-arg CELERY_VERSION="4" .
diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml
index e3f411bc7ea..2875162dc5c 100644
--- a/.github/workflows/integration-tests.yml
+++ b/.github/workflows/integration-tests.yml
@@ -11,7 +11,7 @@ on:
         description: 'JSON array of Python versions to test'
         required: false
         type: string
-        default: '["3.9", "3.10", "3.11", "3.12", "3.13", "3.14", "3.14t"]'
+        default: '["3.10", "3.11", "3.12", "3.13", "3.14", "3.14t"]'
       tox_environments:
         description: 'JSON array of tox environments to test'
         required: false
@@ -21,7 +21,7 @@ on:
 jobs:
   testing-with:
     timeout-minutes: 240
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     strategy:
       fail-fast: false
       matrix:
@@ -48,11 +48,11 @@ jobs:
     steps:
       - name: Install apt packages
         run: |
-          sudo apt-get update && sudo apt-get install -f libcurl4-openssl-dev libssl-dev libgnutls28-dev httping expect libmemcached-dev
+          sudo apt-get update && sudo apt-get install -f libcurl4-openssl-dev libev-dev libssl-dev libgnutls28-dev httping expect libmemcached-dev
 
       - uses: actions/checkout@v6
       - name: Set up Python ${{ matrix.python-version }}
-        uses: useblacksmith/setup-python@v6
+        uses: actions/setup-python@v6
         with:
           python-version: ${{ matrix.python-version }}
           allow-prereleases: true
diff --git a/.github/workflows/linter.yml b/.github/workflows/linter.yml
index 249592d4fb3..dd222114dc1 100644
--- a/.github/workflows/linter.yml
+++ b/.github/workflows/linter.yml
@@ -7,7 +7,7 @@ permissions:
 
 jobs:
   linter:
-    runs-on: blacksmith-4vcpu-ubuntu-2204
+    runs-on: ubuntu-22.04
     steps:
 
       - name: Checkout branch
diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
index 997585a0b9a..dbc923e6d80 100644
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/python-package.yml
@@ -33,11 +33,9 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.9', '3.10', '3.11', '3.12', '3.13', '3.14', 'pypy3.11']
-        os: ["blacksmith-4vcpu-ubuntu-2404", "windows-latest"]
+        python-version: ['3.10', '3.11', '3.12', '3.13', '3.14', 'pypy3.11']
+        os: ["ubuntu-24.04", "windows-latest"]
         exclude:
-            - python-version: '3.9'
-              os: "windows-latest"
             - python-version: '3.10'
               os: "windows-latest"
             - python-version: '3.11'
@@ -53,12 +51,12 @@ jobs:
 
     steps:
     - name: Install apt packages
-      if: startsWith(matrix.os, 'blacksmith-4vcpu-ubuntu')
+      if: startsWith(matrix.os, 'ubuntu')
       run: |
-        sudo apt-get update && sudo apt-get install -f libcurl4-openssl-dev libssl-dev libgnutls28-dev httping expect libmemcached-dev
+        sudo apt-get update && sudo apt-get install -f libcurl4-openssl-dev libev-dev libssl-dev libgnutls28-dev httping expect libmemcached-dev
     - uses: actions/checkout@v6
     - name: Set up Python ${{ matrix.python-version }}
-      uses: useblacksmith/setup-python@v6
+      uses: actions/setup-python@v6
       with:
         python-version: ${{ matrix.python-version }}
         allow-prereleases: true
@@ -77,7 +75,7 @@ jobs:
     - uses: codecov/codecov-action@v5
       with:
         flags: unittests # optional
-        fail_ci_if_error: true # optional (default = false)
+        fail_ci_if_error: false # fork has no CODECOV_TOKEN
         token: ${{ secrets.CODECOV_TOKEN }}
         verbose: true # optional (default = false)
 
@@ -87,9 +85,11 @@ jobs:
       with:
         token: ${{ secrets.CODECOV_TOKEN }}
 
+  # Integration/smoke fan out to ~370 jobs and need pytest-celery's PyPI celery
+  # image, so on the fork they only run on manual dispatch.
   Integration-tests:
     needs: [Unit]
-    if: needs.Unit.result == 'success'
+    if: needs.Unit.result == 'success' && github.event_name == 'workflow_dispatch'
     strategy:
       matrix:
         module: [
@@ -113,7 +113,7 @@ jobs:
 
   Smoke-tests:
     needs: [Unit]
-    if: needs.Unit.result == 'success'
+    if: needs.Unit.result == 'success' && github.event_name == 'workflow_dispatch'
     strategy:
       matrix:
         module: [
diff --git a/.github/workflows/smoke-tests.yml b/.github/workflows/smoke-tests.yml
index d40307c578d..6a516c98cb3 100644
--- a/.github/workflows/smoke-tests.yml
+++ b/.github/workflows/smoke-tests.yml
@@ -11,14 +11,14 @@ on:
         description: 'JSON array of Python versions to test'
         required: false
         type: string
-        default: '["3.9", "3.10", "3.11", "3.12", "3.13", "3.14", "3.14t"]'
+        default: '["3.10", "3.11", "3.12", "3.13", "3.14", "3.14t"]'
 
 permissions:
   contents: read
 
 jobs:
   testing-with:
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     strategy:
       fail-fast: false
       matrix:
@@ -38,9 +38,9 @@ jobs:
 
       - uses: actions/checkout@v6
       - name: Setup Docker Builder
-        uses: useblacksmith/setup-docker-builder@v1
+        uses: docker/setup-buildx-action@v3
       - name: Set up Python ${{ matrix.python-version }}
-        uses: useblacksmith/setup-python@v6
+        uses: actions/setup-python@v6
         with:
           python-version: ${{ matrix.python-version }}
           allow-prereleases: true
diff --git a/celery/__init__.py b/celery/__init__.py
index 3467851685f..07281e693f2 100644
--- a/celery/__init__.py
+++ b/celery/__init__.py
@@ -24,7 +24,7 @@
 
 SERIES = 'recovery'
 
-__version__ = '5.6.3+gumloop_0.2.0'
+__version__ = '5.6.3+gumloop_0.2.1'
 __author__ = 'Rahul Behal'
 __contact__ = 'rahul@gumloop.com'
 __homepage__ = 'https://github.com/gumloop/gumloop-celery'
diff --git a/celery/concurrency/asynpool.py b/celery/concurrency/asynpool.py
index 10783847a96..deac56fa0fa 100644
--- a/celery/concurrency/asynpool.py
+++ b/celery/concurrency/asynpool.py
@@ -1080,7 +1080,17 @@ def flush(self):
             self.outbound_buffer.clear()
             self._active_writers.clear()
             self._active_writes.clear()
-            self._busy_workers.clear()
+
+            # _write_to is only set once the body has been written; a dead
+            # worker's inqW_fd may already have been reused by its replacement.
+            still_busy = set()
+            for job in self._cache.values():
+                if not job._accepted:
+                    continue
+                proc = job._write_to or job._scheduled_for
+                if proc is not None and proc._is_alive():
+                    still_busy.add(proc.inqW_fd)
+            self._busy_workers.intersection_update(still_busy)
 
     def _flush_writer(self, proc, writer):
         fds = {proc.inq._writer}
diff --git a/celery/worker/consumer/consumer.py b/celery/worker/consumer/consumer.py
index 2a5d955fef3..cc41e44e101 100644
--- a/celery/worker/consumer/consumer.py
+++ b/celery/worker/consumer/consumer.py
@@ -459,10 +459,11 @@ def on_close(self):
         for bucket in self.task_buckets.values():
             if bucket:
                 bucket.clear_pending()
-        for request_id in reserved_requests:
-            if request_id in requests:
-                del requests[request_id]
+        for r in tuple(reserved_requests):
+            if r not in active_requests:
+                requests.pop(r.id, None)
         reserved_requests.clear()
+        reserved_requests.update(tuple(active_requests))
         if self.pool and self.pool.flush:
             self.pool.flush()
 
diff --git a/docker/Dockerfile b/docker/Dockerfile
index cebba9314ac..42cb5042c19 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -18,6 +18,7 @@ RUN apt-get update && apt-get install -y build-essential \
     lsb-release \
     git \
     libmemcached-dev \
+    libev-dev \
     make \
     liblzma-dev \
     libreadline-dev \
@@ -70,12 +71,11 @@ RUN pyenv install 3.13 && \
     pyenv install 3.12 && \
     pyenv install 3.11 && \
     pyenv install 3.10 && \
-    pyenv install 3.9 && \
     pyenv install pypy3.11
 
 
 # Set global Python versions
-RUN pyenv global 3.13 3.12 3.11 3.10 3.9 pypy3.11
+RUN pyenv global 3.13 3.12 3.11 3.10 pypy3.11
 
 # Install celery
 WORKDIR $HOME
@@ -84,14 +84,13 @@ COPY --chown=1000:1000 docker/entrypoint /entrypoint
 RUN chmod gu+x /entrypoint
 
 # Define the local pyenvs
-RUN pyenv local 3.13 3.12 3.11 3.10 3.9 pypy3.11
+RUN pyenv local 3.13 3.12 3.11 3.10 pypy3.11
 
 RUN --mount=type=cache,target=/home/$CELERY_USER/.cache/pip \
     pyenv exec python3.13 -m pip install --upgrade pip setuptools wheel && \
     pyenv exec python3.12 -m pip install --upgrade pip setuptools wheel && \
     pyenv exec python3.11 -m pip install --upgrade pip setuptools wheel && \
     pyenv exec python3.10 -m pip install --upgrade pip setuptools wheel && \
-    pyenv exec python3.9 -m pip install --upgrade pip setuptools wheel && \
     pyenv exec pypy3.11 -m pip install --upgrade pip setuptools wheel
 
 # Install requirements first to leverage Docker layer caching
@@ -140,17 +139,6 @@ RUN --mount=type=cache,target=/home/$CELERY_USER/.cache/pip \
     -r requirements/test.txt \
     --build-constraint requirements/constraints.txt
 
-RUN --mount=type=cache,target=/home/$CELERY_USER/.cache/pip \
-    pyenv exec python3.9 -m pip install -r requirements/default.txt \
-    -r requirements/dev.txt \
-    -r requirements/docs.txt \
-    -r requirements/pkgutils.txt \
-    -r requirements/test-ci-base.txt \
-    -r requirements/test-ci-default.txt \
-    -r requirements/test-integration.txt \
-    -r requirements/test.txt \
-    --build-constraint requirements/constraints.txt
-
 RUN --mount=type=cache,target=/home/$CELERY_USER/.cache/pip \
     pyenv exec pypy3.11 -m pip install -r requirements/default.txt \
     -r requirements/dev.txt \
@@ -170,7 +158,6 @@ RUN --mount=type=cache,target=/home/$CELERY_USER/.cache/pip \
     pyenv exec python3.12 -m pip install --no-deps -e $HOME/celery && \
     pyenv exec python3.11 -m pip install --no-deps -e $HOME/celery && \
     pyenv exec python3.10 -m pip install --no-deps -e $HOME/celery && \
-    pyenv exec python3.9 -m pip install --no-deps -e $HOME/celery && \
     pyenv exec pypy3.11 -m pip install --no-deps -e $HOME/celery
 
 WORKDIR $HOME/celery
diff --git a/t/unit/app/test_preload_cli.py b/t/unit/app/test_preload_cli.py
index cb07b7866cb..889e43e6822 100644
--- a/t/unit/app/test_preload_cli.py
+++ b/t/unit/app/test_preload_cli.py
@@ -38,7 +38,7 @@ def test_preload_options(subcommand_with_params: Tuple[str, ...], isolated_cli_r
         catch_exceptions=False,
     )
 
-    assert "No such option: --ini" in res_without_preload.output
+    assert "No such option" in res_without_preload.output
     assert res_without_preload.exit_code == 2
 
     res_with_preload = isolated_cli_runner.invoke(
diff --git a/t/unit/concurrency/test_prefork.py b/t/unit/concurrency/test_prefork.py
index 8f266f78302..7e362d5fcf9 100644
--- a/t/unit/concurrency/test_prefork.py
+++ b/t/unit/concurrency/test_prefork.py
@@ -735,6 +735,139 @@ def test_flush_not_started_write_job_is_discarded(self, _create_worker_process):
         job.discard.assert_called_once()
         assert gen not in pool._active_writers
 
+    @t.skip.if_pypy
+    @patch('billiard.pool.Pool._create_worker_process')
+    def test_flush_preserves_busy_worker_for_accepted_job(self, _create_worker_process):
+        pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
+        pool._state = asynpool.RUN
+        pool.maintain_pool = Mock(name='maintain_pool')
+
+        proc = Mock(name='proc')
+        proc.inqW_fd = 7
+        proc._is_alive.return_value = True
+        job = Mock(name='job')
+        job._accepted = True
+        job._write_to = proc
+        job._scheduled_for = proc
+        job._writer.return_value = None
+
+        pool._cache = {1: job}
+        pool._busy_workers = {7}
+        pool._active_writers.clear()
+        pool.outbound_buffer.clear()
+
+        pool.flush()
+
+        assert 7 in pool._busy_workers
+
+    @t.skip.if_pypy
+    @patch('billiard.pool.Pool._create_worker_process')
+    def test_flush_preserves_busy_worker_via_scheduled_for_fallback(self, _create_worker_process):
+        pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
+        pool._state = asynpool.RUN
+        pool.maintain_pool = Mock(name='maintain_pool')
+
+        proc = Mock(name='proc')
+        proc.inqW_fd = 5
+        proc._is_alive.return_value = True
+        job = Mock(name='job')
+        job._accepted = True
+        job._write_to = None
+        job._scheduled_for = proc
+        job._writer.return_value = None
+
+        pool._cache = {1: job}
+        pool._busy_workers = {5}
+        pool._active_writers.clear()
+        pool.outbound_buffer.clear()
+
+        pool.flush()
+
+        assert 5 in pool._busy_workers
+
+    @t.skip.if_pypy
+    @patch('billiard.pool.Pool._create_worker_process')
+    def test_flush_releases_busy_worker_for_unaccepted_job(self, _create_worker_process):
+        pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
+        pool._state = asynpool.RUN
+        pool.maintain_pool = Mock(name='maintain_pool')
+
+        proc = Mock(name='proc')
+        proc.inqW_fd = 9
+        job = Mock(name='job')
+        job._accepted = False
+        job._write_to = proc
+        job._scheduled_for = proc
+        job._writer.return_value = None
+
+        pool._cache = {1: job}
+        pool._busy_workers = {9}
+        pool._active_writers.clear()
+        pool.outbound_buffer.clear()
+
+        pool.flush()
+
+        assert 9 not in pool._busy_workers
+
+    @t.skip.if_pypy
+    @patch('billiard.pool.Pool._create_worker_process')
+    def test_flush_preserves_accepted_and_releases_unaccepted_together(self, _create_worker_process):
+        pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
+        pool._state = asynpool.RUN
+        pool.maintain_pool = Mock(name='maintain_pool')
+
+        accepted_proc = Mock(name='accepted_proc')
+        accepted_proc.inqW_fd = 7
+        accepted_proc._is_alive.return_value = True
+        accepted_job = Mock(name='accepted_job')
+        accepted_job._accepted = True
+        accepted_job._write_to = accepted_proc
+        accepted_job._scheduled_for = accepted_proc
+        accepted_job._writer.return_value = None
+
+        unaccepted_proc = Mock(name='unaccepted_proc')
+        unaccepted_proc.inqW_fd = 9
+        unaccepted_job = Mock(name='unaccepted_job')
+        unaccepted_job._accepted = False
+        unaccepted_job._write_to = unaccepted_proc
+        unaccepted_job._scheduled_for = unaccepted_proc
+        unaccepted_job._writer.return_value = None
+
+        pool._cache = {1: accepted_job, 2: unaccepted_job}
+        pool._busy_workers = {7, 9}
+        pool._active_writers.clear()
+        pool.outbound_buffer.clear()
+
+        pool.flush()
+
+        assert 7 in pool._busy_workers
+        assert 9 not in pool._busy_workers
+
+    @t.skip.if_pypy
+    @patch('billiard.pool.Pool._create_worker_process')
+    def test_flush_releases_busy_worker_for_dead_process(self, _create_worker_process):
+        pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
+        pool._state = asynpool.RUN
+        pool.maintain_pool = Mock(name='maintain_pool')
+
+        proc = Mock(name='proc')
+        proc.inqW_fd = 3
+        proc._is_alive.return_value = False
+        job = Mock(name='job')
+        job._accepted = True
+        job._write_to = proc
+        job._scheduled_for = proc
+        job._writer.return_value = None
+
+        pool._cache = {1: job}
+        pool._busy_workers = {3}
+        pool._active_writers.clear()
+        pool.outbound_buffer.clear()
+
+        pool.flush()
+
+        assert 3 not in pool._busy_workers
+
     def test_process_result(self):
         x = asynpool.ResultHandler(
             Mock(), Mock(), {}, Mock(),
diff --git a/t/unit/worker/test_consumer.py b/t/unit/worker/test_consumer.py
index 8cdb908795f..e28e2d9b610 100644
--- a/t/unit/worker/test_consumer.py
+++ b/t/unit/worker/test_consumer.py
@@ -389,6 +389,39 @@ def test_on_close_clears_semaphore_and_reqs(self):
             c.pool = None
             c.on_close()
 
+    def test_on_close_purges_orphan_reservations_from_requests_dict(self):
+        from celery.worker import state
+        from celery.worker.consumer.consumer import Consumer
+
+        class FakeRequest:
+            def __init__(self, id):
+                self.id = id
+
+        consumer = Mock()
+        consumer.controller = Mock()
+        consumer.controller.semaphore = Mock()
+        consumer.task_buckets = {}
+        consumer.pool = Mock()
+        consumer.pool.flush = Mock()
+
+        state.reset_state()
+        try:
+            orphan = FakeRequest('orphan-1')
+            state.requests[orphan.id] = orphan
+            state.reserved_requests.add(orphan)
+
+            Consumer.on_close(consumer)
+
+            assert orphan.id not in state.requests, (
+                "on_close() did not purge an orphan reserved-but-not-"
+                "accepted Request from state.requests; this is the leak "
+                "PR #7771 tried to fix but its loop variable ('request_id') "
+                "actually held Request objects, so the membership check "
+                "never matched."
+            )
+        finally:
+            state.reset_state()
+
     def test_connect_error_handler(self):
         self.app._connection = _amqp_connection()
         conn = self.app._connection.return_value
@@ -685,6 +718,68 @@ def test_disable_prefetch_respects_autoscale_max_concurrency(self):
             # Should not be able to consume when at autoscale limit
             assert consumer.task_consumer.channel.qos.can_consume() is False
 
+    def test_disable_prefetch_after_connection_loss_keeps_gate_closed(self):
+        from celery.worker import state
+        from celery.worker.consumer.consumer import Consumer
+        from celery.worker.consumer.tasks import Tasks
+
+        self.app.conf.worker_disable_prefetch = True
+
+        # Mock() isn't weakref-able; state's WeakSets would silently drop it.
+        class FakeRequest:
+            def __init__(self, id):
+                self.id = id
+
+        consumer = Mock()
+        consumer.app = self.app
+        consumer.pool = Mock()
+        consumer.pool.num_processes = 1
+        consumer.pool.flush = Mock()
+        consumer.controller = Mock()
+        consumer.controller.max_concurrency = None
+        consumer.controller.semaphore = Mock()
+        consumer.task_buckets = {}
+        consumer.initial_prefetch_count = 1
+        consumer.connection = Mock()
+        consumer.connection.connection_errors = ()
+        consumer.connection.channel_errors = ()
+        consumer.connection.default_channel = Mock()
+        consumer.connection.transport = Mock()
+        consumer.connection.transport.driver_type = 'redis'
+        consumer.update_strategies = Mock()
+        consumer.on_decode_error = Mock()
+
+        consumer.task_consumer = Mock()
+        consumer.task_consumer.channel = Mock()
+        consumer.task_consumer.channel.qos = Mock()
+        consumer.task_consumer.channel.qos.can_consume = Mock(return_value=True)
+        consumer.task_consumer.qos = Mock()
+        consumer.app.amqp = Mock()
+        consumer.app.amqp.TaskConsumer = Mock(return_value=consumer.task_consumer)
+
+        Tasks(consumer).start(consumer)
+        can_consume = consumer.task_consumer.channel.qos.can_consume
+
+        state.reset_state()
+        try:
+            in_flight = FakeRequest('task-1')
+            state.reserved_requests.add(in_flight)
+            state.active_requests.add(in_flight)
+
+            assert can_consume() is False
+
+            Consumer.on_close(consumer)
+
+            assert in_flight in state.active_requests
+            assert can_consume() is False, (
+                "After a connection loss, can_consume() returned True while "
+                "a task is still running in the pool. Consumer.on_close() "
+                "cleared reserved_requests, hiding the in-flight task from "
+                "the worker_disable_prefetch gate."
+            )
+        finally:
+            state.reset_state()
+
     def test_disable_prefetch_ignored_for_non_redis_brokers(self):
         """Test that disable_prefetch is ignored for non-Redis brokers."""
         self.app.conf.worker_disable_prefetch = True