From a8137bf0d4a18eb2b6a119f3ba6c5a82d2fb7086 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 11 Jul 2026 08:30:26 +0530
Subject: [PATCH 1/8] feat(core): add requeue_stuck storage operation

Atomically resets a stuck Running job to Pending, releases its
execution claim, and clears any stale cancel request, without
consuming retry budget. No automatic path recovers a Running job
whose worker is hung-but-heartbeating or whose claim row is gone.
---
 .../src/storage/diesel_common/jobs.rs         | 30 +++++++++
 crates/taskito-core/src/storage/mod.rs        |  6 ++
 .../src/storage/redis_backend/jobs/state.rs   | 67 +++++++++++++++++++
 crates/taskito-core/src/storage/traits.rs     |  7 ++
 .../taskito-core/tests/rust/storage_tests.rs  | 40 +++++++++++
 5 files changed, 150 insertions(+)

diff --git a/crates/taskito-core/src/storage/diesel_common/jobs.rs b/crates/taskito-core/src/storage/diesel_common/jobs.rs
index 022a1f6a..28dd4a53 100644
--- a/crates/taskito-core/src/storage/diesel_common/jobs.rs
+++ b/crates/taskito-core/src/storage/diesel_common/jobs.rs
@@ -676,6 +676,36 @@ macro_rules! impl_diesel_job_ops {
                 Ok(())
             }
 
