diff --git a/tests/backends/test_base.py b/tests/backends/test_base.py
index 76c659d..b72deda 100644
--- a/tests/backends/test_base.py
+++ b/tests/backends/test_base.py
@@ -86,6 +86,16 @@ def test_telemetry__raise_not_implemented_error(self) -> None:
         with pytest.raises(NotImplementedError):
             BackendDouble(alias="default", params={}).telemetry()
 
+    def test_dequeue__raise_not_implemented_error(self) -> None:
+        """Raise NotImplementedError for backend dequeue API."""
+        with pytest.raises(NotImplementedError):
+            BackendDouble(alias="default", params={}).dequeue(task_result=None)
+
+    def test_purge__raise_not_implemented_error(self) -> None:
+        """Raise NotImplementedError for backend purge_queue API."""
+        with pytest.raises(NotImplementedError):
+            BackendDouble(alias="default", params={}).purge("default")
+
     def test_validate_task__accepts_module_level_function(self) -> None:
         """validate_task accepts a module-level retry callback."""
         backend = BackendDouble(alias="default", params={})
diff --git a/tests/backends/test_redis.py b/tests/backends/test_redis.py
index a4adff3..aba9206 100644
--- a/tests/backends/test_redis.py
+++ b/tests/backends/test_redis.py
@@ -89,12 +89,10 @@ def test_main__trims_stale_telemetry(self):
             ingress_key = backend.INGRESS_KEY.format(
                 prefix=backend.key_prefix, queue_name="default"
             )
-            successful_key = backend.SUCCESSFUL_RESULTS_KEY.format(
-                prefix=backend.key_prefix, queue_name="default"
-            )
-            failed_key = backend.FAILED_RESULTS_KEY.format(
-                prefix=backend.key_prefix, queue_name="default"
+            successful_key = backend._segment_key(
+                TaskResultStatus.SUCCESSFUL, "default"
             )
+            failed_key = backend._segment_key(TaskResultStatus.FAILED, "default")
             old = (timezone.now() - datetime.timedelta(seconds=120)).timestamp() * 1000
             backend.client.zadd(ingress_key, {task_result.id: old})
             backend.client.zadd(successful_key, {task_result.id: old})
@@ -138,9 +136,7 @@ def test_acquire__moves_to_running_set(self):
             assert acquired.id == task_result.id
 
             # Verify task is in running set, not in any processing set
-            running_key = backend.RUNNING_KEY.format(
-                prefix=backend.key_prefix, queue_name="default"
-            )
+            running_key = backend._segment_key(TaskResultStatus.RUNNING, "default")
             assert backend.client.zscore(running_key, task_result.id) is not None
 
             # Verify task data was updated with worker info
@@ -401,12 +397,10 @@ def test_telemetry__successful_failed_evicted_by_result_ttl(self):
             },
         )
         try:
-            successful_key = backend.SUCCESSFUL_RESULTS_KEY.format(
-                prefix=backend.key_prefix, queue_name="default"
-            )
-            failed_key = backend.FAILED_RESULTS_KEY.format(
-                prefix=backend.key_prefix, queue_name="default"
+            successful_key = backend._segment_key(
+                TaskResultStatus.SUCCESSFUL, "default"
             )
+            failed_key = backend._segment_key(TaskResultStatus.FAILED, "default")
 
             def _ack(status: TaskResultStatus) -> str:
                 enqueued = backend.enqueue(echo, args=[1])
@@ -472,8 +466,8 @@ def test_telemetry__ingress_egress_age_out_of_window(self):
             ingress_key = backend.INGRESS_KEY.format(
                 prefix=backend.key_prefix, queue_name="default"
             )
-            successful_results_key = backend.SUCCESSFUL_RESULTS_KEY.format(
-                prefix=backend.key_prefix, queue_name="default"
+            successful_results_key = backend._segment_key(
+                TaskResultStatus.SUCCESSFUL, "default"
             )
             old = (timezone.now() - datetime.timedelta(seconds=120)).timestamp() * 1000
             backend.client.zadd(ingress_key, {task_result.id: old})
@@ -556,8 +550,8 @@ def test_telemetry__egress_window_follows_display_interval(self):
                     finished_at=timezone.now(),
                 )
             )
