feat: add public queue admin APIs (requeue, shutdown, resources)

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> {

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)
     }

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)? {

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>;

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);

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));

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. |

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()`

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

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

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.
 

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}
 ```
 

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

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>