+            /// Force a `Running` job back to `Pending` and delete its
+            /// execution claim in one transaction. The status filter gates
+            /// the operation (missing / not-Running rows update nothing);
+            /// the claim must be deleted, not transferred, because
+            /// `claim_execution` is insert-only and a leftover row would
+            /// block the next worker's claim. Clearing `cancel_requested`
+            /// keeps a stale cancel request from killing the fresh attempt.
+            pub fn requeue_stuck(&self, id: &str, now: i64) -> Result<bool> {
+                self.write_transaction(|conn| {
+                    let affected = diesel::update(jobs::table)
+                        .filter(jobs::id.eq(id))
+                        .filter(jobs::status.eq(JobStatus::Running as i32))
+                        .set((
+                            jobs::status.eq(JobStatus::Pending as i32),
+                            jobs::scheduled_at.eq(now),
+                            jobs::started_at.eq(None::<i64>),
+                            jobs::completed_at.eq(None::<i64>),
+                            jobs::error.eq(None::<String>),
+                            jobs::cancel_requested.eq(0),
+                        ))
+                        .execute(conn)?;
+                    if affected == 0 {
+                        return Ok(false);
+                    }
+                    diesel::delete(execution_claims::table.filter(execution_claims::job_id.eq(id)))
+                        .execute(conn)?;
+                    Ok(true)
+                })
+            }
+
             /// Cancel a pending job and cascade-cancel its dependents. The
             /// cancelled job moves from `jobs` into `archived_jobs`.
             pub fn cancel_job(&self, id: &str) -> Result<bool> {
diff --git a/crates/taskito-core/src/storage/mod.rs b/crates/taskito-core/src/storage/mod.rs
index efd8057f..6c3139d0 100644
--- a/crates/taskito-core/src/storage/mod.rs
+++ b/crates/taskito-core/src/storage/mod.rs
@@ -155,6 +155,9 @@ macro_rules! impl_storage {
             fn reschedule(&self, id: &str, next_scheduled_at: i64) -> $crate::error::Result<()> {
                 self.reschedule(id, next_scheduled_at)
             }
+            fn requeue_stuck(&self, id: &str, now: i64) -> $crate::error::Result<bool> {
+                self.requeue_stuck(id, now)
+            }
             fn cancel_job(&self, id: &str) -> $crate::error::Result<bool> {
                 self.cancel_job(id)
             }
@@ -735,6 +738,9 @@ impl Storage for StorageBackend {
     fn reschedule(&self, id: &str, next_scheduled_at: i64) -> Result<()> {
         delegate!(self, reschedule, id, next_scheduled_at)
     }
+    fn requeue_stuck(&self, id: &str, now: i64) -> Result<bool> {
+        delegate!(self, requeue_stuck, id, now)
+    }
     fn cancel_job(&self, id: &str) -> Result<bool> {
         delegate!(self, cancel_job, id)
     }
diff --git a/crates/taskito-core/src/storage/redis_backend/jobs/state.rs b/crates/taskito-core/src/storage/redis_backend/jobs/state.rs
index 9e7054b6..1d8a9c01 100644
--- a/crates/taskito-core/src/storage/redis_backend/jobs/state.rs
+++ b/crates/taskito-core/src/storage/redis_backend/jobs/state.rs
@@ -3,6 +3,7 @@
 
 use redis::Commands;
 
+use super::dequeue_score;
 use crate::error::{QueueError, Result};
 use crate::job::{now_millis, JobStatus};
 use crate::storage::redis_backend::{map_err, RedisStorage};
@@ -27,6 +28,24 @@ const REQUEST_CANCEL_IF_RUNNING: &str = r#"
     return 1
 "#;
 
+/// Lua: move a job from Running back to Pending and release its execution
+/// claim, all-or-nothing. Guarded by Running-set membership (never decodes
+/// `job.status` in Lua) so a job archived between the Rust read and this
+/// script is a no-op. Also drops any pending cancel request so the fresh
+/// attempt isn't insta-cancelled, and clears the claim key + its time index
+/// so a healthy worker can re-claim. Returns 1 if applied, 0 otherwise.
+const REQUEUE_STUCK_IF_RUNNING: &str = r#"
+    if redis.call('SISMEMBER', KEYS[2], ARGV[1]) == 0 then return 0 end
+    redis.call('SET', KEYS[1], ARGV[2])
+    redis.call('SREM', KEYS[2], ARGV[1])
+    redis.call('SADD', KEYS[3], ARGV[1])
+    redis.call('ZADD', KEYS[4], ARGV[3], ARGV[1])
+    redis.call('SREM', KEYS[5], ARGV[1])
+    redis.call('DEL', KEYS[6])
+    redis.call('ZREM', KEYS[7], ARGV[1])
+    return 1
+"#;
+
 impl RedisStorage {
     pub fn complete(&self, id: &str, result_bytes: Option<Vec<u8>>) -> Result<()> {
         let mut conn = self.conn()?;
@@ -108,6 +127,54 @@ impl RedisStorage {
         Ok(())
     }
 
+    /// Force a `Running` job back to `Pending` and release its execution
+    /// claim atomically. Preserves retry budget (no `retry_count` bump) and
+    /// clears any pending cancel request. Returns `false` when the job is
+    /// missing or not `Running`.
+    pub fn requeue_stuck(&self, id: &str, now: i64) -> Result<bool> {
+        let mut conn = self.conn()?;
+        let mut job = match self.get_job(id)? {
+            Some(j) => j,
+            None => return Ok(false),
+        };
+        if job.status != JobStatus::Running {
+            return Ok(false);
+        }
+
+        job.status = JobStatus::Pending;
+        job.scheduled_at = now;
+        job.started_at = None;
+        job.completed_at = None;
+        job.error = None;
+        job.cancel_requested = false;
+
+        let job_json = serde_json::to_string(&job).map_err(|e| QueueError::Other(e.to_string()))?;
+        let job_key = self.key(&["job", id]);
+        let running_key = self.key(&["jobs", "status", &(JobStatus::Running as i32).to_string()]);
+        let pending_key = self.key(&["jobs", "status", &(JobStatus::Pending as i32).to_string()]);
+        let queue_key = self.key(&["queue", &job.queue, "pending"]);
+        let cancel_set = self.key(&["jobs", "cancel_requested"]);
+        let claim_key = self.key(&["exec_claim", id]);
+        let claim_index_key = self.key(&["exec_claims", "by_time"]);
+        let score = dequeue_score(job.priority, job.scheduled_at);
+
+        let applied: i32 = redis::Script::new(REQUEUE_STUCK_IF_RUNNING)
+            .key(&job_key)
+            .key(&running_key)
+            .key(&pending_key)
+            .key(&queue_key)
+            .key(&cancel_set)
+            .key(&claim_key)
+            .key(&claim_index_key)
+            .arg(id)
+            .arg(&job_json)
+            .arg(score)
+            .invoke(&mut conn)
+            .map_err(map_err)?;
+
+        Ok(applied == 1)
+    }
+
     pub fn cancel_job(&self, id: &str) -> Result<bool> {
         let mut conn = self.conn()?;
         let job = match self.get_job(id)? {
diff --git a/crates/taskito-core/src/storage/traits.rs b/crates/taskito-core/src/storage/traits.rs
index 06e75cdc..21489623 100644
--- a/crates/taskito-core/src/storage/traits.rs
+++ b/crates/taskito-core/src/storage/traits.rs
@@ -48,6 +48,13 @@ pub trait Storage: Send + Sync + Clone {
     /// concurrency cap, channel backpressure) where the job never executed,
     /// unlike [`retry`](Self::retry) which increments `retry_count`.
     fn reschedule(&self, id: &str, next_scheduled_at: i64) -> Result<()>;
+    /// Force a `Running` job back to `Pending` and release its execution
+    /// claim atomically, so a healthy worker can re-claim it. Preserves the
+    /// retry budget (operator action, mirrors [`reschedule`](Self::reschedule))
+    /// and clears any pending cancel request. Returns `false` when the job is
+    /// missing or not `Running`. Only for confirmed-dead/hung workers: a
+    /// still-alive owner may finish the old attempt, double-executing the job.
+    fn requeue_stuck(&self, id: &str, now: i64) -> Result<bool>;
     fn cancel_job(&self, id: &str) -> Result<bool>;
     fn request_cancel(&self, id: &str) -> Result<bool>;
     fn is_cancel_requested(&self, id: &str) -> Result<bool>;
diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs
index 5ac813f6..81841f76 100644
--- a/crates/taskito-core/tests/rust/storage_tests.rs
+++ b/crates/taskito-core/tests/rust/storage_tests.rs
@@ -492,6 +492,45 @@ fn test_reclaim_execution(s: &impl Storage) {
     s.complete_execution(colon_job).unwrap();
 }
 
+fn test_requeue_stuck(s: &impl Storage) {
+    // Operator rescue for a stuck Running job: back to Pending, claim
+    // released, retry budget and cancel flag reset — all atomically.
+    let q = "q-requeue-stuck";
+    let job = s.enqueue(make_job(q, "stuck_task")).unwrap();
+    let t0 = now_millis();
+    s.dequeue(q, t0, None).unwrap().unwrap(); // Running
+    assert!(s.claim_execution(&job.id, "hung-worker").unwrap());
+    assert!(s.request_cancel(&job.id).unwrap());
+
+    assert!(s.requeue_stuck(&job.id, t0).unwrap());
+
+    let requeued = s.get_job(&job.id).unwrap().unwrap();
+    assert_eq!(requeued.status, JobStatus::Pending);
+    assert_eq!(
+        requeued.retry_count, 0,
+        "operator rescue must not consume retry budget"
+    );
+    assert!(requeued.started_at.is_none());
+    assert!(
+        !s.is_cancel_requested(&job.id).unwrap(),
+        "a stale cancel request must not kill the fresh attempt"
+    );
+    // The claim was deleted, not transferred — an insert-only claim succeeds.
+    assert!(s.claim_execution(&job.id, "rescuer").unwrap());
+    // And the job is dequeuable again.
+    let redispatched = s.dequeue(q, now_millis() + 1000, None).unwrap().unwrap();
+    assert_eq!(redispatched.id, job.id);
+
+    // Not-Running and missing jobs are a no-op `false`, never an error.
+    s.complete(&job.id, None).unwrap();
+    s.complete_execution(&job.id).unwrap();
+    assert!(
+        !s.requeue_stuck(&job.id, t0).unwrap(),
+        "completed jobs are not requeueable"
+    );
+    assert!(!s.requeue_stuck("no-such-job", t0).unwrap());
+}
+
 fn test_reap_orphaned_jobs(s: &impl Storage) {
     // A running job whose claim owner is not in the live set is orphaned and
     // paired with that dead owner; a live owner or an empty set yields nothing.
@@ -876,6 +915,7 @@ fn run_storage_tests(s: &impl Storage) {
     test_execution_claims_purge(s);
     test_reap_stale_jobs(s);
     test_reclaim_execution(s);
+    test_requeue_stuck(s);
     test_reap_orphaned_jobs(s);
     test_dashboard_settings(s);
     test_immediate_archival(s);

From ffd4915e658878b242b13f028ee40dc8ac4f853f Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 11 Jul 2026 08:35:37 +0530
Subject: [PATCH 2/8] feat(python): add requeue_job for stuck running jobs

---
 .../taskito-python/src/py_queue/inspection.rs |  8 +++
 sdks/python/taskito/_taskito.pyi              |  1 +
 sdks/python/taskito/async_support/mixins.py   |  5 ++
 sdks/python/taskito/mixins/operations.py      | 11 ++++
 sdks/python/tests/core/test_requeue.py        | 58 +++++++++++++++++++
 5 files changed, 83 insertions(+)
 create mode 100644 sdks/python/tests/core/test_requeue.py

diff --git a/crates/taskito-python/src/py_queue/inspection.rs b/crates/taskito-python/src/py_queue/inspection.rs
index 8fcd42a8..8a499918 100644
--- a/crates/taskito-python/src/py_queue/inspection.rs
+++ b/crates/taskito-python/src/py_queue/inspection.rs
@@ -161,6 +161,14 @@ impl PyQueue {
             .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
     }
 
+    /// Force a stuck Running job back to Pending, releasing its execution
+    /// claim. Returns false when the job is missing or not Running.
+    pub fn requeue_job(&self, job_id: &str) -> PyResult<bool> {
+        self.storage
+            .requeue_stuck(job_id, now_millis())
+            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
+    }
+
     /// Purge dead letter entries older than given seconds ago.
     pub fn purge_dead(&self, older_than_seconds: i64) -> PyResult<u64> {
         let cutoff = now_millis().saturating_sub(older_than_seconds.saturating_mul(1000));
diff --git a/sdks/python/taskito/_taskito.pyi b/sdks/python/taskito/_taskito.pyi
index 4a4a8fdf..78204319 100644
--- a/sdks/python/taskito/_taskito.pyi
+++ b/sdks/python/taskito/_taskito.pyi
@@ -169,6 +169,7 @@ class PyQueue:
     ) -> list[dict[str, Any]]: ...
     def purge_dead_by_task(self, task_name: str) -> int: ...
     def retry_dead(self, dead_id: str) -> str: ...
+    def requeue_job(self, job_id: str) -> bool: ...
     def purge_dead(self, older_than_seconds: int) -> int: ...
     def delete_dead(self, dead_id: str) -> bool: ...
     def register_periodic(
diff --git a/sdks/python/taskito/async_support/mixins.py b/sdks/python/taskito/async_support/mixins.py
index 5d9ca50e..a41b95f5 100644
--- a/sdks/python/taskito/async_support/mixins.py
+++ b/sdks/python/taskito/async_support/mixins.py
@@ -80,6 +80,7 @@ def query_logs(
         ) -> list[dict]: ...
         def dead_letters(self, limit: int = ..., offset: int = ...) -> list[dict]: ...
         def retry_dead(self, dead_id: str) -> str: ...
+        def requeue_job(self, job_id: str) -> bool: ...
         def delete_dead(self, dead_id: str) -> bool: ...
         def replay(self, job_id: str) -> JobResult: ...
         def replay_history(self, job_id: str) -> list[dict]: ...
@@ -146,6 +147,10 @@ async def aretry_dead(self, dead_id: str) -> str:
         """Async version of :meth:`retry_dead`."""
         return await self._run_sync(self.retry_dead, dead_id)
 
+    async def arequeue_job(self, job_id: str) -> bool:
+        """Async version of :meth:`requeue_job`."""
+        return await self._run_sync(self.requeue_job, job_id)
+
     async def adelete_dead(self, dead_id: str) -> bool:
         """Async version of :meth:`delete_dead`."""
         return await self._run_sync(self.delete_dead, dead_id)
diff --git a/sdks/python/taskito/mixins/operations.py b/sdks/python/taskito/mixins/operations.py
index e9c86b4b..31753d42 100644
--- a/sdks/python/taskito/mixins/operations.py
+++ b/sdks/python/taskito/mixins/operations.py
@@ -24,6 +24,17 @@ def retry_dead(self, dead_id: str) -> str:
         """Re-enqueue a dead letter job. Returns new job ID."""
         return self._inner.retry_dead(dead_id)  # type: ignore[no-any-return]
 
+    def requeue_job(self, job_id: str) -> bool:
+        """Force a stuck Running job back to Pending so a healthy worker re-runs it.
+
+        Releases the job's execution claim atomically and preserves its retry
+        budget. Returns False when the job doesn't exist or isn't Running.
+
+        Warning: only use when the owning worker is confirmed dead or hung —
+        a still-alive worker may finish the old attempt and the job runs twice.
+        """
+        return self._inner.requeue_job(job_id)  # type: ignore[no-any-return]
+
     def purge_dead(self, older_than: int = 86400) -> int:
         """Delete dead letter entries older than a given age."""
         return self._inner.purge_dead(older_than)  # type: ignore[no-any-return]
diff --git a/sdks/python/tests/core/test_requeue.py b/sdks/python/tests/core/test_requeue.py
new file mode 100644
index 00000000..bae16f09
--- /dev/null
+++ b/sdks/python/tests/core/test_requeue.py
@@ -0,0 +1,58 @@
+"""Tests for Queue.requeue_job (stuck Running job recovery)."""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections.abc import Callable
+
+from taskito import Queue
+
+
+def test_requeue_pending_job_returns_false(queue: Queue) -> None:
+    """Only Running jobs are requeueable — a pending job is a no-op."""
+
+    @queue.task()
+    def idle() -> None:
+        pass
+
+    job = idle.delay()
+    assert queue.requeue_job(job.id) is False
+
+
+def test_requeue_nonexistent_job_returns_false(queue: Queue) -> None:
+    assert queue.requeue_job("nonexistent-id") is False
+
+
+def test_requeue_running_job_reruns_it(
+    queue: Queue,
+    run_worker: threading.Thread,
+    poll_until: Callable[..., None],
+) -> None:
+    """A stuck Running job goes back to Pending and is dispatched again."""
+    runs: list[float] = []
+    first_attempt_release = threading.Event()
+
+    @queue.task(max_retries=0)
+    def stuck() -> None:
+        runs.append(time.monotonic())
+        if len(runs) == 1:
+            # Simulate a hung attempt: block until the test releases it.
+            first_attempt_release.wait(timeout=30)
+
+    job = stuck.delay()
+    poll_until(lambda: len(runs) >= 1, message="first attempt never started")
+
+    assert queue.requeue_job(job.id) is True
+
+    poll_until(lambda: len(runs) >= 2, message="requeued job was never re-dispatched")
+    status = queue.get_job(job.id)
+    assert status is not None
+
+    first_attempt_release.set()
+
+    def job_completed() -> bool:
+        refreshed = queue.get_job(job.id)
+        return refreshed is not None and refreshed.status == "complete"
+
+    poll_until(job_completed, message="requeued attempt never completed")

From 70599aeac289765316a0b29942875ff990ac0f73 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 11 Jul 2026 08:37:22 +0530
Subject: [PATCH 3/8] feat(python): add public shutdown for graceful worker
 stop

---
 sdks/python/taskito/mixins/lifecycle.py | 10 ++++++++++
 sdks/python/tests/conftest.py           |  4 ++--
 sdks/python/tests/core/test_shutdown.py | 13 +++++++++----
 3 files changed, 21 insertions(+), 6 deletions(-)

diff --git a/sdks/python/taskito/mixins/lifecycle.py b/sdks/python/taskito/mixins/lifecycle.py
index a58c03bd..ef8f69ec 100644
--- a/sdks/python/taskito/mixins/lifecycle.py
+++ b/sdks/python/taskito/mixins/lifecycle.py
@@ -103,6 +103,16 @@ def _print_banner(self, queues: list[str]) -> None:
 
         print("\n".join(lines))
 
+    def shutdown(self) -> None:
+        """Request graceful worker shutdown (drain running tasks, then stop).
+
+        Programmatic equivalent of SIGINT/SIGTERM: the scheduler stops
+        dispatching and :meth:`run_worker` returns once running tasks finish,
+        bounded by the drain timeout. Non-blocking and safe to call from any
+        thread; a no-op when no worker is running.
+        """
+        self._inner.request_shutdown()
+
     def run_worker(
         self,
         queues: Sequence[str] | None = None,
diff --git a/sdks/python/tests/conftest.py b/sdks/python/tests/conftest.py
index 0de71a1f..fec0b6bd 100644
--- a/sdks/python/tests/conftest.py
+++ b/sdks/python/tests/conftest.py
@@ -67,7 +67,7 @@ def run_worker(queue: Queue) -> Generator[threading.Thread]:
     thread = threading.Thread(target=queue.run_worker, daemon=True)
     thread.start()
     yield thread
-    queue._inner.request_shutdown()
+    queue.shutdown()
     thread.join(timeout=5)
 
 
@@ -88,7 +88,7 @@ def _ctx() -> Generator[threading.Thread]:
         try:
             yield thread
         finally:
-            queue._inner.request_shutdown()
+            queue.shutdown()
             thread.join(timeout=5)
 
     return _ctx
diff --git a/sdks/python/tests/core/test_shutdown.py b/sdks/python/tests/core/test_shutdown.py
index 276ed5e0..0b2dd820 100644
--- a/sdks/python/tests/core/test_shutdown.py
+++ b/sdks/python/tests/core/test_shutdown.py
@@ -32,8 +32,8 @@ def slow_task() -> str:
         message="slow_task never reached running state",
     )
 
-    # Request graceful shutdown
-    queue._inner.request_shutdown()
+    # Request graceful shutdown via the public API
+    queue.shutdown()
 
     # Worker should finish the in-flight task
     worker_thread.join(timeout=10)
@@ -45,7 +45,7 @@ def slow_task() -> str:
 
 
 def test_shutdown_stops_worker(queue: Queue) -> None:
-    """request_shutdown causes run_worker to return."""
+    """queue.shutdown() causes run_worker to return."""
 
     @queue.task()
     def noop() -> None:
@@ -56,7 +56,12 @@ def noop() -> None:
 
     # Tiny grace window so the worker reaches its poll loop before shutdown.
     time.sleep(0.1)
-    queue._inner.request_shutdown()
+    queue.shutdown()
 
     worker_thread.join(timeout=10)
     assert not worker_thread.is_alive()
+
+
+def test_shutdown_without_worker_is_noop(queue: Queue) -> None:
+    """shutdown() on a queue with no running worker does nothing."""
+    queue.shutdown()  # must not raise

From fbd813191d4969bebbab7c714f94e5d01bf3a96f Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 11 Jul 2026 08:40:58 +0530
Subject: [PATCH 4/8] feat(python): add reload_resources for programmatic hot
 reload

---
 sdks/python/taskito/async_support/mixins.py   |  5 ++++
 sdks/python/taskito/mixins/resources.py       | 18 +++++++++++++
 sdks/python/tests/resources/test_resources.py | 27 +++++++++++++++++++
 3 files changed, 50 insertions(+)

diff --git a/sdks/python/taskito/async_support/mixins.py b/sdks/python/taskito/async_support/mixins.py
index a41b95f5..4de7bb42 100644
--- a/sdks/python/taskito/async_support/mixins.py
+++ b/sdks/python/taskito/async_support/mixins.py
@@ -102,6 +102,7 @@ def pause(self, queue_name: str = ...) -> None: ...
         def resume(self, queue_name: str = ...) -> None: ...
         def paused_queues(self) -> list[str]: ...
         def resource_status(self) -> list[dict[str, Any]]: ...
+        def reload_resources(self, names: list[str] | None = ...) -> dict[str, bool]: ...
 
     # -----------------------------------------------------------------
 
@@ -331,3 +332,7 @@ def _shutdown_once() -> None:
     async def aresource_status(self) -> list[dict[str, Any]]:
         """Async version of :meth:`resource_status`."""
         return self.resource_status()
+
+    async def areload_resources(self, names: list[str] | None = None) -> dict[str, bool]:
+        """Async version of :meth:`reload_resources`."""
+        return await self._run_sync(self.reload_resources, names)
diff --git a/sdks/python/taskito/mixins/resources.py b/sdks/python/taskito/mixins/resources.py
index 5ec810d5..b7fb9466 100644
--- a/sdks/python/taskito/mixins/resources.py
+++ b/sdks/python/taskito/mixins/resources.py
@@ -122,6 +122,24 @@ def health_check(self, name: str) -> bool:
         except Exception:
             return False
 
+    def reload_resources(self, names: list[str] | None = None) -> dict[str, bool]:
+        """Hot-reload reloadable worker resources (programmatic SIGHUP).
+
+        Tears down and re-creates each target resource in dependency order.
+
+        Args:
+            names: Resource names to reload. ``None`` reloads every resource
+                registered with ``reloadable=True``.
+
+        Returns:
+            Mapping of resource name to reload success. Empty when no worker
+            resource runtime is active in this process (no worker running).
+        """
+        runtime = self._resource_runtime
+        if runtime is None:
+            return {}
+        return runtime.reload(names)
+
     def load_resources(self, toml_path: str) -> None:
         """Load resource definitions from a TOML file.
 
diff --git a/sdks/python/tests/resources/test_resources.py b/sdks/python/tests/resources/test_resources.py
index 9d95aa16..63dd2abb 100644
--- a/sdks/python/tests/resources/test_resources.py
+++ b/sdks/python/tests/resources/test_resources.py
@@ -170,6 +170,33 @@ def test_from_test_overrides() -> None:
     assert rt.resolve("cache") == "mock_cache"
 
 
+def test_reload_resources_without_worker_returns_empty(queue: Queue) -> None:
+    """queue.reload_resources() is an empty no-op when no runtime is active."""
+    assert queue.reload_resources() == {}
+
+
+def test_reload_resources_recreates_reloadable(queue: Queue) -> None:
+    """queue.reload_resources() re-runs reloadable factories in the live runtime."""
+    counter = {"n": 0}
+
+    @queue.worker_resource("svc", reloadable=True)
+    def svc() -> int:
+        counter["n"] += 1
+        return counter["n"]
+
+    queue._resource_runtime = ResourceRuntime(queue._resource_definitions)
+    queue._resource_runtime.initialize()
+    try:
+        assert queue._resource_runtime.resolve("svc") == 1
+        assert queue.reload_resources() == {"svc": True}
+        assert queue._resource_runtime.resolve("svc") == 2
+        assert queue.reload_resources(["svc"]) == {"svc": True}
+        assert queue._resource_runtime.resolve("svc") == 3
+    finally:
+        queue._resource_runtime.teardown()
+        queue._resource_runtime = None
+
+
 # ---------------------------------------------------------------------------
 # Async factory
 # ---------------------------------------------------------------------------

From 32dbd42d55aaa684516bf55dde8ad3aefbb4dc38 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 11 Jul 2026 08:43:58 +0530
Subject: [PATCH 5/8] feat(python): add predicate_stats accessor

---
 sdks/python/taskito/mixins/predicates.py      |  7 ++++
 .../tests/core/test_predicates_enqueue.py     | 33 +++++++++++++++++++
 2 files changed, 40 insertions(+)

diff --git a/sdks/python/taskito/mixins/predicates.py b/sdks/python/taskito/mixins/predicates.py
index fc84e259..81c79dbd 100644
--- a/sdks/python/taskito/mixins/predicates.py
+++ b/sdks/python/taskito/mixins/predicates.py
@@ -91,6 +91,13 @@ def predicate_for(self, task_name: str) -> dict[str, Any] | None:
         """Return the serialized predicate for ``task_name`` or ``None``."""
         return self._task_predicate_serialized.get(task_name)
 
+    def predicate_stats(self) -> dict[str, int]:
+        """Return predicate outcome counters for this queue instance.
+
+        Keys: ``allowed``, ``denied``, ``deferred``, ``cancelled``, ``errors``.
+        """
+        return self._predicate_metrics.snapshot()
+
     def register_predicate(self, op: str, *, replace: bool = False) -> Callable[[type], type]:
         """Class decorator: register a custom :class:`Predicate` subclass.
 
diff --git a/sdks/python/tests/core/test_predicates_enqueue.py b/sdks/python/tests/core/test_predicates_enqueue.py
index c54711e3..d476d464 100644
--- a/sdks/python/tests/core/test_predicates_enqueue.py
+++ b/sdks/python/tests/core/test_predicates_enqueue.py
@@ -185,3 +185,36 @@ def t() -> str:
 
     poll_until(lambda: len(received) >= 1, timeout=2)
     assert received[0]["reason"] == "no good"
+
+
+def test_predicate_stats_counts_outcomes(queue: Queue) -> None:
+    """queue.predicate_stats() exposes the outcome counters publicly."""
+
+    @queue.task(predicate=_Const(True))
+    def allowed_task() -> str:
+        return "ok"
+
+    @queue.task(predicate=_Const(False), name="denied_task")
+    def denied_task() -> str:
+        return "ok"
+
+    @queue.task(predicate=_Const(Defer(seconds=60.0)), name="deferred_task")
+    def deferred_task() -> str:
+        return "ok"
+
+    @queue.task(predicate=_Const(Cancel(reason="nope")), name="cancelled_task")
+    def cancelled_task() -> str:
+        return "ok"
+
+    allowed_task.delay()
+    denied_task.delay()  # bare False counts as denied (then defers by default)
+    deferred_task.delay()
+    with pytest.raises(PredicateRejectedError):
+        cancelled_task.delay()
+
+    stats = queue.predicate_stats()
+    assert stats["allowed"] == 1
+    assert stats["denied"] == 1
+    assert stats["deferred"] == 1
+    assert stats["cancelled"] == 1
+    assert stats["errors"] == 0

From 705f0faad3654e6f5f3bfce6526112234593e7b8 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 11 Jul 2026 08:45:35 +0530
Subject: [PATCH 6/8] feat(python): add analyze_arguments interception dry-run

---
 sdks/python/taskito/mixins/resources.py         | 17 +++++++++++++++++
 .../python/tests/resources/test_interception.py | 17 +++++++++++++++++
 2 files changed, 34 insertions(+)

diff --git a/sdks/python/taskito/mixins/resources.py b/sdks/python/taskito/mixins/resources.py
index b7fb9466..a6d9af0c 100644
--- a/sdks/python/taskito/mixins/resources.py
+++ b/sdks/python/taskito/mixins/resources.py
@@ -6,11 +6,13 @@
 from typing import TYPE_CHECKING, Any
 
 from taskito.exceptions import CircularDependencyError
+from taskito.interception import InterceptionReport
 from taskito.resources.definition import ResourceDefinition, ResourceScope
 from taskito.resources.graph import detect_cycle
 from taskito.resources.toml_config import load_resources_from_toml
 
 if TYPE_CHECKING:
+    from taskito.interception.interceptor import ArgumentInterceptor
     from taskito.interception.metrics import InterceptionMetrics
     from taskito.proxies.metrics import ProxyMetrics
     from taskito.resources.runtime import ResourceRuntime
@@ -23,6 +25,7 @@ class QueueResourceMixin:
     _resource_runtime: ResourceRuntime | None
     _proxy_metrics: ProxyMetrics
     _interception_metrics: InterceptionMetrics | None
+    _interceptor: ArgumentInterceptor | None
 
     def worker_resource(
         self,
@@ -160,3 +163,17 @@ def interception_stats(self) -> dict[str, Any]:
         if self._interception_metrics is not None:
             return self._interception_metrics.to_dict()
         return {}
+
+    def analyze_arguments(
+        self, args: tuple = (), kwargs: dict[str, Any] | None = None
+    ) -> InterceptionReport:
+        """Dry-run the argument interceptor without enqueuing anything.
+
+        Returns an :class:`~taskito.interception.InterceptionReport` describing
+        the strategy chosen for each argument. The report is empty when the
+        queue was created with ``interception="off"`` (the default). For a
+        per-task variant, see :meth:`taskito.task.TaskWrapper.analyze`.
+        """
+        if self._interceptor is None:
+            return InterceptionReport()
+        return self._interceptor.analyze(args, kwargs or {})
diff --git a/sdks/python/tests/resources/test_interception.py b/sdks/python/tests/resources/test_interception.py
index 2ef882c7..eb4497d9 100644
--- a/sdks/python/tests/resources/test_interception.py
+++ b/sdks/python/tests/resources/test_interception.py
@@ -401,6 +401,23 @@ def my_task(x: int) -> None:
         report = my_task.analyze(42)
         assert len(report.entries) == 0
 
+    def test_queue_analyze_arguments(self, tmp_path: Any) -> None:
+        q = Queue(db_path=str(tmp_path / "test.db"), interception="strict")
+        uid = uuid.uuid4()
+
+        report = q.analyze_arguments((42, uid), {"created_at": datetime.datetime.now()})
+        assert isinstance(report, InterceptionReport)
+        assert len(report.entries) == 3
+        strategies = [e.strategy for e in report.entries]
+        assert Strategy.PASS in strategies
+        assert Strategy.CONVERT in strategies
+
+    def test_queue_analyze_arguments_off_mode(self, tmp_path: Any) -> None:
+        q = Queue(db_path=str(tmp_path / "test.db"))
+
+        report = q.analyze_arguments((42,), {})
+        assert len(report.entries) == 0
+
 
 # -- Custom type registration --
 

From 7b5e74669eeaab84b910976c5f8be8215996bca2 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 11 Jul 2026 08:52:06 +0530
Subject: [PATCH 7/8] feat(python): support expires default in task decorator

Also forwards the default through delay() and map(), which never
passed expires before.
---
 sdks/python/taskito/mixins/decorators.py |  6 ++
 sdks/python/taskito/task.py              |  9 ++-
 sdks/python/tests/core/test_expires.py   | 87 ++++++++++++++++++++++++
 3 files changed, 100 insertions(+), 2 deletions(-)
 create mode 100644 sdks/python/tests/core/test_expires.py

diff --git a/sdks/python/taskito/mixins/decorators.py b/sdks/python/taskito/mixins/decorators.py
index 6d701f1b..0f5616c2 100644
--- a/sdks/python/taskito/mixins/decorators.py
+++ b/sdks/python/taskito/mixins/decorators.py
@@ -363,6 +363,7 @@ def task(
         max_retries: int = 3,
         retry_backoff: float = 1.0,
         timeout: int = 300,
+        expires: float | None = None,
         priority: int = 0,
         rate_limit: str | None = None,
         queue: str = "default",
@@ -392,6 +393,10 @@ def task(
             max_retries: Max retry attempts on failure before moving to DLQ.
             retry_backoff: Base delay in seconds for exponential backoff between retries.
             timeout: Max execution time in seconds before the task is killed.
+            expires: Default job expiry in seconds — a job is skipped
+                (cancelled and archived) if it hasn't started within this
+                window after enqueue. Per-call ``apply_async(expires=...)``
+                overrides it. ``None`` (default) means jobs never expire.
             priority: Priority level (higher = more urgent).
             rate_limit: Rate limit string, e.g. ``"100/m"``, ``"10/s"``, ``"3600/h"``.
             queue: Named queue to submit to.
@@ -622,6 +627,7 @@ def decorator(fn: Callable) -> TaskWrapper:
                 default_queue=queue,
                 default_max_retries=max_retries,
                 default_timeout=timeout,
+                default_expires=expires,
                 inject=final_inject or None,
             )
 
diff --git a/sdks/python/taskito/task.py b/sdks/python/taskito/task.py
index db4b3023..afe26f2f 100644
--- a/sdks/python/taskito/task.py
+++ b/sdks/python/taskito/task.py
@@ -29,6 +29,7 @@ def __init__(
         default_queue: str,
         default_max_retries: int,
         default_timeout: int,
+        default_expires: float | None = None,
         inject: list[str] | None = None,
     ):
         self._fn = fn
@@ -38,6 +39,7 @@ def __init__(
         self._default_queue = default_queue
         self._default_max_retries = default_max_retries
         self._default_timeout = default_timeout
+        self._default_expires = default_expires
         self._inject = inject or []
         self._taskito_is_async: bool = False
         self._taskito_async_fn: Callable | None = None
@@ -69,6 +71,7 @@ def delay(self, *args: Any, **kwargs: Any) -> JobResult:
             queue=self._default_queue,
             max_retries=self._default_max_retries,
             timeout=self._default_timeout,
+            expires=self._default_expires,
         )
 
     def apply_async(
@@ -104,7 +107,8 @@ def apply_async(
             notes: Structured annotations dict (≤ 15 top-level fields). See
                 :mod:`taskito.notes`.
             depends_on: Job ID or list of job IDs that must complete first.
-            expires: Seconds until the job expires (skipped if not started by then).
+            expires: Seconds until the job expires (skipped if not started by
+                then). Overrides the task's ``expires=`` default.
             result_ttl: Per-job result TTL in seconds.
             idempotency_key: Explicit dedup key. A second submission with the
                 same key while the first job is pending or running returns the
@@ -127,7 +131,7 @@ def apply_async(
             metadata=metadata,
             notes=notes,
             depends_on=depends_on,
-            expires=expires,
+            expires=expires if expires is not None else self._default_expires,
             result_ttl=result_ttl,
             idempotency_key=idempotency_key,
             idempotent=idempotent,
@@ -142,6 +146,7 @@ def map(self, iterable: list[tuple]) -> list[JobResult]:
             queue=self._default_queue,
             max_retries=self._default_max_retries,
             timeout=self._default_timeout,
+            expires=self._default_expires,
         )
 
     def s(self, *args: Any, **kwargs: Any) -> Signature:
diff --git a/sdks/python/tests/core/test_expires.py b/sdks/python/tests/core/test_expires.py
new file mode 100644
index 00000000..d500de13
--- /dev/null
+++ b/sdks/python/tests/core/test_expires.py
@@ -0,0 +1,87 @@
+"""Tests for job expiry — the task-level default and per-call override."""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections.abc import Callable
+from pathlib import Path
+
+from taskito import Queue
+from taskito.middleware import TaskMiddleware
+
+
+class _CaptureEnqueue(TaskMiddleware):
+    """Records the options dict of every enqueue for assertions."""
+
+    def __init__(self) -> None:
+        self.options: list[dict] = []
+
+    def on_enqueue(self, task_name: str, args: tuple, kwargs: dict, options: dict) -> None:
+        self.options.append(dict(options))
+
+
+def _queue_with_capture(tmp_path: Path) -> tuple[Queue, _CaptureEnqueue]:
+    capture = _CaptureEnqueue()
+    queue = Queue(db_path=str(tmp_path / "test.db"), middleware=[capture])
+    return queue, capture
+
+
+def test_task_expires_default_flows_through_delay(tmp_path: Path) -> None:
+    queue, capture = _queue_with_capture(tmp_path)
+
+    @queue.task(expires=30)
+    def ephemeral() -> None:
+        pass
+
+    ephemeral.delay()
+    assert capture.options[-1]["expires"] == 30
+
+
+def test_apply_async_expires_overrides_task_default(tmp_path: Path) -> None:
+    queue, capture = _queue_with_capture(tmp_path)
+
+    @queue.task(expires=30)
+    def ephemeral() -> None:
+        pass
+
+    ephemeral.apply_async(expires=5)
+    assert capture.options[-1]["expires"] == 5
+
+    ephemeral.apply_async()
+    assert capture.options[-1]["expires"] == 30
+
+
+def test_no_expires_by_default(tmp_path: Path) -> None:
+    queue, capture = _queue_with_capture(tmp_path)
+
+    @queue.task()
+    def durable() -> None:
+        pass
+
+    durable.delay()
+    assert capture.options[-1]["expires"] is None
+
+
+def test_expired_job_never_executes(queue: Queue, poll_until: Callable[..., None]) -> None:
+    """A job past its expiry window is skipped, not run."""
+    ran = threading.Event()
+
+    @queue.task(expires=0.05)
+    def ephemeral() -> None:
+        ran.set()
+
+    job = ephemeral.delay()
+    time.sleep(0.2)  # let the expiry window lapse before any worker exists
+
+    worker_thread = threading.Thread(target=queue.run_worker, daemon=True)
+    worker_thread.start()
+
+    def job_expired() -> bool:
+        refreshed = queue.get_job(job.id)
+        return refreshed is not None and refreshed.status == "cancelled"
+
+    poll_until(job_expired, message="expired job never archived as cancelled")
+    assert not ran.is_set()
+    queue.shutdown()
+    worker_thread.join(timeout=5)

From a4205b0c1203e8ea2fd4af021fcbeb5d4c41dfe2 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 11 Jul 2026 08:59:13 +0530
Subject: [PATCH 8/8] docs: replace internal accessors with public APIs

---
 .../docs/python/api-reference/queue/index.mdx |  2 ++
 .../python/api-reference/queue/queues.mdx     | 13 +++++++++++
 .../python/api-reference/queue/resources.mdx  | 22 +++++++++++++++++++
 .../python/api-reference/queue/workers.mdx    | 11 ++++++++++
 .../python/guides/resources/configuration.mdx | 13 +++++++----
 .../docs/shared/guides/core/predicates.mdx    |  2 +-
 .../content/docs/shared/guides/core/tasks.mdx | 10 ++++-----
 .../docs/shared/guides/core/workers.mdx       |  5 +++--
 .../guides/operations/troubleshooting.mdx     | 16 ++++++++++++++
 .../shared/guides/resources/interception.mdx  |  9 ++++----
 10 files changed, 87 insertions(+), 16 deletions(-)

diff --git a/docs/content/docs/python/api-reference/queue/index.mdx b/docs/content/docs/python/api-reference/queue/index.mdx
index dfcc4d1c..fdb75106 100644
--- a/docs/content/docs/python/api-reference/queue/index.mdx
+++ b/docs/content/docs/python/api-reference/queue/index.mdx
@@ -78,6 +78,7 @@ Queue(
     max_retry_delay: int | None = None,
     timeout: int = 300,
     soft_timeout: float | None = None,
+    expires: float | None = None,
     priority: int = 0,
     rate_limit: str | None = None,
     queue: str = "default",
@@ -105,6 +106,7 @@ Register a function as a background task. Returns a [`TaskWrapper`](/python/api-
 | `max_retry_delay` | `int \| None` | `None` | Cap on backoff delay in seconds. Defaults to 300 s. |
 | `timeout` | `int` | `300` | Hard execution time limit in seconds. |
 | `soft_timeout` | `float \| None` | `None` | Cooperative time limit checked via `current_job.check_timeout()`. |
+| `expires` | `float \| None` | `None` | Default expiry in seconds — jobs not started within the window are skipped. Per-call `apply_async(expires=)` overrides. |
 | `priority` | `int` | `0` | Default priority (higher = more urgent). |
 | `rate_limit` | `str \| None` | `None` | Rate limit string, e.g. `"100/m"`. |
 | `queue` | `str` | `"default"` | Named queue to submit to. |
diff --git a/docs/content/docs/python/api-reference/queue/queues.mdx b/docs/content/docs/python/api-reference/queue/queues.mdx
index 0c8be079..623e80fd 100644
--- a/docs/content/docs/python/api-reference/queue/queues.mdx
+++ b/docs/content/docs/python/api-reference/queue/queues.mdx
@@ -154,6 +154,19 @@ queue.purge_dead(older_than: int = 86400) -> int
 
 Purge dead letter entries older than `older_than` seconds. Returns count deleted.
 
+### `queue.requeue_job()`
+
+```python
+queue.requeue_job(job_id: str) -> bool
+```
+
+Force a stuck `running` job back to `pending`, releasing its execution claim
+so a healthy worker can re-claim it. Preserves the retry budget and clears any
+pending cancel request. Returns `False` when the job doesn't exist or isn't
+running. Only for jobs whose owning worker is confirmed dead or hung — a
+still-alive worker may finish the old attempt and the job runs twice. Async
+twin: `arequeue_job()`.
+
 ## Cleanup
 
 ### `queue.purge_completed()`
diff --git a/docs/content/docs/python/api-reference/queue/resources.mdx b/docs/content/docs/python/api-reference/queue/resources.mdx
index 3e86104e..8b84a27f 100644
--- a/docs/content/docs/python/api-reference/queue/resources.mdx
+++ b/docs/content/docs/python/api-reference/queue/resources.mdx
@@ -74,6 +74,18 @@ queue.health_check(name: str) -> bool
 Run a resource's health check immediately. Returns `True` if healthy, `False`
 otherwise or if the runtime is not initialized.
 
+### `queue.reload_resources()`
+
+```python
+queue.reload_resources(names: list[str] | None = None) -> dict[str, bool]
+```
+
+Hot-reload reloadable worker resources — the programmatic equivalent of
+SIGHUP. Tears down and re-creates each target in dependency order. `None`
+reloads every resource registered with `reloadable=True`. Returns a
+name-to-success mapping; empty when no worker resource runtime is active in
+this process. Async twin: `areload_resources()`.
+
 ### `queue.resource_status()`
 
 ```python
@@ -112,6 +124,16 @@ Register a custom type with the interception system. Requires
 | `type_key` | `str \| None` | Dispatch key for the converter reconstructor. |
 | `proxy_handler` | `str \| None` | Handler name for `"proxy"` strategy. |
 
+### `queue.analyze_arguments()`
+
+```python
+queue.analyze_arguments(args: tuple = (), kwargs: dict | None = None) -> InterceptionReport
+```
+
+Dry-run the argument interceptor without enqueuing — the report describes the
+strategy chosen for each argument. Empty when the queue was created with
+`interception="off"`. Per-task variant: `my_task.analyze(...)`.
+
 ### `queue.interception_stats()`
 
 ```python
diff --git a/docs/content/docs/python/api-reference/queue/workers.mdx b/docs/content/docs/python/api-reference/queue/workers.mdx
index 2bd1784e..d474f6e5 100644
--- a/docs/content/docs/python/api-reference/queue/workers.mdx
+++ b/docs/content/docs/python/api-reference/queue/workers.mdx
@@ -28,6 +28,17 @@ Start the worker loop. **Blocks** until interrupted.
 | `pool` | `str` | `"thread"` | Worker pool type: `"thread"` or `"prefork"`. |
 | `app` | `str \| None` | `None` | Import path to Queue (e.g. `"myapp:queue"`). Required when `pool="prefork"`. |
 
+### `queue.shutdown()`
+
+```python
+queue.shutdown() -> None
+```
+
+Request graceful worker shutdown — the programmatic equivalent of
+SIGINT/SIGTERM. The scheduler stops dispatching and `run_worker()` returns
+once running tasks finish (bounded by `drain_timeout`). Non-blocking, safe
+from any thread; a no-op when no worker is running.
+
 ### `queue.workers()`
 
 ```python
diff --git a/docs/content/docs/python/guides/resources/configuration.mdx b/docs/content/docs/python/guides/resources/configuration.mdx
index 5c42be7a..9d7783f8 100644
--- a/docs/content/docs/python/guides/resources/configuration.mdx
+++ b/docs/content/docs/python/guides/resources/configuration.mdx
@@ -143,15 +143,20 @@ taskito reload --pid <worker-pid>
 taskito reload --pid <worker-pid> --resource feature_flags  # reload one resource
 ```
 
-Or programmatically from application code (internal accessor — the supported
-triggers are `SIGHUP` and `taskito reload` above; this attribute is not yet part
-of the stable API):
+Or programmatically from application code — reload everything reloadable, or
+name specific resources:
 
 ```python
-results = queue._resource_runtime.reload()
+results = queue.reload_resources()
 # {"feature_flags": True}
+
+queue.reload_resources(["feature_flags"])  # just one
 ```
 
+`reload_resources()` returns an empty dict when no worker is running in the
+process (there is no live resource runtime to reload). An async twin,
+`areload_resources()`, is available for async code.
+
 Only resources declared with `reloadable=True` are affected. Non-reloadable
 resources are left running — no teardown or reconnection.
 
diff --git a/docs/content/docs/shared/guides/core/predicates.mdx b/docs/content/docs/shared/guides/core/predicates.mdx
index 5242c3f7..f4cd4fcc 100644
--- a/docs/content/docs/shared/guides/core/predicates.mdx
+++ b/docs/content/docs/shared/guides/core/predicates.mdx
@@ -536,7 +536,7 @@ restored = Predicate.from_dict(blob)
 ## Metrics & events
 
 ```python
-queue._predicate_metrics.snapshot()  # internal accessor — not yet part of the stable API
+queue.predicate_stats()
 # {"allowed": 412, "denied": 3, "deferred": 8, "cancelled": 1, "errors": 0}
 ```
 
diff --git a/docs/content/docs/shared/guides/core/tasks.mdx b/docs/content/docs/shared/guides/core/tasks.mdx
index 3b7f3698..a65bce48 100644
--- a/docs/content/docs/shared/guides/core/tasks.mdx
+++ b/docs/content/docs/shared/guides/core/tasks.mdx
@@ -534,16 +534,16 @@ audit tasks) or when the payload isn't picklable. See
 
 ## Job expiration
 
-`expires` is an **enqueue-time** option (not a decorator option) — skip a job
-that wasn't started within the deadline:
+`expires` skips a job that wasn't started within the deadline. Set a default
+on the task, override it per call, or pass it only at enqueue time:
 
 ```python
-@queue.task()
+@queue.task(expires=300)  # every job skipped if not started within 5 minutes
 def time_sensitive():
     ...
 
-# Skip if not started within 5 minutes
-time_sensitive.apply_async(args=(), expires=300)
+time_sensitive.delay()                      # uses the 300s default
+time_sensitive.apply_async(args=(), expires=60)  # tighter window for this call
 ```
 
 ## Task naming
diff --git a/docs/content/docs/shared/guides/core/workers.mdx b/docs/content/docs/shared/guides/core/workers.mdx
index 3c198f89..1a507af2 100644
--- a/docs/content/docs/shared/guides/core/workers.mdx
+++ b/docs/content/docs/shared/guides/core/workers.mdx
@@ -426,8 +426,9 @@ endpoint:
   <Tab sdk="python">
 
 ```python
-# From another thread or signal handler
-queue._inner.request_shutdown()
+# From another thread or signal handler — non-blocking; run_worker()
+# returns once running tasks drain
+queue.shutdown()
 ```
 
   </Tab>
diff --git a/docs/content/docs/shared/guides/operations/troubleshooting.mdx b/docs/content/docs/shared/guides/operations/troubleshooting.mdx
index c0fbb522..71acdcc4 100644
--- a/docs/content/docs/shared/guides/operations/troubleshooting.mdx
+++ b/docs/content/docs/shared/guides/operations/troubleshooting.mdx
@@ -192,6 +192,22 @@ hood.
   forever — nothing detects that case without a timeout.
 </Callout>
 
+<SdkOnly sdk="python">
+
+For that last case — a job neither recovery mechanism will touch — force it
+back to `pending` manually:
+
+```python
+queue.requeue_job(job_id)  # True if it was Running and is now Pending
+```
+
+`requeue_job()` releases the job's execution claim so a healthy worker can
+pick it up, and preserves the retry budget. Only use it when the owning
+worker is confirmed dead or hung: if the old attempt is actually still
+running, it may finish later and the job executes twice.
+
+</SdkOnly>
+
 ## Worker is unresponsive
 
 **Symptom**: the worker process is alive but not processing jobs; its
diff --git a/docs/content/docs/shared/guides/resources/interception.mdx b/docs/content/docs/shared/guides/resources/interception.mdx
index 99b48327..2ec93932 100644
--- a/docs/content/docs/shared/guides/resources/interception.mdx
+++ b/docs/content/docs/shared/guides/resources/interception.mdx
@@ -255,7 +255,7 @@ transforming them:
 ```python
 from myapp.tasks import queue
 
-report = queue._interceptor.analyze(
+report = queue.analyze_arguments(
     args=(user_session, "Hello"),
     kwargs={"attachment": open("file.pdf", "rb")},
 )
@@ -266,9 +266,10 @@ print(report)
 #   kwargs.attachment  (BufferedReader)        → PROXY (handler=file)
 ```
 
-`analyze()` is a development and debugging tool. It reads the registry but
-makes no changes to arguments. `_interceptor` is an internal accessor — it is
-not yet part of the stable public API and may change between releases.
+`analyze_arguments()` is a development and debugging tool. It reads the
+registry but makes no changes to arguments, and returns an empty report when
+the queue was created with `interception="off"`. The per-task variant
+`my_task.analyze(...)` takes the task's own call arguments directly.
 
 ### Interception metrics