-            successful_key = backend.SUCCESSFUL_RESULTS_KEY.format(
-                prefix=backend.key_prefix, queue_name="default"
+            successful_key = backend._segment_key(
+                TaskResultStatus.SUCCESSFUL, "default"
             )
 
             # Inside result_ttl (60s) but outside a 10s display window.
@@ -720,9 +714,7 @@ def test_requeue__moves_from_running_to_deferred(self) -> None:
             run_after = timezone.now() + datetime.timedelta(seconds=10)
             backend.requeue(failed, run_after)
 
-            running_key = backend.RUNNING_KEY.format(
-                prefix=backend.key_prefix, queue_name="default"
-            )
+            running_key = backend._segment_key(TaskResultStatus.RUNNING, "default")
             deferred_key = backend.DEFERRED_KEY.format(
                 prefix=backend.key_prefix, queue_name="default"
             )
@@ -832,3 +824,221 @@ def test_requeue__task_is_re_acquirable_after_delay(self) -> None:
             assert re_acquired.id == task_result.id
         finally:
             backend.close()
+
+    def test_requeue__cleans_up_failed_and_result_keys(self) -> None:
+        """requeue() removes the task from the failed zset and deletes its result key."""
+        backend = RedisTaskBackend(
+            "requeue_cleanup_test",
+            {
+                "QUEUES": ["default"],
+                "REDIS_URL": "redis://localhost:6379/0",
+                "OPTIONS": {
+                    "lease_ttl": datetime.timedelta(hours=1),
+                    "result_ttl": datetime.timedelta(seconds=60),
+                },
+            },
+        )
+        try:
+            backend.enqueue(echo, args=[1])
+            acquired = backend.acquire(
+                timeout=datetime.timedelta(seconds=1), worker="cleanup-test"
+            )
+            assert acquired is not None
+            backend.acknowledge(
+                dataclasses.replace(
+                    acquired,
+                    status=TaskResultStatus.FAILED,
+                    finished_at=timezone.now(),
+                )
+            )
+            failed = next(
+                backend.peek(
+                    queue_name="default", status=TaskResultStatus.FAILED, count=10
+                )
+            )
+            failed_key = backend._segment_key(TaskResultStatus.FAILED, "default")
+            result_key = backend.RESULT_KEY.format(
+                prefix=backend.key_prefix, result_id=failed.id
+            )
+            assert backend.client.zscore(failed_key, failed.id) is not None
+            assert backend.client.exists(result_key)
+
+            backend.requeue(failed, timezone.now() + datetime.timedelta(seconds=10))
+
+            assert backend.client.zscore(failed_key, failed.id) is None
+            assert not backend.client.exists(result_key)
+        finally:
+            backend.close()
+
+    def test_dequeue__removes_ready_task_from_queue(self) -> None:
+        """dequeue() removes a ready task from the queue zset."""
+        backend = RedisTaskBackend(
+            "dequeue_ready_test",
+            {
+                "QUEUES": ["default"],
+                "REDIS_URL": "redis://localhost:6379/0",
+                "OPTIONS": {
+                    "lease_ttl": datetime.timedelta(hours=1),
+                    "result_ttl": datetime.timedelta(seconds=60),
+                },
+            },
+        )
+        try:
+            task_result = backend.enqueue(echo, args=[1])
+            queue_key = backend._segment_key(TaskResultStatus.READY, "default")
+            assert backend.client.zscore(queue_key, task_result.id) is not None
+
+            backend.dequeue(task_result)
+
+            assert backend.client.zscore(queue_key, task_result.id) is None
+        finally:
+            backend.close()
+
+    def test_dequeue__removes_failed_task_from_results(self) -> None:
+        """dequeue() removes a failed task from the failed zset."""
+        backend = RedisTaskBackend(
+            "dequeue_failed_test",
+            {
+                "QUEUES": ["default"],
+                "REDIS_URL": "redis://localhost:6379/0",
+                "OPTIONS": {
+                    "lease_ttl": datetime.timedelta(hours=1),
+                    "result_ttl": datetime.timedelta(seconds=60),
+                },
+            },
+        )
+        try:
+            backend.enqueue(echo, args=[1])
+            acquired = backend.acquire(
+                timeout=datetime.timedelta(seconds=1), worker="dequeue-failed-test"
+            )
+            assert acquired is not None
+            backend.acknowledge(
+                dataclasses.replace(
+                    acquired,
+                    status=TaskResultStatus.FAILED,
+                    finished_at=timezone.now(),
+                )
+            )
+            failed = next(
+                backend.peek(
+                    queue_name="default", status=TaskResultStatus.FAILED, count=10
+                )
+            )
+            failed_key = backend._segment_key(TaskResultStatus.FAILED, "default")
+
+            backend.dequeue(failed)
+
+            assert backend.client.zscore(failed_key, failed.id) is None
+        finally:
+            backend.close()
+
+    def test_purge__removes_all_tasks_across_segments(self) -> None:
+        """purge_queue() deletes every task across all segments."""
+        backend = RedisTaskBackend(
+            "purge_test",
+            {
+                "QUEUES": ["default"],
+                "REDIS_URL": "redis://localhost:6379/0",
+                "OPTIONS": {
+                    "lease_ttl": datetime.timedelta(hours=1),
+                    "result_ttl": datetime.timedelta(seconds=60),
+                },
+            },
+        )
+        try:
+            # Two ready tasks
+            backend.enqueue(echo, args=[1])
+            backend.enqueue(echo, args=[2])
+            # One running task
+            backend.acquire(timeout=datetime.timedelta(seconds=1), worker="purge-test")
+            # One failed task
+            backend.enqueue(echo, args=[3])
+            acquired = backend.acquire(
+                timeout=datetime.timedelta(seconds=1), worker="purge-test-2"
+            )
+            assert acquired is not None
+            backend.acknowledge(
+                dataclasses.replace(
+                    acquired,
+                    status=TaskResultStatus.FAILED,
+                    finished_at=timezone.now(),
+                )
+            )
+            # One successful task
+            backend.enqueue(echo, args=[4])
+            acquired = backend.acquire(
+                timeout=datetime.timedelta(seconds=1), worker="purge-test-3"
+            )
+            assert acquired is not None
+            backend.acknowledge(
+                dataclasses.replace(
+                    acquired,
+                    status=TaskResultStatus.SUCCESSFUL,
+                    finished_at=timezone.now(),
+                )
+            )
+
+            backend.purge("default")
+            assert (
+                list(
+                    backend.peek(
+                        queue_name="default", status=TaskResultStatus.READY, count=10
+                    )
+                )
+                == []
+            )
+            assert (
+                list(
+                    backend.peek(
+                        queue_name="default", status=TaskResultStatus.RUNNING, count=10
+                    )
+                )
+                == []
+            )
+            assert (
+                list(
+                    backend.peek(
+                        queue_name="default", status=TaskResultStatus.FAILED, count=10
+                    )
+                )
+                == []
+            )
+            assert (
+                list(
+                    backend.peek(
+                        queue_name="default",
+                        status=TaskResultStatus.SUCCESSFUL,
+                        count=10,
+                    )
+                )
+                == []
+            )
+        finally:
+            backend.close()
+
+    def test_purge__empty_queue_is_noop(self) -> None:
+        """purge_queue() on an empty queue is a no-op."""
+        backend = RedisTaskBackend(
+            "purge_empty_test",
+            {
+                "QUEUES": ["default"],
+                "REDIS_URL": "redis://localhost:6379/0",
+                "OPTIONS": {
+                    "lease_ttl": datetime.timedelta(hours=1),
+                    "result_ttl": datetime.timedelta(seconds=60),
+                },
+            },
+        )
+        try:
+            backend.purge("default")
+            assert (
+                list(
+                    backend.peek(
+                        queue_name="default", status=TaskResultStatus.READY, count=10
+                    )
+                )
+                == []
+            )
+        finally:
+            backend.close()
diff --git a/tests/test_inspector.py b/tests/test_inspector.py
index 6da6ad2..0b006d9 100644
--- a/tests/test_inspector.py
+++ b/tests/test_inspector.py
@@ -5,6 +5,7 @@
 
 pytest.importorskip("textual.widgets")
 
+from django.conf import settings
 from django.tasks import (
     DEFAULT_TASK_QUEUE_NAME,
     TaskResult,
@@ -24,6 +25,7 @@
     QueueStats,
     ThreadmillTaskBackend,
 )
+from threadmill.backends.redis import RedisTaskBackend
 from threadmill.inspector.app import (
     InspectorApp,
     QueueList,
@@ -31,6 +33,7 @@
     TaskList,
     si_prefix,
 )
+from threadmill.inspector.screens import ConfirmScreen, PurgeScreen
 
 
 class FailingBackend(ThreadmillTaskBackend):
@@ -46,6 +49,19 @@ def telemetry(self, *, interval=None):
         raise RuntimeError("telemetry failed")
 
 
+class ErroringBackend(RedisTaskBackend):
+    """Redis backend whose requeue/dequeue/purge raise for error-path tests."""
+
+    def requeue(self, task_result, run_after):
+        raise RuntimeError("boom")
+
+    def dequeue(self, task_result):
+        raise RuntimeError("boom")
+
+    def purge(self, queue_name):
+        raise RuntimeError("boom")
+
+
 def _failed_result() -> TaskResult:
     """Build a failed TaskResult with an error for detail-view tests."""
     return TaskResult(
@@ -347,7 +363,7 @@ async def test_action_refresh_polls_backend(self):
         async with app.run_test() as pilot:
             await pilot.pause()
             default_task_backend.enqueue(echo, args=[1])
-            app.action_refresh()
+            app.query_one("#task-list", TaskList).action_refresh()
             await pilot.pause()
             assert app.telemetry.queues["default"].counts.ready == 1
 
@@ -396,7 +412,7 @@ async def test_tab_columns_match_status(self):
             ]
 
     async def test_digit_binding_switches_tab(self):
-        """Pressing a digit key activates the matching status tab and refreshes it."""
+        """The digit binding activates the matching status tab and refreshes it."""
         default_task_backend.enqueue(echo, args=[1])
         default_task_backend.acquire(
             timeout=datetime.timedelta(seconds=1), worker="tab-test"
@@ -406,7 +422,7 @@ async def test_digit_binding_switches_tab(self):
             await pilot.pause()
             await pilot.pause()
             task_list = app.query_one("#task-list", TaskList)
-            await pilot.press("1")
+            app.action_switch_tab("tab-running")
             await pilot.pause()
             await pilot.pause()
             assert task_list._tabs.active == "tab-running"
@@ -505,7 +521,446 @@ async def test_action_refresh_refreshes_task_list(self):
             default_task_backend.acquire(
                 timeout=datetime.timedelta(seconds=1), worker="f5-test"
             )
-            app.action_refresh()
+            app.query_one("#task-list", TaskList).action_refresh()
             await pilot.pause()
             await pilot.pause()
             assert table.row_count == before - 1
+
+
+def _acknowledge_failed() -> str:
+    """Enqueue, acquire, and acknowledge a task as FAILED. Return its ID."""
+    task_result = default_task_backend.enqueue(echo, args=[1])
+    acquired = default_task_backend.acquire(
+        timeout=datetime.timedelta(seconds=1), worker="inspector-test"
+    )
+    assert acquired is not None
+    default_task_backend.acknowledge(
+        dataclasses.replace(
+            acquired, status=TaskResultStatus.FAILED, finished_at=timezone.now()
+        )
+    )
+    return task_result.id
+
+
+def _acknowledge_successful() -> str:
+    """Enqueue, acquire, and acknowledge a task as SUCCESSFUL. Return its ID."""
+    task_result = default_task_backend.enqueue(echo, args=[1])
+    acquired = default_task_backend.acquire(
+        timeout=datetime.timedelta(seconds=1), worker="inspector-test"
+    )
+    assert acquired is not None
+    default_task_backend.acknowledge(
+        dataclasses.replace(
+            acquired, status=TaskResultStatus.SUCCESSFUL, finished_at=timezone.now()
+        )
+    )
+    return task_result.id
+
+
+async def _select_failed_task(pilot, app) -> tuple[TaskList, TaskResult]:
+    """Set up the task list with a failed task selected on the Failed tab."""
+    task_id = _acknowledge_failed()
+    task_list = app.query_one("#task-list", TaskList)
+    task_list.queue_name = "default"
+    await pilot.pause()
+    task_list.switch_tab("tab-failed")
+    await pilot.pause()
+    await pilot.pause()
+    failed_task = next(
+        (r for r in task_list._current_results if r.id == task_id),
+        None,
+    )
+    assert failed_task is not None, "Failed task not found in task list"
+    task_list.selected_task = failed_task
+    return task_list, failed_task
+
+
+class TestCheckAction:
+    """Tests for TaskList.check_action conditional binding visibility."""
+
+    async def test_check_action__requeue_visible_only_on_failed_tab(self):
+        """check_action shows requeue only when the Failed tab is active."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            task_list.switch_tab("tab-failed")
+            await pilot.pause()
+            assert task_list.check_action("requeue", ()) is True
+            task_list.switch_tab("tab-ready")
+            await pilot.pause()
+            assert task_list.check_action("requeue", ()) is False
+
+    async def test_check_action__dequeue_hidden_on_running_tab(self):
+        """check_action hides dequeue on the Running tab."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            task_list.switch_tab("tab-running")
+            await pilot.pause()
+            assert task_list.check_action("dequeue", ()) is False
+
+    async def test_check_action__dequeue_visible_on_non_running_tabs(self):
+        """check_action shows dequeue on Ready, Successful, and Failed tabs."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            for tab in ("tab-ready", "tab-successful", "tab-failed"):
+                task_list.switch_tab(tab)
+                await pilot.pause()
+                assert task_list.check_action("dequeue", ()) is True, tab
+
+    async def test_check_action__other_actions_always_visible(self):
+        """check_action always shows refresh binding."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            assert task_list.check_action("refresh", ()) is True
+
+
+class TestRequeueAction:
+    """Tests for TaskList.action_requeue and _do_requeue."""
+
+    async def test_action_requeue__noop_without_selected_task(self):
+        """Requeue action does nothing when no task is selected."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            task_list.switch_tab("tab-failed")
+            await pilot.pause()
+            task_list.action_requeue()
+            await pilot.pause()
+            assert not isinstance(app.screen, ConfirmScreen)
+
+    async def test_action_requeue__noop_on_non_failed_tab(self):
+        """Requeue action does nothing when not on the Failed tab."""
+        _acknowledge_failed()
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            task_list.queue_name = "default"
+            await pilot.pause()
+            task_list.switch_tab("tab-ready")
+            await pilot.pause()
+            task_list.selected_task = next(
+                (r for r in task_list._current_results),
+                None,
+            )
+            task_list.action_requeue()
+            await pilot.pause()
+            assert not isinstance(app.screen, ConfirmScreen)
+
+    async def test_action_requeue__pushes_confirm_screen(self):
+        """Requeue action pushes a confirmation screen for a failed task."""
+        task_id = _acknowledge_failed()
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            task_list.queue_name = "default"
+            await pilot.pause()
+            task_list.switch_tab("tab-failed")
+            await pilot.pause()
+            await pilot.pause()
+            task_list.selected_task = next(
+                (r for r in task_list._current_results if r.id == task_id),
+                None,
+            )
+            task_list.action_requeue()
+            await pilot.pause()
+            assert isinstance(app.screen, ConfirmScreen)
+
+    async def test_do_requeue__calls_backend_requeue(self):
+        """_do_requeue requeues the failed task after confirmation."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list, failed_task = await _select_failed_task(pilot, app)
+            task_list._do_requeue(True, failed_task)
+            await pilot.pause()
+            failed = list(
+                default_task_backend.peek("default", status=TaskResultStatus.FAILED)
+            )
+            assert failed_task.id not in {r.id for r in failed}
+
+    async def test_do_requeue__noop_when_not_confirmed(self):
+        """_do_requeue leaves the failed task in place when not confirmed."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list, failed_task = await _select_failed_task(pilot, app)
+            task_list._do_requeue(False, failed_task)
+            await pilot.pause()
+            failed = list(
+                default_task_backend.peek("default", status=TaskResultStatus.FAILED)
+            )
+            assert failed_task.id in {r.id for r in failed}
+
+    async def test_do_requeue__notifies_on_error(self):
+        """_do_requeue logs and notifies when backend.requeue raises."""
+        params = {k: v for k, v in settings.TASKS["default"].items() if k != "BACKEND"}
+        erroring = ErroringBackend(alias="default", params=params)
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app.backend = erroring
+            await pilot.pause()
+            task_list, failed_task = await _select_failed_task(pilot, app)
+            task_list._do_requeue(True, failed_task)
+            await pilot.pause()
+            failed = list(
+                default_task_backend.peek("default", status=TaskResultStatus.FAILED)
+            )
+            assert failed_task.id in {r.id for r in failed}
+
+
+class TestDequeueAction:
+    """Tests for TaskList.action_dequeue and _do_dequeue."""
+
+    async def test_action_dequeue__noop_without_selected_task(self):
+        """Dequeue action does nothing when no task is selected."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            task_list.switch_tab("tab-ready")
+            await pilot.pause()
+            task_list.action_dequeue()
+            await pilot.pause()
+            assert not isinstance(app.screen, ConfirmScreen)
+
+    async def test_action_dequeue__pushes_confirm_screen(self):
+        """Dequeue action pushes a confirmation screen for a selected task."""
+        task_result = default_task_backend.enqueue(echo, args=[1])
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            task_list.queue_name = "default"
+            await pilot.pause()
+            await pilot.pause()
+            task_list.selected_task = next(
+                (r for r in task_list._current_results if r.id == task_result.id),
+                None,
+            )
+            task_list.action_dequeue()
+            await pilot.pause()
+            assert isinstance(app.screen, ConfirmScreen)
+
+    async def test_do_dequeue__calls_backend_dequeue(self):
+        """_do_dequeue removes the task from the ready queue after confirmation."""
+        task_result = default_task_backend.enqueue(echo, args=[1])
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            task_list.queue_name = "default"
+            await pilot.pause()
+            await pilot.pause()
+            ready_task = next(
+                (r for r in task_list._current_results if r.id == task_result.id),
+                None,
+            )
+            task_list._do_dequeue(True, ready_task)
+            await pilot.pause()
+            ready = list(
+                default_task_backend.peek("default", status=TaskResultStatus.READY)
+            )
+            assert task_result.id not in {r.id for r in ready}
+
+    async def test_do_dequeue__noop_when_not_confirmed(self):
+        """_do_dequeue leaves the task in the ready queue when not confirmed."""
+        task_result = default_task_backend.enqueue(echo, args=[1])
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            task_list.queue_name = "default"
+            await pilot.pause()
+            await pilot.pause()
+            ready_task = next(
+                (r for r in task_list._current_results if r.id == task_result.id),
+                None,
+            )
+            task_list._do_dequeue(False, ready_task)
+            await pilot.pause()
+            ready = list(
+                default_task_backend.peek("default", status=TaskResultStatus.READY)
+            )
+            assert task_result.id in {r.id for r in ready}
+
+    async def test_do_dequeue__notifies_on_error(self):
+        """_do_dequeue logs and notifies when backend.dequeue raises."""
+        task_result = default_task_backend.enqueue(echo, args=[1])
+        params = {k: v for k, v in settings.TASKS["default"].items() if k != "BACKEND"}
+        erroring = ErroringBackend(alias="default", params=params)
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app.backend = erroring
+            await pilot.pause()
+            task_list = app.query_one("#task-list", TaskList)
+            task_list.queue_name = "default"
+            await pilot.pause()
+            await pilot.pause()
+            ready_task = next(
+                (r for r in task_list._current_results if r.id == task_result.id),
+                None,
+            )
+            task_list._do_dequeue(True, ready_task)
+            await pilot.pause()
+            ready = list(
+                default_task_backend.peek("default", status=TaskResultStatus.READY)
+            )
+            assert task_result.id in {r.id for r in ready}
+
+
+class TestPurgeAction:
+    """Tests for QueueList.action_purge_queue and _do_purge."""
+
+    async def test_action_purge_queue__noop_without_highlighted_item(self):
+        """Purge action does nothing when no queue item is highlighted."""
+        app = InspectorApp(backend=default_task_backend, auto_refresh=False)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            queue_list = app.query_one("#queue-list", QueueList)
+            queue_list.clear()
+            queue_list._items.clear()
+            await pilot.pause()
+            assert queue_list.highlighted_child is None
+            queue_list.action_purge()
+            await pilot.pause()
+            assert not isinstance(app.screen, PurgeScreen)
+
+    async def test_action_purge_queue__pushes_purge_screen(self):
+        """Purge action pushes a PurgeScreen for the highlighted queue."""
+        default_task_backend.enqueue(echo, args=[1])
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            await pilot.pause()
+            queue_list = app.query_one("#queue-list", QueueList)
+            queue_list.action_purge()
+            await pilot.pause()
+            assert isinstance(app.screen, PurgeScreen)
+
+    async def test_do_purge__calls_backend_purge_queue(self):
+        """_do_purge empties the queue after confirmation."""
+        default_task_backend.enqueue(echo, args=[1])
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            await pilot.pause()
+            app.query_one("#queue-list", QueueList)._do_purge(True, "default")
+            await pilot.pause()
+            ready = list(
+                default_task_backend.peek("default", status=TaskResultStatus.READY)
+            )
+            assert ready == []
+
+    async def test_do_purge__noop_when_not_confirmed(self):
+        """_do_purge leaves tasks in the queue when not confirmed."""
+        default_task_backend.enqueue(echo, args=[1])
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            await pilot.pause()
+            app.query_one("#queue-list", QueueList)._do_purge(False, "default")
+            await pilot.pause()
+            ready = list(
+                default_task_backend.peek("default", status=TaskResultStatus.READY)
+            )
+            assert len(ready) == 1
+
+    async def test_do_purge__notifies_on_error(self):
+        """_do_purge logs and notifies when backend.purge_queue raises."""
+        default_task_backend.enqueue(echo, args=[1])
+        params = {k: v for k, v in settings.TASKS["default"].items() if k != "BACKEND"}
+        erroring = ErroringBackend(alias="default", params=params)
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app.backend = erroring
+            await pilot.pause()
+            await pilot.pause()
+            app.query_one("#queue-list", QueueList)._do_purge(True, "default")
+            await pilot.pause()
+            ready = list(
+                default_task_backend.peek("default", status=TaskResultStatus.READY)
+            )
+            assert len(ready) == 1
+
+
+class TestConfirmScreen:
+    """Tests for the ConfirmScreen modal dialog."""
+
+    async def test_confirm_screen__enter_confirms(self):
+        """Pressing Enter confirms and dismisses with True."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            result: list[bool] = []
+            app.push_screen(ConfirmScreen("Confirm?"), lambda r: result.append(r))
+            await pilot.pause()
+            await pilot.press("enter")
+            await pilot.pause()
+            assert result == [True]
+
+    async def test_confirm_screen__escape_cancels(self):
+        """Pressing Escape cancels and dismisses with False."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            result: list[bool] = []
+            app.push_screen(ConfirmScreen("Confirm?"), lambda r: result.append(r))
+            await pilot.pause()
+            await pilot.press("escape")
+            await pilot.pause()
+            assert result == [False]
+
+
+class TestPurgeScreen:
+    """Tests for the PurgeScreen modal dialog."""
+
+    async def test_purge_screen__correct_name_confirms(self):
+        """Typing the correct queue name and pressing Enter confirms."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            result: list[bool] = []
+            app.push_screen(PurgeScreen("default"), lambda r: result.append(r))
+            await pilot.pause()
+            await pilot.press("d", "e", "f", "a", "u", "l", "t")
+            await pilot.press("enter")
+            await pilot.pause()
+            assert result == [True]
+
+    async def test_purge_screen__wrong_name_does_not_confirm(self):
+        """Typing a wrong queue name does not dismiss the screen."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            result: list[bool] = []
+            app.push_screen(PurgeScreen("default"), lambda r: result.append(r))
+            await pilot.pause()
+            await pilot.press("x", "y", "z")
+            await pilot.press("enter")
+            await pilot.pause()
+            assert result == []
+
+    async def test_purge_screen__escape_cancels(self):
+        """Pressing Escape cancels and dismisses with False."""
+        app = InspectorApp(backend=default_task_backend)
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            result: list[bool] = []
+            app.push_screen(PurgeScreen("default"), lambda r: result.append(r))
+            await pilot.pause()
+            await pilot.press("escape")
+            await pilot.pause()
+            assert result == [False]
diff --git a/threadmill/backends/base.py b/threadmill/backends/base.py
index a29f246..313323a 100644
--- a/threadmill/backends/base.py
+++ b/threadmill/backends/base.py
@@ -217,7 +217,12 @@ def acknowledge(self, task_result: TaskResult) -> None:
         raise NotImplementedError
 
     def requeue(self, task_result: TaskResult, run_after: datetime.datetime) -> None:
-        """Re-queue a failed task result for a retry attempt after `run_after`."""
+        """Re-queue a failed task result for a retry attempt after `run_after`.
+
+        Cleans up any persisted failed result so the method works both for
+        in-flight retries (task still running) and inspector-driven requeues
+        of already-failed tasks.
+        """
         raise NotImplementedError
 
     def peek(
@@ -246,3 +251,15 @@ def telemetry(
             interval: The time window for rolling rates.
         """
         raise NotImplementedError
+
+    def dequeue(self, task_result: TaskResult) -> None:
+        """Delete a single task from its current status segment."""
+        raise NotImplementedError
+
+    def purge(self, queue_name: str) -> None:
+        """Delete every task across all segments of a queue.
+
+        Args:
+            queue_name: The queue to purge.
+        """
+        raise NotImplementedError
diff --git a/threadmill/backends/lua/acquire.lua b/threadmill/backends/lua/acquire.lua
index 5b8523f..c2ccf6a 100644
--- a/threadmill/backends/lua/acquire.lua
+++ b/threadmill/backends/lua/acquire.lua
@@ -7,7 +7,7 @@
 --                KEYS[4] = queue, etc.
 -- ARGV[1]     -- current time in milliseconds
 -- ARGV[2]     -- current time as ISO-8601 string
--- ARGV[3]     -- task key prefix (e.g. "threadmill:default:task:")
+-- ARGV[3]     -- task key prefix (e.g. "threadmill:task:")
 -- ARGV[4]     -- number of queue pairs (N/2)
 -- ARGV[5]     -- worker name
 -- ARGV[6]     -- lease TTL in milliseconds
diff --git a/threadmill/backends/redis.py b/threadmill/backends/redis.py
index 48e9311..8029e30 100644
--- a/threadmill/backends/redis.py
+++ b/threadmill/backends/redis.py
@@ -55,14 +55,12 @@ def _move_queue(self, queue_name: str) -> None:
         deferred_key = self.backend.DEFERRED_KEY.format(
             prefix=self.backend.key_prefix, queue_name=queue_name
         )
-        queue_key = self.backend.QUEUE_KEY.format(
-            prefix=self.backend.key_prefix, queue_name=queue_name
-        )
+        queue_key = self.backend._segment_key(TaskResultStatus.READY, queue_name)
         self._mover_script(
             keys=[deferred_key, queue_key],
             args=[
                 str(time.time() * 1000),
-                self.backend.key_prefix + ":task:",
+                f"{self.backend.key_prefix}:task:",
                 str(self.backend.batch_size),
             ],
         )
@@ -72,18 +70,16 @@ def _reap_running_queue(self, queue_name: str) -> None:
         now = timezone.now()
         now_ms = now.timestamp() * 1000
         finished_at_iso = now.isoformat()
-        running_key = self.backend.RUNNING_KEY.format(
-            prefix=self.backend.key_prefix, queue_name=queue_name
-        )
-        failed_results_key = self.backend.FAILED_RESULTS_KEY.format(
-            prefix=self.backend.key_prefix, queue_name=queue_name
+        running_key = self.backend._segment_key(TaskResultStatus.RUNNING, queue_name)
+        failed_results_key = self.backend._segment_key(
+            TaskResultStatus.FAILED, queue_name
         )
         self._reaper_script(
             keys=[running_key, failed_results_key],
             args=[
                 str(now_ms),
-                self.backend.key_prefix + ":task:",
-                self.backend.key_prefix + ":result:",
+                f"{self.backend.key_prefix}:task:",
+                f"{self.backend.key_prefix}:result:",
                 str(self.backend.batch_size),
                 str(int(self.backend.result_ttl.total_seconds())),
                 finished_at_iso,
@@ -125,20 +121,25 @@ class RedisTaskBackend(ThreadmillTaskBackend):
 
     broker_class = RedisBroker
 
-    QUEUE_KEY = "{prefix}:queue:{queue_name}"
-    RUNNING_KEY = "{prefix}:running:{queue_name}"
-    DEFERRED_KEY = "{prefix}:deferred:{queue_name}"
     TASK_KEY = "{prefix}:task:{task_id}"
     RESULT_KEY = "{prefix}:result:{result_id}"
-    SUCCESSFUL_RESULTS_KEY = "{prefix}:results:{queue_name}:successful"
-    FAILED_RESULTS_KEY = "{prefix}:results:{queue_name}:failed"
-    INGRESS_KEY = "{prefix}:ingress:{queue_name}:events"
+    SEGMENT_KEY = "{prefix}:{queue_name}:{status}"
+    DEFERRED_KEY = "{prefix}:{queue_name}:deferred"
+
+    INGRESS_KEY = "{prefix}:{queue_name}:ingress:events"
 
     ACQUIRE_SCRIPT = _load_lua("acquire")
     """Pop the next task from a priority queue and move it directly to the running set."""
     ACKNOWLEDGE_SCRIPT = _load_lua("acknowledge")
     """Remove from running, persist the result, and clean up."""
 
+    def _segment_key(self, status: TaskResultStatus, queue_name: str) -> str:
+        return self.SEGMENT_KEY.format(
+            prefix=self.key_prefix,
+            queue_name=queue_name,
+            status=status.value.lower(),
+        )
+
     def __init__(self, alias: str, params: dict) -> None:
         super().__init__(alias=alias, params=params)
 
@@ -224,9 +225,7 @@ def enqueue(
             run_after_ms = task.run_after.timestamp() * 1000
             pipe.zadd(deferred_key, {task_result.id: run_after_ms})
         else:
-            queue_key = self.QUEUE_KEY.format(
-                prefix=self.key_prefix, queue_name=task.queue_name
-            )
+            queue_key = self._segment_key(TaskResultStatus.READY, task.queue_name)
             pipe.zadd(queue_key, {task_result.id: score})
 
         pipe.execute()
@@ -246,8 +245,8 @@ def acquire(
             key
             for queue_name in queue_names
             for key in (
-                self.RUNNING_KEY.format(prefix=self.key_prefix, queue_name=queue_name),
-                self.QUEUE_KEY.format(prefix=self.key_prefix, queue_name=queue_name),
+                self._segment_key(TaskResultStatus.RUNNING, queue_name),
+                self._segment_key(TaskResultStatus.READY, queue_name),
             )
         ]
 
@@ -261,7 +260,7 @@ def acquire(
                 args=[
                     str(now_ms),
                     now_iso,
-                    self.key_prefix + ":task:",
+                    f"{self.key_prefix}:task:",
                     str(len(queue_names)),
                     worker,
                     str(int(self.lease_ttl.total_seconds() * 1000)),
@@ -281,18 +280,18 @@ def acquire(
 
     def acknowledge(self, task_result: TaskResult) -> None:
         serialized = self.serialize_task_result(task_result)
-        running_key = self.RUNNING_KEY.format(
-            prefix=self.key_prefix, queue_name=task_result.task.queue_name
+        running_key = self._segment_key(
+            TaskResultStatus.RUNNING, task_result.task.queue_name
         )
         result_key = self.RESULT_KEY.format(
             prefix=self.key_prefix, result_id=task_result.id
         )
         task_key = self.TASK_KEY.format(prefix=self.key_prefix, task_id=task_result.id)
-        successful_results_key = self.SUCCESSFUL_RESULTS_KEY.format(
-            prefix=self.key_prefix, queue_name=task_result.task.queue_name
+        successful_results_key = self._segment_key(
+            TaskResultStatus.SUCCESSFUL, task_result.task.queue_name
         )
-        failed_results_key = self.FAILED_RESULTS_KEY.format(
-            prefix=self.key_prefix, queue_name=task_result.task.queue_name
+        failed_results_key = self._segment_key(
+            TaskResultStatus.FAILED, task_result.task.queue_name
         )
         finished_at = task_result.finished_at or timezone.now()
         finish_score = finished_at.timestamp() * 1000
@@ -322,13 +321,19 @@ def requeue(self, task_result: TaskResult, run_after: datetime.datetime) -> None
             finished_at=None,
         )
         serialized = self.serialize_task_result(task_result)
-        running_key = self.RUNNING_KEY.format(
-            prefix=self.key_prefix, queue_name=task_result.task.queue_name
+        running_key = self._segment_key(
+            TaskResultStatus.RUNNING, task_result.task.queue_name
         )
         deferred_key = self.DEFERRED_KEY.format(
             prefix=self.key_prefix, queue_name=task_result.task.queue_name
         )
+        failed_key = self._segment_key(
+            TaskResultStatus.FAILED, task_result.task.queue_name
+        )
         task_key = self.TASK_KEY.format(prefix=self.key_prefix, task_id=task_result.id)
+        result_key = self.RESULT_KEY.format(
+            prefix=self.key_prefix, result_id=task_result.id
+        )
         score = self._compute_score(task_result.task.priority, task_result.enqueued_at)
         run_after_ms = run_after.timestamp() * 1000
         task_data_ttl = int(
@@ -337,11 +342,26 @@ def requeue(self, task_result: TaskResult, run_after: datetime.datetime) -> None
 
         pipe = self.client.pipeline()
         pipe.zrem(running_key, task_result.id)
+        pipe.zrem(failed_key, task_result.id)
+        pipe.delete(result_key)
         pipe.hset(task_key, mapping={"data": serialized, "score": str(score)})
         pipe.expire(task_key, task_data_ttl)
         pipe.zadd(deferred_key, {task_result.id: run_after_ms})
         pipe.execute()
 
+    def dequeue(self, task_result: TaskResult) -> None:
+        self.client.zrem(
+            self._segment_key(task_result.status, task_result.task.queue_name),
+            task_result.id,
+        )
+
+    def purge(self, queue_name: str) -> None:
+        pattern = f"{self.key_prefix}:{queue_name}:*"
+        pipe = self.client.pipeline()
+        for key in self.client.scan_iter(match=pattern):
+            pipe.delete(key)
+        pipe.execute()
+
     def peek(
         self,
         queue_name: str = DEFAULT_TASK_QUEUE_NAME,
@@ -350,34 +370,27 @@ def peek(
         count: int = 1,
     ) -> Generator[TaskResult]:
         match status:
-            case TaskResultStatus.READY:
+            case TaskResultStatus.READY | TaskResultStatus.RUNNING:
                 yield from self._peek(
-                    self.QUEUE_KEY, self.TASK_KEY, queue_name, count, "data"
+                    self._segment_key(status, queue_name),
+                    self.TASK_KEY,
+                    count,
+                    "data",
                 )
-            case TaskResultStatus.RUNNING:
+            case TaskResultStatus.SUCCESSFUL | TaskResultStatus.FAILED:
                 yield from self._peek(
-                    self.RUNNING_KEY, self.TASK_KEY, queue_name, count, "data"
-                )
-            case TaskResultStatus.SUCCESSFUL:
-                yield from self._peek(
-                    self.SUCCESSFUL_RESULTS_KEY, self.RESULT_KEY, queue_name, count
-                )
-            case TaskResultStatus.FAILED:
-                yield from self._peek(
-                    self.FAILED_RESULTS_KEY, self.RESULT_KEY, queue_name, count
+                    self._segment_key(status, queue_name),
+                    self.RESULT_KEY,
+                    count,
                 )
 
     def _peek(
         self,
-        zset_key_template: str,
+        zset_key: str,
         data_key_template: str,
-        queue_name: str,
         count: int,
         field: str | None = None,
     ) -> Generator[TaskResult]:
-        zset_key = zset_key_template.format(
-            prefix=self.key_prefix, queue_name=queue_name
-        )
         pipe = self.client.pipeline()
         for member in self.client.zrange(zset_key, 0, count - 1):
             member_id = member.decode() if isinstance(member, bytes) else member
@@ -412,16 +425,12 @@ def _trim_telemetry(self, queue_name: str, *, now_ms: float) -> None:
             cutoff,
         )
         pipe.zremrangebyscore(
-            self.SUCCESSFUL_RESULTS_KEY.format(
-                prefix=self.key_prefix, queue_name=queue_name
-            ),
+            self._segment_key(TaskResultStatus.SUCCESSFUL, queue_name),
             0,
             cutoff,
         )
         pipe.zremrangebyscore(
-            self.FAILED_RESULTS_KEY.format(
-                prefix=self.key_prefix, queue_name=queue_name
-            ),
+            self._segment_key(TaskResultStatus.FAILED, queue_name),
             0,
             cutoff,
         )
@@ -436,21 +445,15 @@ def telemetry(
         retention_start_ms = now_ms - self.result_ttl.total_seconds() * 1000
         pipe = self.client.pipeline()
         for queue_name in self.queues:
-            pipe.zcard(
-                self.QUEUE_KEY.format(prefix=self.key_prefix, queue_name=queue_name)
-            )
-            pipe.zcard(
-                self.RUNNING_KEY.format(prefix=self.key_prefix, queue_name=queue_name)
-            )
+            pipe.zcard(self._segment_key(TaskResultStatus.READY, queue_name))
+            pipe.zcard(self._segment_key(TaskResultStatus.RUNNING, queue_name))
             pipe.zcard(
                 self.DEFERRED_KEY.format(prefix=self.key_prefix, queue_name=queue_name)
             )
-            successful_results_key = self.SUCCESSFUL_RESULTS_KEY.format(
-                prefix=self.key_prefix, queue_name=queue_name
-            )
-            failed_results_key = self.FAILED_RESULTS_KEY.format(
-                prefix=self.key_prefix, queue_name=queue_name
+            successful_results_key = self._segment_key(
+                TaskResultStatus.SUCCESSFUL, queue_name
             )
+            failed_results_key = self._segment_key(TaskResultStatus.FAILED, queue_name)
             pipe.zcard(successful_results_key)
             pipe.zcard(failed_results_key)
             ingress_key = self.INGRESS_KEY.format(
diff --git a/threadmill/inspector/app.py b/threadmill/inspector/app.py
index 9cc2d5f..3b27ad9 100644
--- a/threadmill/inspector/app.py
+++ b/threadmill/inspector/app.py
@@ -30,6 +30,7 @@
 )
 
 from ..backends.base import BackendTelemetry, ThreadmillTaskBackend
+from .screens import ConfirmScreen, PurgeScreen
 
 logger = logging.getLogger(__name__)
 
@@ -163,6 +164,21 @@ class TaskList(Vertical):
     counts: reactive[dict[str, int]] = reactive({})
     selected_task: reactive[TaskResult | None] = reactive(None)
 
+    BINDINGS = [
+        Binding("f5", "refresh", "Refresh"),
+        Binding("r", "requeue", "Requeue"),
+        Binding("d", "dequeue", "Drop"),
+    ]
+
+    def check_action(self, action: str, parameters: tuple[object, ...]) -> bool | None:
+        """Show requeue only on the Failed tab, dequeue on non-Running tabs."""
+        status = self.active_status()
+        if action == "requeue":
+            return status == TaskResultStatus.FAILED
+        if action == "dequeue":
+            return status is not None and status != TaskResultStatus.RUNNING
+        return True
+
     def compose(self) -> ComposeResult:
         with TabbedContent(initial="tab-ready") as tabs:
             for label, _ in TAB_STATUSES:
@@ -212,6 +228,11 @@ def switch_tab(self, tab_id: str) -> None:
         """Activate the tab with the given id."""
         self._tabs.active = tab_id
 
+    def action_refresh(self) -> None:
+        """Refresh telemetry and re-fetch the task list."""
+        self.app._refresh_telemetry()
+        self.refresh_tasks()
+
     def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
         """Notify the app when a task row is selected."""
         if isinstance(event.row_key.value, str):
@@ -227,6 +248,7 @@ def on_tabbed_content_tab_activated(
     ) -> None:
         """Refresh the visible table when the user switches tabs."""
         self._refresh_data()
+        self.refresh_bindings()
 
     def _select_task_by_id(self, task_id: str) -> None:
         """Find the task result matching the row key in the current results."""
@@ -235,6 +257,57 @@ def _select_task_by_id(self, task_id: str) -> None:
             None,
         )
 
+    def active_status(self) -> TaskResultStatus | None:
+        """Return the status segment of the currently active task tab."""
+        tab_id = self._tabs.active.removeprefix("tab-")
+        return next(
+            (status for label, status in TAB_STATUSES if label.lower() == tab_id),
+            None,
+        )
+
+    def action_requeue(self) -> None:
+        """Requeue the selected failed task after confirmation."""
+        task = self.selected_task
+        if task and self.active_status() == TaskResultStatus.FAILED:
+            self.app.push_screen(
+                ConfirmScreen(f"Requeue task {task.id[:8]}?"),
+                lambda confirmed: self._do_requeue(confirmed, task),
+            )
+
+    def _do_requeue(self, confirmed: bool, task: TaskResult) -> None:
+        if not confirmed:
+            return
+        try:
+            self.backend.requeue(task, datetime.datetime.now(datetime.UTC))
+        except Exception:  # noqa: BLE001
+            logger.exception("Requeue failed for task %s", task.id)
+            self.app.notify("Requeue failed.", severity="error")
+            return
+        self.app._after_action("Task requeued.")
+
+    def action_dequeue(self) -> None:
+        """Remove the selected task from its current segment after confirmation."""
+        task = self.selected_task
+        if task and self.active_status() is not None:
+            self.app.push_screen(
+                ConfirmScreen(
+                    f"Delete task {task.task.func.__name__} ({task.id[:8]})?",
+                    danger=True,
+                ),
+                lambda confirmed: self._do_dequeue(confirmed, task),
+            )
+
+    def _do_dequeue(self, confirmed: bool, task: TaskResult) -> None:
+        if not confirmed:
+            return
+        try:
+            self.backend.dequeue(task)
+        except Exception:  # noqa: BLE001
+            logger.exception("Delete failed for task %s", task.id)
+            self.app.notify("Delete failed.", severity="error")
+            return
+        self.app._after_action("Task deleted.")
+
     def _refresh_data(self) -> None:
         """Fetch and display tasks for the current queue and active tab."""
         backend = self.backend
@@ -291,6 +364,10 @@ class QueueList(ListView):
 
     telemetry: reactive[BackendTelemetry | None] = reactive(None)
 
+    BINDINGS = [
+        Binding("p", "purge", "Purge"),
+    ]
+
     def __init__(self, **kwargs: Any) -> None:
         super().__init__(**kwargs)
         self._items: dict[str, QueueItem] = {}
@@ -300,12 +377,10 @@ def compose(self) -> ComposeResult:
         self.border_title = "Queues"
 
     def watch_telemetry(self, telemetry: BackendTelemetry | None) -> None:
-        """Refresh queue labels when a new telemetry snapshot arrives."""
         if telemetry is not None:
             self.update_telemetry(telemetry)
 
     def update_telemetry(self, telemetry: BackendTelemetry) -> None:
-        """Refresh queue labels from a new telemetry snapshot."""
         queues = telemetry.queues
         for queue_name, stats in sorted(queues.items()):
             theme = BUILTIN_THEMES[self.app.theme]
@@ -334,21 +409,40 @@ def update_telemetry(self, telemetry: BackendTelemetry) -> None:
             self._notify_selection(target)
 
     def _notify_selection(self, queue_name: str) -> None:
-        """Tell the app which queue to display."""
         self.app._task_list.queue_name = queue_name
 
+    def action_purge(self) -> None:
+        item = self.highlighted_child
+        if not isinstance(item, QueueItem):
+            return
+        queue_name = item.queue_name
+        self.app.push_screen(
+            PurgeScreen(queue_name),
+            lambda confirmed: self._do_purge(confirmed, queue_name),
+        )
+
+    def _do_purge(self, confirmed: bool, queue_name: str) -> None:
+        if not confirmed:
+            return
+        try:
+            self.app.backend.purge(queue_name)
+        except Exception:  # noqa: BLE001
+            logger.exception("Purge failed for %r", queue_name)
+            self.app.notify("Purge failed.", severity="error")
+            return
+        self.app._after_action(f"Purged queue {queue_name}.")
+
 
 class InspectorApp(App):
     """Threadmill TUI inspector with backend/queue/task panes."""
 
     CSS_PATH = "inspector.scss"
     BINDINGS = [
-        Binding("q", "quit", "Quit"),
-        Binding("f5", "refresh", "Refresh"),
         *(
             Binding(key, f"switch_tab('tab-{tab_id}')", tab_id.capitalize())
             for tab_id, key in TAB_KEYS.items()
         ),
+        Binding("q", "quit", "Quit"),
     ]
 
     backend: reactive[ThreadmillTaskBackend] = reactive(None)
@@ -415,8 +509,9 @@ def action_quit(self) -> None:
         """Exit the TUI."""
         self.exit()
 
-    def action_refresh(self) -> None:
-        """Refresh queue stats and re-fetch the task list on demand."""
+    def _after_action(self, message: str) -> None:
+        """Refresh telemetry and tasks after a mutating queue action."""
+        self.notify(message)
         self._refresh_telemetry()
         self._task_list.refresh_tasks()
 
diff --git a/threadmill/inspector/inspector.scss b/threadmill/inspector/inspector.scss
index eab55ba..067f694 100644
--- a/threadmill/inspector/inspector.scss
+++ b/threadmill/inspector/inspector.scss
@@ -2,6 +2,11 @@ Screen {
     background: $background;
 }
 
+ConfirmScreen,
+PurgeScreen {
+    align: center middle;
+}
+
 * {
     scrollbar-color: $primary 10%;
     scrollbar-color-hover: $primary 80%;
@@ -104,3 +109,29 @@ TaskDetail {
     overflow: scroll;
     padding: 0 1;
 }
+
+#dialog {
+    width: 60;
+    height: auto;
+    background: $surface;
+    padding: 1 2;
+}
+
+#dialog.warning {
+    border: thick $warning;
+}
+
+#dialog.danger {
+    border: thick $error;
+}
+
+#prompt {
+    text-align: center;
+    margin-bottom: 1;
+}
+
+#instruction {
+    text-align: center;
+    color: $text-muted;
+    margin-top: 1;
+}
diff --git a/threadmill/inspector/screens.py b/threadmill/inspector/screens.py
new file mode 100644
index 0000000..9b1a805
--- /dev/null
+++ b/threadmill/inspector/screens.py
@@ -0,0 +1,52 @@
+"""Modal confirmation screens for inspector queue actions."""
+
+from textual.app import ComposeResult
+from textual.containers import Vertical
+from textual.events import Key
+from textual.screen import ModalScreen
+from textual.widgets import Input, Label
+
+
+class ConfirmScreen(ModalScreen[bool]):
+    """Yes/no confirmation: Enter confirms, Esc cancels."""
+
+    def __init__(self, prompt: str, *, danger: bool = False) -> None:
+        super().__init__()
+        self._prompt = prompt
+        self._danger = danger
+
+    def compose(self) -> ComposeResult:
+        with Vertical(id="dialog", classes="danger" if self._danger else "warning"):
+            yield Label(self._prompt, id="prompt")
+            yield Label("Press Enter to confirm, Esc to cancel.", id="instruction")
+
+    def on_key(self, event: Key) -> None:
+        match event.key:
+            case "enter":
+                self.dismiss(True)
+            case "escape":
+                self.dismiss(False)
+
+
+class PurgeScreen(ModalScreen[bool]):
+    """Type the queue name to confirm a purge."""
+
+    def __init__(self, queue_name: str) -> None:
+        super().__init__()
+        self._queue_name = queue_name
+
+    def compose(self) -> ComposeResult:
+        with Vertical(id="dialog", classes="danger"):
+            yield Label(f"Purge queue [bold]{self._queue_name}[/bold]?", id="prompt")
+            yield Label(
+                "Type the queue name to confirm, Esc to cancel.", id="instruction"
+            )
+            yield Input(placeholder=self._queue_name, id="confirm-input")
+
+    def on_input_submitted(self, event: Input.Submitted) -> None:
+        if event.value == self._queue_name:
+            self.dismiss(True)
+
+    def on_key(self, event: Key) -> None:
+        if event.key == "escape":
+            self.dismiss(False)