diff --git a/crates/taskito-core/BINDING_CONTRACT.md b/crates/taskito-core/BINDING_CONTRACT.md
index b82f63fb..80d32ddf 100644
--- a/crates/taskito-core/BINDING_CONTRACT.md
+++ b/crates/taskito-core/BINDING_CONTRACT.md
@@ -160,7 +160,7 @@ itself must match it.
   acking an older/equal id is a no-op (returns `false`). Like a Kafka offset commit,
   the caller is trusted to pass back an id it actually read.
 - Delivery is **at-least-once**: a consumer that reads but dies before acking
-  re-reads those messages. There is no per-message ack.
+  re-reads those messages. The cursor read has no per-message ack (see below).
 - **Retention** is min-cursor compaction: a message is dropped once every log
   subscriber on its topic has acked past it (Diesel deletes `id <= min(cursor)`;
   Redis `XTRIM MINID`). A topic with an unread subscriber keeps its backlog.
@@ -184,6 +184,26 @@ itself must match it.
 - Storage: the Diesel backends use a `topics` table (migration `m0006`); Redis uses
   a `topics` hash keyed by name. A shell surfaces `Topic { name, mode, retention_ms,
   created_at }`.
+**Per-message ack** (`lease_topic_messages`/`ack_message`/`nack_message`):
+- An opt-in **consumption choice** on a `log` subscription (not a registration
+  flag): instead of the cursor read, a consumer *leases* messages and acks/nacks
+  each individually, so a poison message never blocks its siblings.
+- `lease_topic_messages(topic, sub, limit, visibility_ms, now)` returns up to
+  `limit` **available** messages oldest-first — never delivered, or a prior lease
+  that expired (`now`-relative) or was nacked and never acked — and (re)leases
+  each for `visibility_ms`. In-flight (leased, un-expired) messages are skipped.
+- `ack_message` ends a delivery (never redelivered); `nack_message` makes it
+  available immediately (vs waiting out the timeout). Both return `false` when
+  there is no un-acked delivery. An un-acked lease that times out is redelivered.
+- Delivery state lives per `(subscription, message)`: Diesel `topic_deliveries`
+  table (migration `m0007`); Redis a `pmdeliv:<topic>:<sub>` hash mirroring it.
+- **Retention**: on a topic consumed purely per-message (every log sub has
+  acked), a message is compacted once every per-message subscriber has acked it;
+  a topic that mixes in a cursor subscriber falls back to `expires_at`. Its
+  delivery state is dropped with it (Diesel deletes the rows, Redis `HDEL`s the
+  fields). Both backends implement this — Diesel via `topic_deliveries`, Redis by
+  scanning the `pmdeliv:*` hashes during the purge sweep. A shell only marshals
+  the three calls; the core owns the state.
 
 **Test vector** (assert byte-exact in each shell that salts keys itself):
 key `evt-42`, topic `orders`, subscription `email` →
@@ -214,7 +234,8 @@ Grouped by concern (enumerated, not exhaustive — read the trait):
 - **Dead-letter**: `move_to_dlq`, `list_dead`, `retry_dead`.
 - **Topic pub/sub**: `register_subscription`, `list_subscriptions_for_topic`, `unsubscribe`,
   `publish_message`, `read_topic_messages`, `ack_topic_cursor`, `topic_log_stats`,
-  `declare_topic`, `get_topic`, `list_declared_topics` (see
+  `declare_topic`, `get_topic`, `list_declared_topics`, `lease_topic_messages`,
+  `ack_message`, `nack_message` (see
   [Topic pub/sub](#topic-pubsub-cross-sdk)).
 
 ## Lifecycle the shell drives
diff --git a/crates/taskito-core/migrations/m0007_topic_deliveries.rs b/crates/taskito-core/migrations/m0007_topic_deliveries.rs
new file mode 100644
index 00000000..a49f2e1c
--- /dev/null
+++ b/crates/taskito-core/migrations/m0007_topic_deliveries.rs
@@ -0,0 +1,65 @@
+//! Per-message ack for log topics (`0007_topic_deliveries`).
+//!
+//! S28 log subscriptions consume with a single high-water-mark cursor, so one
+//! poisoned message blocks the whole subscription. This adds an opt-in
+//! per-message mode: instead of the cursor read, a consumer *leases* each
+//! message with a visibility timeout and acks or nacks it individually. It's a
+//! consumption choice (which read methods you call), not a registration flag —
+//! delivery state per `(subscription, message)` lives in `topic_deliveries`, and
+//! an un-acked lease that expires is redelivered. Idempotent:
+//! `.if_not_exists()` on the table and index.
+
+use sea_query::{Alias, ColumnDef, Index, Table};
+
+use crate::storage::migrate::{ddl, Backend, Migration, Stmt};
+
+pub struct M0007TopicDeliveries;
+
+fn col(name: &str) -> ColumnDef {
+    ColumnDef::new(Alias::new(name))
+}
+
+fn t(name: &str) -> Alias {
+    Alias::new(name)
+}
+
+impl Migration for M0007TopicDeliveries {
+    fn version(&self) -> &'static str {
+        "0007_topic_deliveries"
+    }
+
+    fn up(&self, b: Backend) -> Vec<Stmt> {
+        // Per-(subscription, message) delivery state. `lease_expires_at` bounds
+        // an in-flight lease (0 = available for redelivery, e.g. after a nack);
+        // `acked` ends the delivery. `attempts` counts (re)deliveries.
+        let deliveries = Table::create()
+            .table(t("topic_deliveries"))
+            .if_not_exists()
+            .col(col("topic").text().not_null())
+            .col(col("subscription_name").text().not_null())
+            .col(col("message_id").text().not_null())
+            .col(col("acked").boolean().not_null().default(false))
+            .col(col("attempts").integer().not_null().default(0))
+            .col(col("lease_expires_at").big_integer().not_null().default(0))
+            .col(col("delivered_at").big_integer().not_null().default(0))
+            .primary_key(
+                Index::create()
+                    .col(t("topic"))
+                    .col(t("subscription_name"))
+                    .col(t("message_id")),
+            )
+            .to_owned();
+
+        // The lease read's anti-join filters by (topic, subscription, availability).
+        let by_sub_lease = Index::create()
+            .if_not_exists()
+            .name("idx_topic_deliveries_sub_lease")
+            .table(t("topic_deliveries"))
+            .col(t("topic"))
+            .col(t("subscription_name"))
+            .col(t("lease_expires_at"))
+            .to_owned();
+
+        vec![ddl(b, &deliveries), ddl(b, &by_sub_lease)]
+    }
+}
diff --git a/crates/taskito-core/src/storage/diesel_common/pubsub.rs b/crates/taskito-core/src/storage/diesel_common/pubsub.rs
index 74d6bb9a..2bcc02f9 100644
--- a/crates/taskito-core/src/storage/diesel_common/pubsub.rs
+++ b/crates/taskito-core/src/storage/diesel_common/pubsub.rs
@@ -352,7 +352,7 @@ macro_rules! impl_diesel_pubsub_ops {
                     .load(&mut conn)?;
                 ids.extend(expired);
 
-                // Then fully-acked messages per topic, within the remaining budget.
+                // Then cursor-fully-acked messages per topic, within the budget.
                 for (topic, floor) in floors {
                     let Some(floor) = floor else { continue };
                     let remaining = limit - ids.len() as i64;
@@ -368,6 +368,78 @@ macro_rules! impl_diesel_pubsub_ops {
                     ids.extend(acked);
                 }
 
+                // Per-message compaction: on a topic consumed purely per-message
+                // (every log sub has delivery rows — mixing with a cursor sub
+                // falls back to expiry-only cleanup), drop the oldest messages
+                // every per-message subscriber has acked. Inlined so it reuses
+                // `conn` — a second pooled connection would deadlock a size-1 pool.
+                let pm_budget = limit - ids.len() as i64;
+                if pm_budget > 0 {
+                    use diesel::dsl::count_star;
+                    use std::collections::HashSet;
+                    // Per-message subscribers per topic (those with delivery rows).
+                    let pm_pairs: Vec<(String, String)> = topic_deliveries::table
+                        .select((topic_deliveries::topic, topic_deliveries::subscription_name))
+                        .distinct()
+                        .load(&mut conn)?;
+                    let mut pm_subs: HashMap<String, HashSet<String>> = HashMap::new();
+                    for (topic, name) in pm_pairs {
+                        pm_subs.entry(topic).or_default().insert(name);
+                    }
+                    // Log subscribers per topic — skip a topic that mixes in a
+                    // cursor reader (a log sub with no delivery rows).
+                    let mut log_by_topic: HashMap<String, Vec<String>> = HashMap::new();
+                    if !pm_subs.is_empty() {
+                        let log_subs: Vec<(String, String)> = topic_subscriptions::table
+                            .filter(
+                                topic_subscriptions::mode
+                                    .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG),
+                            )
+                            .select((
+                                topic_subscriptions::topic,
+                                topic_subscriptions::subscription_name,
+                            ))
+                            .load(&mut conn)?;
+                        for (topic, name) in log_subs {
+                            log_by_topic.entry(topic).or_default().push(name);
+                        }
+                    }
+                    for (topic, names) in &pm_subs {
+                        let remaining = limit - ids.len() as i64;
+                        if remaining <= 0 {
+                            break;
+                        }
+                        if let Some(logs) = log_by_topic.get(topic) {
+                            if logs.iter().any(|n| !names.contains(n)) {
+                                continue;
+                            }
+                        }
+                        let sub_count = names.len() as i64;
+                        let candidates: Vec<String> = topic_messages::table
+                            .filter(topic_messages::topic.eq(topic))
+                            .order(topic_messages::id.asc())
+                            .select(topic_messages::id)
+                            .limit(remaining)
+                            .load(&mut conn)?;
+                        if candidates.is_empty() {
+                            continue;
+                        }
+                        let acked_counts: Vec<(String, i64)> = topic_deliveries::table
+                            .filter(topic_deliveries::topic.eq(topic))
+                            .filter(topic_deliveries::message_id.eq_any(&candidates))
+                            .filter(topic_deliveries::acked.eq(true))
+                            .group_by(topic_deliveries::message_id)
+                            .select((topic_deliveries::message_id, count_star()))
+                            .load(&mut conn)?;
+                        let counts: HashMap<String, i64> = acked_counts.into_iter().collect();
+                        for id in candidates {
+                            if counts.get(&id).copied().unwrap_or(0) == sub_count {
+                                ids.push(id);
+                            }
+                        }
+                    }
+                }
+
                 if ids.is_empty() {
                     return Ok(0);
                 }
@@ -376,6 +448,12 @@ macro_rules! impl_diesel_pubsub_ops {
                 let removed =
                     diesel::delete(topic_messages::table.filter(topic_messages::id.eq_any(&ids)))
                         .execute(&mut conn)?;
+                // Drop the delivery rows of the purged messages so per-message
+                // state can't outlive the log it tracks.
+                diesel::delete(
+                    topic_deliveries::table.filter(topic_deliveries::message_id.eq_any(&ids)),
+                )
+                .execute(&mut conn)?;
                 Ok(removed as u64)
             }
 
@@ -398,6 +476,134 @@ macro_rules! impl_diesel_pubsub_ops {
                     .load::<TopicRow>(&mut conn)?;
                 Ok(rows.into_iter().map(Into::into).collect())
             }
+
+            /// Lease up to `limit` available messages to `subscription_name` for
+            /// `visibility_ms` (see the `lease_topic_messages` trait method).
+            /// Requires a registered `log` subscription (like the cursor read).
+            /// One bounded anti-join finds the available messages, then a lease
+            /// row is upserted per message in one transaction. Delivery is
+            /// **at-least-once**: the availability window isn't row-locked, so two
+            /// concurrent leasers can briefly claim the same message — handlers
+            /// must be idempotent, the same guarantee as a lease timeout redeliver.
+            pub fn lease_topic_messages(
+                &self,
+                topic: &str,
+                subscription_name: &str,
+                limit: i64,
+                visibility_ms: i64,
+                now: i64,
+            ) -> Result<Vec<$crate::storage::records::TopicMessage>> {
+                if limit <= 0 {
+                    return Ok(Vec::new());
+                }
+                let mut conn = self.conn()?;
+                // Only a registered log subscription may lease — a fan-out or
+                // unknown name must never claim from the log (mirrors read/ack).
+                let is_log_sub = topic_subscriptions::table
+                    .filter(topic_subscriptions::topic.eq(topic))
+                    .filter(topic_subscriptions::subscription_name.eq(subscription_name))
+                    .filter(
+                        topic_subscriptions::mode
+                            .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG),
+                    )
+                    .select(topic_subscriptions::subscription_name)
+                    .first::<String>(&mut conn)
+                    .optional()?;
+                if is_log_sub.is_none() {
+                    return Ok(Vec::new());
+                }
+                conn.transaction(|conn| {
+                    // Available = no delivery row (never delivered), or a prior
+                    // lease that expired and was never acked.
+                    let rows: Vec<TopicMessageRow> = topic_messages::table
+                        .left_join(
+                            topic_deliveries::table.on(topic_deliveries::topic
+                                .eq(topic_messages::topic)
+                                .and(topic_deliveries::subscription_name.eq(subscription_name))
+                                .and(topic_deliveries::message_id.eq(topic_messages::id))),
+                        )
+                        .filter(topic_messages::topic.eq(topic))
+                        .filter(
+                            topic_deliveries::message_id
+                                .is_null()
+                                .or(topic_deliveries::acked
+                                    .eq(false)
+                                    .and(topic_deliveries::lease_expires_at.le(now))),
+                        )
+                        .order(topic_messages::id.asc())
+                        .limit(limit)
+                        .select(TopicMessageRow::as_select())
+                        .load(conn)?;
+
+                    let lease_until = now + visibility_ms;
+                    for row in &rows {
+                        let lease = NewTopicDeliveryRow {
+                            topic,
+                            subscription_name,
+                            message_id: &row.id,
+                            acked: false,
+                            attempts: 1,
+                            lease_expires_at: lease_until,
+                            delivered_at: now,
+                        };
+                        diesel::insert_into(topic_deliveries::table)
+                            .values(&lease)
+                            .on_conflict((
+                                topic_deliveries::topic,
+                                topic_deliveries::subscription_name,
+                                topic_deliveries::message_id,
+                            ))
+                            .do_update()
+                            .set((
+                                topic_deliveries::acked.eq(false),
+                                topic_deliveries::attempts.eq(topic_deliveries::attempts + 1),
+                                topic_deliveries::lease_expires_at.eq(lease_until),
+                                topic_deliveries::delivered_at.eq(now),
+                            ))
+                            .execute(conn)?;
+                    }
+                    Ok(rows.into_iter().map(Into::into).collect())
+                })
+            }
+
+            /// Ack one leased message — the delivery is done. Returns false when
+            /// there was no un-acked delivery row to ack.
+            pub fn ack_message(
+                &self,
+                topic: &str,
+                subscription_name: &str,
+                message_id: &str,
+            ) -> Result<bool> {
+                let mut conn = self.conn()?;
+                let affected = diesel::update(
+                    topic_deliveries::table
+                        .find((topic, subscription_name, message_id))
+                        .filter(topic_deliveries::acked.eq(false)),
+                )
+                .set(topic_deliveries::acked.eq(true))
+                .execute(&mut conn)?;
+                Ok(affected > 0)
+            }
+
+            /// Nack one leased message — make it immediately available for
+            /// redelivery (`lease_expires_at = 0`). Returns false when there was
+            /// no un-acked delivery row to nack.
+            pub fn nack_message(
+                &self,
+                topic: &str,
+                subscription_name: &str,
+                message_id: &str,
+            ) -> Result<bool> {
+                let mut conn = self.conn()?;
+                let affected = diesel::update(
+                    topic_deliveries::table
+                        .find((topic, subscription_name, message_id))
+                        .filter(topic_deliveries::acked.eq(false)),
+                )
+                .set(topic_deliveries::lease_expires_at.eq(0))
+                .execute(&mut conn)?;
+                Ok(affected > 0)
+            }
         }
     };
 }
diff --git a/crates/taskito-core/src/storage/mod.rs b/crates/taskito-core/src/storage/mod.rs
index 21fea11c..d67084a8 100644
--- a/crates/taskito-core/src/storage/mod.rs
+++ b/crates/taskito-core/src/storage/mod.rs
@@ -747,6 +747,32 @@ macro_rules! impl_storage {
             ) -> $crate::error::Result<Vec<$crate::storage::records::Topic>> {
                 self.list_declared_topics()
             }
+            fn lease_topic_messages(
+                &self,
+                topic: &str,
+                subscription_name: &str,
+                limit: i64,
+                visibility_ms: i64,
+                now: i64,
+            ) -> $crate::error::Result<Vec<$crate::storage::records::TopicMessage>> {
+                self.lease_topic_messages(topic, subscription_name, limit, visibility_ms, now)
+            }
+            fn ack_message(
+                &self,
+                topic: &str,
+                subscription_name: &str,
+                message_id: &str,
+            ) -> $crate::error::Result<bool> {
+                self.ack_message(topic, subscription_name, message_id)
+            }
+            fn nack_message(
+                &self,
+                topic: &str,
+                subscription_name: &str,
+                message_id: &str,
+            ) -> $crate::error::Result<bool> {
+                self.nack_message(topic, subscription_name, message_id)
+            }
             fn record_metric(
                 &self,
                 task_name: &str,
@@ -1459,6 +1485,30 @@ impl Storage for StorageBackend {
     fn list_declared_topics(&self) -> Result<Vec<records::Topic>> {
         delegate!(self, list_declared_topics)
     }
+    fn lease_topic_messages(
+        &self,
+        topic: &str,
+        subscription_name: &str,
+        limit: i64,
+        visibility_ms: i64,
+        now: i64,
+    ) -> Result<Vec<records::TopicMessage>> {
+        delegate!(
+            self,
+            lease_topic_messages,
+            topic,
+            subscription_name,
+            limit,
+            visibility_ms,
+            now
+        )
+    }
+    fn ack_message(&self, topic: &str, subscription_name: &str, message_id: &str) -> Result<bool> {
+        delegate!(self, ack_message, topic, subscription_name, message_id)
+    }
+    fn nack_message(&self, topic: &str, subscription_name: &str, message_id: &str) -> Result<bool> {
+        delegate!(self, nack_message, topic, subscription_name, message_id)
+    }
     fn record_metric(
         &self,
         task_name: &str,
diff --git a/crates/taskito-core/src/storage/models.rs b/crates/taskito-core/src/storage/models.rs
index 50755b69..61e107c8 100644
--- a/crates/taskito-core/src/storage/models.rs
+++ b/crates/taskito-core/src/storage/models.rs
@@ -8,7 +8,8 @@ use super::records::{
 use super::schema::{
     archived_jobs, circuit_breakers, dashboard_settings, dead_letter, distributed_locks,
     execution_claims, job_dependencies, job_errors, jobs, periodic_tasks, queue_state, rate_limits,
-    replay_history, task_logs, task_metrics, topic_messages, topic_subscriptions, topics, workers,
+    replay_history, task_logs, task_metrics, topic_deliveries, topic_messages, topic_subscriptions,
+    topics, workers,
 };
 
 /// A row in the `jobs` table (for SELECT queries).
@@ -526,6 +527,23 @@ pub struct NewTopicRow<'a> {
     pub created_at: i64,
 }
 
+// ── Topic Deliveries (per-message ack) ──────────────────────────
+
+/// Per-`(subscription, message)` delivery state for a per-message log consumer.
+/// `lease_expires_at` bounds an in-flight lease (0 = available for redelivery);
+/// `acked` ends the delivery; `attempts` counts (re)deliveries.
+#[derive(Insertable, Debug)]
+#[diesel(table_name = topic_deliveries)]
+pub struct NewTopicDeliveryRow<'a> {
+    pub topic: &'a str,
+    pub subscription_name: &'a str,
+    pub message_id: &'a str,
+    pub acked: bool,
+    pub attempts: i32,
+    pub lease_expires_at: i64,
+    pub delivered_at: i64,
+}
+
 // ── Distributed Locks ───────────────────────────────────────────
 
 #[derive(Queryable, Selectable, QueryableByName, Debug, Clone)]
diff --git a/crates/taskito-core/src/storage/postgres/pubsub.rs b/crates/taskito-core/src/storage/postgres/pubsub.rs
index 79e19000..a000b30d 100644
--- a/crates/taskito-core/src/storage/postgres/pubsub.rs
+++ b/crates/taskito-core/src/storage/postgres/pubsub.rs
@@ -2,7 +2,7 @@ use diesel::prelude::*;
 
 use super::super::models::*;
 use super::super::records::NewSubscription;
-use super::super::schema::{topic_messages, topic_subscriptions, topics};
+use super::super::schema::{topic_deliveries, topic_messages, topic_subscriptions, topics};
 use super::PostgresStorage;
 use crate::error::Result;
 
diff --git a/crates/taskito-core/src/storage/redis_backend/pubsub.rs b/crates/taskito-core/src/storage/redis_backend/pubsub.rs
index 5b6cbf62..b94e1d97 100644
--- a/crates/taskito-core/src/storage/redis_backend/pubsub.rs
+++ b/crates/taskito-core/src/storage/redis_backend/pubsub.rs
@@ -710,38 +710,47 @@ impl RedisStorage {
 
     /// Compact each log topic: `XTRIM MINID` to the min cursor across its subs,
     /// then `XDEL` any entry past `expires_at` (the TTL safety net, so a NULL or
-    /// stalled cursor can't block reclamation) — matching the Diesel backend.
-    /// The expiry scan is bounded by `limit` per topic.
+    /// stalled cursor can't block reclamation), then — on a topic consumed purely
+    /// per-message — `XDEL` any entry every per-message subscriber has acked.
+    /// Matches the Diesel backend. The scans are bounded by `limit` per topic.
     pub fn purge_topic_messages(&self, now: i64, limit: i64) -> Result<u64> {
         use std::collections::HashMap;
-        let subs = self.list_subscriptions()?;
-        let mut floors: HashMap<String, Option<String>> = HashMap::new();
-        for sub in subs.into_iter().filter(|s| s.mode == SUBSCRIPTION_MODE_LOG) {
-            match floors.entry(sub.topic) {
-                std::collections::hash_map::Entry::Vacant(slot) => {
-                    slot.insert(sub.cursor);
-                }
-                std::collections::hash_map::Entry::Occupied(mut slot) => {
-                    let merged = match (slot.get().clone(), sub.cursor) {
-                        (Some(current), Some(cursor)) => Some(
-                            if Self::stream_id_parts(&cursor) < Self::stream_id_parts(&current) {
-                                cursor
-                            } else {
-                                current
-                            },
-                        ),
-                        _ => None,
-                    };
-                    slot.insert(merged);
-                }
-            }
+        // Group each topic's log subs so cursor floor and per-message compaction
+        // both see the full set (a mixed cursor+per-message topic must be spotted).
+        let mut by_topic: HashMap<String, Vec<Subscription>> = HashMap::new();
+        for sub in self
+            .list_subscriptions()?
+            .into_iter()
+            .filter(|s| s.mode == SUBSCRIPTION_MODE_LOG)
+        {
+            by_topic.entry(sub.topic.clone()).or_default().push(sub);
         }
 
         let mut conn = self.conn()?;
         let mut removed = 0u64;
-        for (topic, floor) in floors {
-            let stream_key = self.topic_stream_key(&topic);
-            if let Some(floor) = floor {
+        for (topic, subs) in &by_topic {
+            let stream_key = self.topic_stream_key(topic);
+
+            // Min cursor across the topic's subs; `None` once any read nothing.
+            let mut floor: Option<String> = None;
+            let mut floor_poisoned = false;
+            for sub in subs {
+                match &sub.cursor {
+                    None => {
+                        floor_poisoned = true;
+                        break;
+                    }
+                    Some(cursor) => {
+                        let take = floor.as_ref().is_none_or(|cur| {
+                            Self::stream_id_parts(cursor) < Self::stream_id_parts(cur)
+                        });
+                        if take {
+                            floor = Some(cursor.clone());
+                        }
+                    }
+                }
+            }
+            if let Some(floor) = floor.filter(|_| !floor_poisoned) {
                 let trimmed: u64 = redis::cmd("XTRIM")
                     .arg(&stream_key)
                     .arg("MINID")
@@ -751,10 +760,69 @@ impl RedisStorage {
                 removed += trimmed;
             }
             removed += self.xdel_expired(&mut conn, &stream_key, now, limit)?;
+            removed += self.xdel_per_message_acked(&mut conn, topic, subs, limit)?;
         }
         Ok(removed)
     }
 
+    /// On a topic consumed purely per-message (every log sub has a delivery hash),
+    /// `XDEL` up to `limit` entries every per-message subscriber has acked, and
+    /// drop their delivery fields so per-message state can't outlive the log. A
+    /// topic mixing in a cursor reader (a log sub with no deliveries) is left to
+    /// the cursor/expiry paths, matching the Diesel backend.
+    fn xdel_per_message_acked(
+        &self,
+        conn: &mut redis::Connection,
+        topic: &str,
+        subs: &[Subscription],
+        limit: i64,
+    ) -> Result<u64> {
+        use std::collections::HashMap;
+        // Delivery state per sub; a sub with an empty hash is a cursor reader.
+        let mut deliveries: Vec<(String, HashMap<String, String>)> = Vec::with_capacity(subs.len());
+        for sub in subs {
+            let key = self.pm_deliveries_key(topic, &sub.subscription_name);
+            let map: HashMap<String, String> = conn.hgetall(&key).map_err(map_err)?;
+            deliveries.push((key, map));
+        }
+        // Purely per-message only: skip if any log sub has no delivery hash.
+        if deliveries.iter().any(|(_, map)| map.is_empty()) {
+            return Ok(0);
+        }
+
+        let reply: StreamRangeReply = redis::cmd("XRANGE")
+            .arg(self.topic_stream_key(topic))
+            .arg("-")
+            .arg("+")
+            .arg("COUNT")
+            .arg(limit)
+            .query(conn)
+            .map_err(map_err)?;
+        let acked: Vec<String> = reply
+            .ids
+            .into_iter()
+            .filter(|entry| {
+                deliveries
+                    .iter()
+                    .all(|(_, map)| map.get(&entry.id).is_some_and(|s| s == "acked"))
+            })
+            .map(|entry| entry.id)
+            .collect();
+        if acked.is_empty() {
+            return Ok(0);
+        }
+
+        let deleted: u64 = redis::cmd("XDEL")
+            .arg(self.topic_stream_key(topic))
+            .arg(&acked)
+            .query(conn)
+            .map_err(map_err)?;
+        for (key, _) in &deliveries {
+            conn.hdel::<_, _, ()>(key, &acked).map_err(map_err)?;
+        }
+        Ok(deleted)
+    }
+
     /// `XDEL` up to `limit` entries in `stream_key` whose `expires_at` field is
     /// at or before `now`, regardless of any subscription cursor.
     fn xdel_expired(
@@ -834,4 +902,135 @@ impl RedisStorage {
             .map(|(name, json)| Ok(serde_json::from_str::<TopicEntry>(&json)?.into_topic(name)))
             .collect()
     }
+
+    /// Per-message delivery state for a subscription: a hash at
+    /// `pmdeliv:<topic>:<subscription>`, field = message id, value = `"acked"`
+    /// or the lease's `expires_at` (`0` = available for redelivery, e.g. nacked).
+    /// This mirrors the Diesel `topic_deliveries` table so the semantics match.
+    fn pm_deliveries_key(&self, topic: &str, subscription_name: &str) -> String {
+        self.key(&["pmdeliv", topic, subscription_name])
+    }
+
+    /// Lease up to `limit` available messages to `subscription_name` for
+    /// `visibility_ms`. Available = never delivered, or a lease that expired /
+    /// was nacked and never acked. See the `lease_topic_messages` trait method.
+    /// Requires a registered `log` subscription (like the cursor read). The
+    /// stream is scanned in `COUNT`-bounded pages (not loaded whole) until
+    /// `limit` are collected or it ends. Delivery is **at-least-once**: the
+    /// `HGETALL` availability snapshot and the per-message `HSET` lease aren't a
+    /// single atomic step, so two concurrent leasers can briefly claim the same
+    /// message — handlers must be idempotent, as with a lease-timeout redeliver.
+    pub fn lease_topic_messages(
+        &self,
+        topic: &str,
+        subscription_name: &str,
+        limit: i64,
+        visibility_ms: i64,
+        now: i64,
+    ) -> Result<Vec<TopicMessage>> {
+        if limit <= 0 {
+            return Ok(Vec::new());
+        }
+        let mut conn = self.conn()?;
+        // Only a registered log subscription may lease — a fan-out or unknown
+        // name must never claim from the log (mirrors read/ack).
+        let blob_key = self.key(&["sub", topic, subscription_name]);
+        let Some(data): Option<String> = conn.get(&blob_key).map_err(map_err)? else {
+            return Ok(Vec::new());
+        };
+        if serde_json::from_str::<SubEntry>(&data)?.mode != SUBSCRIPTION_MODE_LOG {
+            return Ok(Vec::new());
+        }
+
+        let deliv_key = self.pm_deliveries_key(topic, subscription_name);
+        let deliveries: std::collections::HashMap<String, String> =
+            conn.hgetall(&deliv_key).map_err(map_err)?;
+        let stream_key = self.topic_stream_key(topic);
+
+        let lease_until = now + visibility_ms;
+        let mut out = Vec::new();
+        let mut start = "-".to_string();
+        while (out.len() as i64) < limit {
+            let page: StreamRangeReply = redis::cmd("XRANGE")
+                .arg(&stream_key)
+                .arg(&start)
+                .arg("+")
+                .arg("COUNT")
+                .arg(limit)
+                .query(&mut conn)
+                .map_err(map_err)?;
+            if page.ids.is_empty() {
+                break;
+            }
+            let exhausted = (page.ids.len() as i64) < limit;
+            // Resume the next page exclusively after this page's last id.
+            let next_start = page.ids.last().map(|entry| format!("({}", entry.id));
+            for entry in page.ids {
+                if (out.len() as i64) >= limit {
+                    break;
+                }
+                let available = match deliveries.get(&entry.id) {
+                    None => true,
+                    Some(state) if state == "acked" => false,
+                    // A leased entry: available once its lease expired (nack = 0).
+                    Some(state) => state.parse::<i64>().map(|exp| exp <= now).unwrap_or(true),
+                };
+                if !available {
+                    continue;
+                }
+                conn.hset::<_, _, _, ()>(&deliv_key, &entry.id, lease_until)
+                    .map_err(map_err)?;
+                out.push(Self::stream_id_to_message(topic, entry));
+            }
+            match next_start {
+                Some(next) if !exhausted => start = next,
+                _ => break,
+            }
+        }
+        Ok(out)
+    }
+
+    /// Ack one leased message (delivery done). Returns false when there was no
+    /// un-acked delivery to ack.
+    pub fn ack_message(
+        &self,
+        topic: &str,
+        subscription_name: &str,
+        message_id: &str,
+    ) -> Result<bool> {
+        self.set_delivery_state(topic, subscription_name, message_id, "acked")
+    }
+
+    /// Nack one leased message — make it immediately available (`0`). Returns
+    /// false when there was no un-acked delivery to nack.
+    pub fn nack_message(
+        &self,
+        topic: &str,
+        subscription_name: &str,
+        message_id: &str,
+    ) -> Result<bool> {
+        self.set_delivery_state(topic, subscription_name, message_id, "0")
+    }
+
+    /// Set a delivery's state, but only when it exists and is not already acked
+    /// (ack/nack are no-ops on an unknown or already-acked delivery).
+    fn set_delivery_state(
+        &self,
+        topic: &str,
+        subscription_name: &str,
+        message_id: &str,
+        state: &str,
+    ) -> Result<bool> {
+        let mut conn = self.conn()?;
+        let deliv_key = self.pm_deliveries_key(topic, subscription_name);
+        let current: Option<String> = conn.hget(&deliv_key, message_id).map_err(map_err)?;
+        match current {
+            Some(v) if v != "acked" => {
+                conn.hset::<_, _, _, ()>(&deliv_key, message_id, state)
+                    .map_err(map_err)?;
+                Ok(true)
+            }
+            _ => Ok(false),
+        }
+    }
 }
diff --git a/crates/taskito-core/src/storage/schema.rs b/crates/taskito-core/src/storage/schema.rs
index 739efb31..faeb5341 100644
--- a/crates/taskito-core/src/storage/schema.rs
+++ b/crates/taskito-core/src/storage/schema.rs
@@ -268,4 +268,17 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    topic_deliveries (topic, subscription_name, message_id) {
+        topic -> Text,
+        subscription_name -> Text,
+        message_id -> Text,
+        acked -> Bool,
+        attempts -> Integer,
+        lease_expires_at -> BigInt,
+        delivered_at -> BigInt,
+    }
+}
+
 diesel::allow_tables_to_appear_in_same_query!(jobs, job_dependencies);
+diesel::allow_tables_to_appear_in_same_query!(topic_messages, topic_deliveries);
diff --git a/crates/taskito-core/src/storage/sqlite/pubsub.rs b/crates/taskito-core/src/storage/sqlite/pubsub.rs
index 54ba422e..cedf2748 100644
--- a/crates/taskito-core/src/storage/sqlite/pubsub.rs
+++ b/crates/taskito-core/src/storage/sqlite/pubsub.rs
@@ -2,7 +2,7 @@ use diesel::prelude::*;
 
 use super::super::models::*;
 use super::super::records::NewSubscription;
-use super::super::schema::{topic_messages, topic_subscriptions, topics};
+use super::super::schema::{topic_deliveries, topic_messages, topic_subscriptions, topics};
 use super::SqliteStorage;
 use crate::error::Result;
 
diff --git a/crates/taskito-core/src/storage/traits.rs b/crates/taskito-core/src/storage/traits.rs
index 93bd08dc..68cf45cd 100644
--- a/crates/taskito-core/src/storage/traits.rs
+++ b/crates/taskito-core/src/storage/traits.rs
@@ -301,6 +301,30 @@ pub trait Storage: Send + Sync + Clone {
     /// List every declared topic in the registry.
     fn list_declared_topics(&self) -> Result<Vec<Topic>>;
 
+    // ── Per-message ack (opt-in, log topics) ────────────────────────
+
+    /// Lease up to `limit` available messages of a log topic to `subscription_name`
+    /// for `visibility_ms`, oldest first. "Available" = never delivered, or a
+    /// prior lease that expired (`lease_expires_at <= now`) and was never acked.
+    /// Each leased message's delivery row is upserted (lease extended, `attempts`
+    /// bumped). Unlike the cursor read this tracks per-message state, so a nacked
+    /// or timed-out message is redelivered without blocking its siblings.
+    fn lease_topic_messages(
+        &self,
+        topic: &str,
+        subscription_name: &str,
+        limit: i64,
+        visibility_ms: i64,
+        now: i64,
+    ) -> Result<Vec<TopicMessage>>;
+    /// Ack one leased message — the delivery is done and never redelivered.
+    /// Returns false if there was no un-acked delivery to ack.
+    fn ack_message(&self, topic: &str, subscription_name: &str, message_id: &str) -> Result<bool>;
+    /// Negative-ack one leased message — makes it immediately available for
+    /// redelivery (vs waiting for the visibility timeout). Returns false if there
+    /// was no un-acked delivery to nack.
+    fn nack_message(&self, topic: &str, subscription_name: &str, message_id: &str) -> Result<bool>;
+
     // ── Metrics operations ──────────────────────────────────────────
 
     /// Record one execution measurement for a task.
diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs
index 6e8a5562..33e8d7ea 100644
--- a/crates/taskito-core/tests/rust/storage_tests.rs
+++ b/crates/taskito-core/tests/rust/storage_tests.rs
@@ -1427,6 +1427,84 @@ fn test_topic_log_purge(s: &impl Storage) {
     );
 }
 
+fn payloads(msgs: &[taskito_core::storage::records::TopicMessage]) -> Vec<Vec<u8>> {
+    msgs.iter().map(|m| m.payload.clone()).collect()
+}
+
+fn test_per_message_ack(s: &impl Storage) {
+    let topic = "pm-ack";
+    s.register_subscription(&log_sub(topic, "w")).unwrap();
+    let now = now_millis();
+    let vis = 60_000;
+    let m0 = s.publish_message(topic, b"m0", None, None, None).unwrap();
+    let m1 = s.publish_message(topic, b"m1", None, None, None).unwrap();
+    let _m2 = s.publish_message(topic, b"m2", None, None, None).unwrap();
+
+    // Lease 2 (m0, m1). A second lease within the window returns only m2 — the
+    // in-flight ones are not re-leased.
+    assert_eq!(
+        payloads(&s.lease_topic_messages(topic, "w", 2, vis, now).unwrap()),
+        vec![b"m0".to_vec(), b"m1".to_vec()]
+    );
+    assert_eq!(
+        payloads(&s.lease_topic_messages(topic, "w", 10, vis, now).unwrap()),
+        vec![b"m2".to_vec()]
+    );
+
+    // Ack m0 (done forever); nack m1 (available now). Acking m0 again is a no-op.
+    assert!(s.ack_message(topic, "w", &m0.id).unwrap());
+    assert!(s.nack_message(topic, "w", &m1.id).unwrap());
+    assert!(!s.ack_message(topic, "w", &m0.id).unwrap());
+
+    // Within the window: only the nacked m1 comes back (m0 acked, m2 in-flight).
+    assert_eq!(
+        payloads(&s.lease_topic_messages(topic, "w", 10, vis, now).unwrap()),
+        vec![b"m1".to_vec()]
+    );
+
+    // After the visibility timeout: every un-acked lease (m1, m2) is redelivered
+    // oldest-first; the acked m0 never returns.
+    let later = now + vis + 1;
+    let redelivered = s.lease_topic_messages(topic, "w", 10, vis, later).unwrap();
+    assert_eq!(payloads(&redelivered), vec![b"m1".to_vec(), b"m2".to_vec()]);
+
+    // Drain the topic so its acked deliveries don't get compacted by a later
+    // test's (globally-scanning) purge.
+    for m in &redelivered {
+        assert!(s.ack_message(topic, "w", &m.id).unwrap());
+    }
+    s.purge_topic_messages(later, 100).unwrap();
+    s.unsubscribe(topic, "w").unwrap();
+}
+
+fn test_per_message_purge(s: &impl Storage) {
+    let topic = "pm-purge";
+    s.register_subscription(&log_sub(topic, "w")).unwrap();
+    let now = now_millis();
+    let vis = 60_000;
+    let m0 = s.publish_message(topic, b"m0", None, None, None).unwrap();
+    let m1 = s.publish_message(topic, b"m1", None, None, None).unwrap();
+
+    // Lease both; ack only m0. A purge compacts the message every per-message
+    // subscriber acked (m0); the un-acked m1 survives.
+    s.lease_topic_messages(topic, "w", 10, vis, now).unwrap();
+    assert!(s.ack_message(topic, "w", &m0.id).unwrap());
+    assert_eq!(s.purge_topic_messages(now, 100).unwrap(), 1);
+
+    // m0 is gone (delivery row too); past the timeout only m1 redelivers.
+    let later = now + vis + 1;
+    assert_eq!(
+        payloads(&s.lease_topic_messages(topic, "w", 10, vis, later).unwrap()),
+        vec![b"m1".to_vec()]
+    );
+
+    // Acking m1 lets the next purge drain the topic.
+    assert!(s.ack_message(topic, "w", &m1.id).unwrap());
+    assert_eq!(s.purge_topic_messages(later, 100).unwrap(), 1);
+
+    s.unsubscribe(topic, "w").unwrap();
+}
+
 fn test_enqueue_unique_batch(s: &impl Storage) {
     let q = "q-eub";
     let keyed = |uk: &str| {
@@ -1518,6 +1596,8 @@ fn run_storage_tests(s: &impl Storage) {
     test_topic_log_messages(s);
     test_topic_log_purge(s);
     test_topic_registry(s);
+    test_per_message_ack(s);
+    test_per_message_purge(s);
     test_circuit_breakers(s);
     test_execution_claims_purge(s);
     test_reap_stale_jobs(s);
diff --git a/crates/taskito-java/src/queue/pubsub.rs b/crates/taskito-java/src/queue/pubsub.rs
index 5c7e76c8..21980a4d 100644
--- a/crates/taskito-java/src/queue/pubsub.rs
+++ b/crates/taskito-java/src/queue/pubsub.rs
@@ -239,6 +239,87 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_ackTopicCu
     })
 }
 
+/// `String leaseTopicMessages(long handle, String topic, String subscriptionName,
+/// long limit, long visibilityMs)` — lease up to `limit` available messages for
+/// `visibilityMs`, tracking per-message state, as a JSON array of `TopicMessageView`.
+/// A nack or an expired lease redelivers just that message. `now` is taken here so
+/// the SDK never passes a clock.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_leaseTopicMessages<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    topic: JString<'local>,
+    subscription_name: JString<'local>,
+    limit: jlong,
+    visibility_ms: jlong,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let topic = read_string(env, &topic)?;
+        let subscription_name = read_string(env, &subscription_name)?;
+        let messages = queue.storage.lease_topic_messages(
+            &topic,
+            &subscription_name,
+            limit,
+            visibility_ms,
+            now_millis(),
+        )?;
+        let views: Vec<TopicMessageView> = messages.iter().map(TopicMessageView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}
+
+/// `boolean ackMessage(long handle, String topic, String subscriptionName,
+/// String messageId)` — ack one leased message; it is done and never redelivered.
+/// Returns false when there was no un-acked delivery to ack.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_ackMessage<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    topic: JString<'local>,
+    subscription_name: JString<'local>,
+    message_id: JString<'local>,
+) -> jboolean {
+    guard(&mut env, JNI_FALSE, |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let topic = read_string(env, &topic)?;
+        let subscription_name = read_string(env, &subscription_name)?;
+        let message_id = read_string(env, &message_id)?;
+        Ok(to_jboolean(queue.storage.ack_message(
+            &topic,
+            &subscription_name,
+            &message_id,
+        )?))
+    })
+}
+
+/// `boolean nackMessage(long handle, String topic, String subscriptionName,
+/// String messageId)` — nack one leased message; make it available for redelivery
+/// now. Returns false when there was no un-acked delivery to nack.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_nackMessage<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    topic: JString<'local>,
+    subscription_name: JString<'local>,
+    message_id: JString<'local>,
+) -> jboolean {
+    guard(&mut env, JNI_FALSE, |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let topic = read_string(env, &topic)?;
+        let subscription_name = read_string(env, &subscription_name)?;
+        let message_id = read_string(env, &message_id)?;
+        Ok(to_jboolean(queue.storage.nack_message(
+            &topic,
+            &subscription_name,
+            &message_id,
+        )?))
+    })
+}
+
 /// `String topicLogStats(long handle)` — a JSON array of `TopicLogStatsView`, one
 /// lag snapshot per log subscription.
 #[no_mangle]
diff --git a/crates/taskito-node/src/queue/pubsub.rs b/crates/taskito-node/src/queue/pubsub.rs
index a0eab5c6..f0646700 100644
--- a/crates/taskito-node/src/queue/pubsub.rs
+++ b/crates/taskito-node/src/queue/pubsub.rs
@@ -109,6 +109,70 @@ impl JsQueue {
         .map_err(join_to_napi_err)?
     }
 
+    /// Lease up to `limit` available messages for `visibility_ms`, tracking
+    /// per-message state so a nack or lease timeout redelivers just that message.
+    /// `now` is computed here so the SDK never passes a clock.
+    #[napi]
+    pub async fn lease_topic_messages(
+        &self,
+        topic: String,
+        subscription_name: String,
+        limit: i64,
+        visibility_ms: i64,
+    ) -> Result<Vec<JsTopicMessage>> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            let messages = storage
+                .lease_topic_messages(
+                    &topic,
+                    &subscription_name,
+                    limit,
+                    visibility_ms,
+                    now_millis(),
+                )
+                .map_err(to_napi_err)?;
+            Ok(messages.into_iter().map(topic_message_to_js).collect())
+        })
+        .await
+        .map_err(join_to_napi_err)?
+    }
+
+    /// Ack one leased message — done, never redelivered. False if nothing to ack.
+    #[napi]
+    pub async fn ack_message(
+        &self,
+        topic: String,
+        subscription_name: String,
+        message_id: String,
+    ) -> Result<bool> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            storage
+                .ack_message(&topic, &subscription_name, &message_id)
+                .map_err(to_napi_err)
+        })
+        .await
+        .map_err(join_to_napi_err)?
+    }
+
+    /// Nack one leased message — redeliver it now. False if nothing to nack.
+    #[napi]
+    pub async fn nack_message(
+        &self,
+        topic: String,
+        subscription_name: String,
+        message_id: String,
+    ) -> Result<bool> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            storage
+                .nack_message(&topic, &subscription_name, &message_id)
+                .map_err(to_napi_err)
+        })
+        .await
+        .map_err(join_to_napi_err)?
+    }
+
     /// Lag snapshot per log subscription.
     #[napi]
     pub async fn topic_log_stats(&self) -> Result<Vec<JsTopicLogStat>> {
diff --git a/crates/taskito-python/src/py_queue/pubsub.rs b/crates/taskito-python/src/py_queue/pubsub.rs
index ed3600e2..10b5de37 100644
--- a/crates/taskito-python/src/py_queue/pubsub.rs
+++ b/crates/taskito-python/src/py_queue/pubsub.rs
@@ -187,6 +187,60 @@ impl PyQueue {
             .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
     }
 
+    /// Lease up to `limit` available messages for `visibility_ms`, tracking
+    /// per-message state so a nack or lease timeout redelivers just that message.
+    /// Returns `(id, payload, metadata, notes, created_at)` tuples.
+    pub fn lease_topic_messages(
+        &self,
+        py: Python<'_>,
+        topic: &str,
+        subscription_name: &str,
+        limit: i64,
+        visibility_ms: i64,
+    ) -> PyResult<Vec<TopicMessageTuple>> {
+        let storage = &self.storage;
+        Ok(py
+            .detach(|| {
+                storage.lease_topic_messages(
+                    topic,
+                    subscription_name,
+                    limit,
+                    visibility_ms,
+                    now_millis(),
+                )
+            })
+            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?
+            .into_iter()
+            .map(|m| (m.id, m.payload, m.metadata, m.notes, m.created_at))
+            .collect())
+    }
+
+    /// Ack one leased message — done, never redelivered. False if nothing to ack.
+    pub fn ack_message(
+        &self,
+        py: Python<'_>,
+        topic: &str,
+        subscription_name: &str,
+        message_id: &str,
+    ) -> PyResult<bool> {
+        let storage = &self.storage;
+        py.detach(|| storage.ack_message(topic, subscription_name, message_id))
+            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
+    }
+
+    /// Nack one leased message — redeliver it now. False if nothing to nack.
+    pub fn nack_message(
+        &self,
+        py: Python<'_>,
+        topic: &str,
+        subscription_name: &str,
+        message_id: &str,
+    ) -> PyResult<bool> {
+        let storage = &self.storage;
+        py.detach(|| storage.nack_message(topic, subscription_name, message_id))
+            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
+    }
+
     /// Per-log-subscription lag snapshot:
     /// `(topic, subscription, cursor, lag, oldest_unacked_age_ms)`.
     pub fn topic_log_stats(&self, py: Python<'_>) -> PyResult<Vec<TopicLogStatsTuple>> {
diff --git a/docs/content/docs/java/api-reference/queue/pubsub.mdx b/docs/content/docs/java/api-reference/queue/pubsub.mdx
index 1174cd28..c2652181 100644
--- a/docs/content/docs/java/api-reference/queue/pubsub.mdx
+++ b/docs/content/docs/java/api-reference/queue/pubsub.mdx
@@ -20,6 +20,8 @@ semantics, lifecycle, cross-SDK topics).
 | `listDeclaredTopics()` → `List<Topic>` | List declared topics: `name`, `mode`, `retentionMs`, `createdAt`. |
 | `readTopic(topic, name)` / `readTopic(topic, name, limit)` → `List<TopicMessage>` | Pull up to `limit` (default 100) messages after `name`'s cursor, oldest first and exclusive of it. Empty once caught up. At-least-once — process, then `ackTopic`. |
 | `ackTopic(topic, name, cursor)` | Advance a log subscription's cursor to `cursor` (a message `id`) — a monotonic high-water mark. `false` if nothing moved. |
+| `leaseTopic(topic, name)` / `leaseTopic(topic, name, limit, visibility)` → `List<TopicMessage>` | **Per-message** alternative to the cursor read: lease up to `limit` (default 100) messages for `visibility` (a `Duration`, default 30s), tracked individually so a nack or lease timeout redelivers just that message. Don't mix with `readTopic`/`ackTopic` on one subscription. |
+| `ackMessage(topic, name, messageId)` / `nackMessage(topic, name, messageId)` → `boolean` | Ack (done, never redelivered) or nack (redeliver now) one leased message. `false` if there was no un-acked delivery. |
 | `topicLogStats()` → `List<TopicLogStat>` | Lag snapshot for every log subscription: `topic`, `subscription`, `cursor`, `lag`, `oldestUnackedAgeMs`. |
 
 See [Log topics](/java/guides/core/pubsub#log-topics) for the cursor,
@@ -27,11 +29,11 @@ at-least-once, and retention semantics.
 
 ## `TopicMessage`
 
-One message pulled from a log topic, returned by `readTopic`.
+One message pulled from a log topic, returned by `readTopic` and `leaseTopic`.
 
 | Field | Type | Description |
 |---|---|---|
-| `id` | `String` | Cursor token — pass to `ackTopic`. |
+| `id` | `String` | Message id — pass to `ackTopic` (cursor read) or to `ackMessage`/`nackMessage` (per-message lease). |
 | `payload` | `byte[]` | Opaque publish payload. Unlike Python/Node, this is **not** decoded for you — Java is statically typed and there's no target type to infer it into, so decode it yourself with the same serializer the publisher used. |
 | `metadata` | `Map<String, Object>` | Caller metadata, decoded, or `null` if none was set. |
 | `notes` | `Map<String, Object>` | Structured notes, decoded, or `null` if none were set. |
diff --git a/docs/content/docs/node/api-reference/queue/pubsub.mdx b/docs/content/docs/node/api-reference/queue/pubsub.mdx
index 6b277d34..7b89f410 100644
--- a/docs/content/docs/node/api-reference/queue/pubsub.mdx
+++ b/docs/content/docs/node/api-reference/queue/pubsub.mdx
@@ -22,6 +22,8 @@ semantics, lifecycle, cross-SDK topics).
 | `listDeclaredTopics()` → `Promise<DeclaredTopic[]>` | List declared topics: `name`, `mode`, `retentionMs`, `createdAt`. |
 | `readTopic(topic, name, limit?)` → `Promise<TopicMessage[]>` | Pull up to `limit` (default 100) messages after `name`'s cursor, oldest first and exclusive of it. Empty once caught up. At-least-once — process, then `ackTopic()`. |
 | `ackTopic(topic, name, cursor)` → `Promise<boolean>` | Advance a log subscription's cursor to `cursor` (a message `id`) — a monotonic high-water mark. `false` if nothing moved. |
+| `leaseTopic(topic, name, opts?)` → `Promise<TopicMessage[]>` | **Per-message** alternative to the cursor read: lease up to `opts.limit` (default 100) messages for `opts.visibility` seconds (default 30), tracked individually so a nack or lease timeout redelivers just that message. Don't mix with `readTopic`/`ackTopic` on one subscription. |
+| `ackMessage(topic, name, messageId)` / `nackMessage(topic, name, messageId)` → `Promise<boolean>` | Ack (done, never redelivered) or nack (redeliver now) one leased message. `false` if there was no un-acked delivery. |
 | `topicLogStats()` → `Promise<TopicLogStat[]>` | Lag snapshot for every log subscription: `topic`, `subscription`, `cursor`, `lag`, `oldestUnackedAgeMs`. |
 
 See [Log topics](/node/guides/core/pubsub#log-topics) for the cursor,
@@ -29,11 +31,11 @@ at-least-once, and retention semantics.
 
 ## `TopicMessage`
 
-One message pulled from a log topic, returned by `readTopic()`.
+One message pulled from a log topic, returned by `readTopic()` and `leaseTopic()`.
 
 | Field | Type | Description |
 |---|---|---|
-| `id` | `string` | Cursor token — pass to `ackTopic()`. |
+| `id` | `string` | Message id — pass to `ackTopic()` (cursor read) or to `ackMessage()`/`nackMessage()` (per-message lease). |
 | `args` | `unknown[]` | Deserialized positional args from the `publish()` call. |
 | `metadata` | `Record<string, unknown>` | Caller metadata, if any. |
 | `notes` | `Record<string, unknown>` | Structured notes, if any. |
diff --git a/docs/content/docs/python/api-reference/queue/pubsub.mdx b/docs/content/docs/python/api-reference/queue/pubsub.mdx
index df1ed342..0ceb768f 100644
--- a/docs/content/docs/python/api-reference/queue/pubsub.mdx
+++ b/docs/content/docs/python/api-reference/queue/pubsub.mdx
@@ -173,6 +173,32 @@ high-water mark — acking an id acks everything up to and including it.
 Monotonic: acking an id older than what's already acked is a no-op and
 returns `False`.
 
+### `queue.lease_topic()`
+
+```python
+queue.lease_topic(
+    topic: str, name: str, limit: int = 100, visibility: float = 30.0
+) -> list[TopicMessage]
+```
+
+**Per-message** alternative to the cursor read. Leases up to `limit`
+available messages for `visibility` seconds and tracks each individually:
+:meth:`ack_message` when done, :meth:`nack_message` to redeliver now, or let
+the lease expire to redeliver — so one poison message no longer blocks its
+siblings. Oldest first; in-flight (leased, un-expired) messages are skipped.
+Don't mix with `read_topic`/`ack_topic` on the same subscription.
+
+### `queue.ack_message()` / `queue.nack_message()`
+
+```python
+queue.ack_message(topic: str, name: str, message_id: str) -> bool
+queue.nack_message(topic: str, name: str, message_id: str) -> bool
+```
+
+`ack_message` ends a leased delivery (never redelivered); `nack_message`
+makes it available for redelivery immediately. Both return `False` when there
+was no un-acked delivery for `message_id`.
+
 ### `queue.topic_log_stats()`
 
 ```python
@@ -190,11 +216,11 @@ up).
 from taskito import TopicMessage
 ```
 
-One message pulled from a log topic, returned by `read_topic()`.
+One message pulled from a log topic, returned by `read_topic()` and `lease_topic()`.
 
 | Field | Type | Description |
 |---|---|---|
-| `id` | `str` | Cursor token — pass to `ack_topic()`. |
+| `id` | `str` | Message id — pass to `ack_topic()` (cursor read) or to `ack_message()`/`nack_message()` (per-message lease). |
 | `args` | `tuple[Any, ...]` | Decoded positional args from the `publish()` call. |
 | `kwargs` | `dict[str, Any]` | Decoded keyword args from the `publish()` call. |
 | `metadata` | `dict[str, Any] \| None` | Caller metadata, if any. |
diff --git a/docs/content/docs/shared/guides/core/pubsub.mdx b/docs/content/docs/shared/guides/core/pubsub.mdx
index be7f8db3..572b3a14 100644
--- a/docs/content/docs/shared/guides/core/pubsub.mdx
+++ b/docs/content/docs/shared/guides/core/pubsub.mdx
@@ -604,6 +604,69 @@ poll), `batch_size` / `batchSize` (messages per poll), and `on_error` /
 re-reads, while `"skip"` acks past it and moves on. A non-empty batch drains
 immediately; only an empty one waits the poll interval.
 
+### Per-message ack
+
+The cursor is a single high-water mark, so one message a consumer can't
+process blocks everything behind it. **Per-message** consumption is the
+alternative: instead of the cursor read, *lease* messages and ack or nack each
+one individually. A leased message is skipped by later lease reads *on that
+same subscription* until its visibility window elapses (a separate subscription
+consuming the same topic is unaffected); ack it when done, nack it to redeliver
+immediately, or let the lease time out to have it redelivered — all without
+blocking its siblings. Delivery is **at-least-once**: a timed-out lease
+redelivers, and two concurrent leasers on one subscription can briefly claim the
+same message, so handlers must be idempotent. It's a consumption choice on the
+same log subscription, not a separate registration.
+
+<CodeTabs>
+  <Tab sdk="python">
+
+```python
+for msg in queue.lease_topic("orders", "audit-log", visibility=30):
+    try:
+        audit_sink.write(msg.args)
+        queue.ack_message("orders", "audit-log", msg.id)
+    except TransientError:
+        queue.nack_message("orders", "audit-log", msg.id)  # redeliver now
+```
+
+  </Tab>
+  <Tab sdk="node">
+
+```ts
+for (const msg of await queue.leaseTopic("orders", "audit-log", { visibility: 30 })) {
+  try {
+    auditSink.write(msg.args);
+    await queue.ackMessage("orders", "audit-log", msg.id);
+  } catch {
+    await queue.nackMessage("orders", "audit-log", msg.id); // redeliver now
+  }
+}
+```
+
+  </Tab>
+  <Tab sdk="java">
+
+```java
+for (TopicMessage msg : taskito.leaseTopic("orders", "audit-log", 100, Duration.ofSeconds(30))) {
+    try {
+        auditSink.write(serializer.deserialize(msg.payload, Order.class));
+        taskito.ackMessage("orders", "audit-log", msg.id);
+    } catch (TransientException e) {
+        taskito.nackMessage("orders", "audit-log", msg.id); // redeliver now
+    }
+}
+```
+
+  </Tab>
+</CodeTabs>
+
+Pick one style per subscription — mixing the cursor read and leasing on the
+same subscription lets their positions diverge. On every backend, a message
+every per-message consumer has acked is compacted like the cursor case, and its
+delivery state is dropped with it. A topic that mixes a per-message consumer
+with a plain cursor reader falls back to `expires` / retention-window cleanup.
+
 ### Retention compacts, it doesn't expire
 
 A log message is deleted once every log subscriber on its topic has acked
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java
index 93864d11..0da8fd7c 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java
@@ -913,14 +913,7 @@ public List<TopicMessage> readTopic(String topic, String name) {
 
     @Override
     public List<TopicMessage> readTopic(String topic, String name, int limit) {
-        List<TopicMessageRow> rows =
-                decodeList(backend.readTopicMessagesJson(topic, name, limit), TopicMessageRow.class);
-        List<TopicMessage> messages = new ArrayList<>(rows.size());
-        for (TopicMessageRow row : rows) {
-            messages.add(new TopicMessage(
-                    row.id(), row.payload(), parseJsonMap(row.metadata()), parseJsonMap(row.notes()), row.createdAt()));
-        }
-        return messages;
+        return decodeTopicMessages(backend.readTopicMessagesJson(topic, name, limit));
     }
 
     @Override
@@ -928,6 +921,26 @@ public boolean ackTopic(String topic, String name, String cursor) {
         return backend.ackTopicCursor(topic, name, cursor);
     }
 
+    @Override
+    public List<TopicMessage> leaseTopic(String topic, String name) {
+        return leaseTopic(topic, name, 100, Duration.ofSeconds(30));
+    }
+
+    @Override
+    public List<TopicMessage> leaseTopic(String topic, String name, int limit, Duration visibility) {
+        return decodeTopicMessages(backend.leaseTopicMessagesJson(topic, name, limit, visibility.toMillis()));
+    }
+
+    @Override
+    public boolean ackMessage(String topic, String name, String messageId) {
+        return backend.ackMessage(topic, name, messageId);
+    }
+
+    @Override
+    public boolean nackMessage(String topic, String name, String messageId) {
+        return backend.nackMessage(topic, name, messageId);
+    }
+
     @Override
     public List<TopicLogStat> topicLogStats() {
         return decodeList(backend.topicLogStatsJson(), TopicLogStat.class);
@@ -982,6 +995,17 @@ private static <T> Consumer<Object> castConsumer(Consumer<T> handler) {
         return (Consumer<Object>) handler;
     }
 
+    /** Decode a native JSON array of message views into {@link TopicMessage}s (shared by read/lease). */
+    private List<TopicMessage> decodeTopicMessages(String json) {
+        List<TopicMessageRow> rows = decodeList(json, TopicMessageRow.class);
+        List<TopicMessage> messages = new ArrayList<>(rows.size());
+        for (TopicMessageRow row : rows) {
+            messages.add(new TopicMessage(
+                    row.id(), row.payload(), parseJsonMap(row.metadata()), parseJsonMap(row.notes()), row.createdAt()));
+        }
+        return messages;
+    }
+
     /**
      * Native wire view of a log message: the payload crosses as base64 (Jackson
      * decodes it into {@code byte[]}); metadata/notes are JSON strings this SDK
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java
index 56dd7b47..61113af9 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java
@@ -446,6 +446,32 @@ static Builder builder() {
      */
     boolean ackTopic(String topic, String name, String cursor);
 
+    /** As {@link #leaseTopic(String, String, int, Duration)} with a default limit of 100 and a 30s visibility. */
+    List<TopicMessage> leaseTopic(String topic, String name);
+
+    /**
+     * Lease up to {@code limit} messages for <b>per-message</b> consumption. Unlike
+     * {@link #readTopic}'s cursor, each message is leased for {@code visibility} and
+     * tracked individually: {@link #ackMessage} it when done, or {@link #nackMessage}
+     * to redeliver it now. A lease that expires un-acked is redelivered, so one poison
+     * message no longer blocks its siblings. In-flight (leased, un-expired) messages
+     * are skipped; oldest first.
+     */
+    List<TopicMessage> leaseTopic(String topic, String name, int limit, Duration visibility);
+
+    /**
+     * Ack one leased message — the delivery is done and never redelivered. Returns
+     * false when there was no un-acked delivery to ack.
+     */
+    boolean ackMessage(String topic, String name, String messageId);
+
+    /**
+     * Nack one leased message — make it available for redelivery now, rather than
+     * waiting out the visibility timeout. Returns false when there was no un-acked
+     * delivery to nack.
+     */
+    boolean nackMessage(String topic, String name, String messageId);
+
     /** Lag snapshot per log subscription (cursor position and un-acked backlog). */
     List<TopicLogStat> topicLogStats();
 
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
index 3390ecef..49ee1f10 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java
@@ -408,6 +408,22 @@ public boolean ackTopicCursor(String topic, String subscriptionName, String curs
         return withOpenHandle(() -> NativeQueue.ackTopicCursor(handle, topic, subscriptionName, cursor));
     }
 
+    @Override
+    public String leaseTopicMessagesJson(String topic, String subscriptionName, long limit, long visibilityMs) {
+        return withOpenHandle(
+                () -> NativeQueue.leaseTopicMessages(handle, topic, subscriptionName, limit, visibilityMs));
+    }
+
+    @Override
+    public boolean ackMessage(String topic, String subscriptionName, String messageId) {
+        return withOpenHandle(() -> NativeQueue.ackMessage(handle, topic, subscriptionName, messageId));
+    }
+
+    @Override
+    public boolean nackMessage(String topic, String subscriptionName, String messageId) {
+        return withOpenHandle(() -> NativeQueue.nackMessage(handle, topic, subscriptionName, messageId));
+    }
+
     @Override
     public String topicLogStatsJson() {
         return withOpenHandle(() -> NativeQueue.topicLogStats(handle));
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java
index c3aef36e..80e3a49e 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java
@@ -184,6 +184,16 @@ public static native boolean setSubscriptionActive(
     /** Advance a log subscription's cursor (monotonic); false if nothing moved. */
     public static native boolean ackTopicCursor(long handle, String topic, String subscriptionName, String cursor);
 
+    /** Lease up to {@code limit} available messages for {@code visibilityMs}, as a JSON array of message views. */
+    public static native String leaseTopicMessages(
+            long handle, String topic, String subscriptionName, long limit, long visibilityMs);
+
+    /** Ack one leased message; false if there was no un-acked delivery to ack. */
+    public static native boolean ackMessage(long handle, String topic, String subscriptionName, String messageId);
+
+    /** Nack one leased message; false if there was no un-acked delivery to nack. */
+    public static native boolean nackMessage(long handle, String topic, String subscriptionName, String messageId);
+
     /** A JSON array of per-log-subscription lag snapshots. */
     public static native String topicLogStats(long handle);
 
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java
index c735607d..233fcaf5 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java
@@ -312,6 +312,25 @@ default boolean ackTopicCursor(String topic, String subscriptionName, String cur
         throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED);
     }
 
+    /**
+     * Lease up to {@code limit} available messages for {@code visibilityMs}, tracking
+     * per-message state, as a JSON array of message views. A nack or an expired lease
+     * redelivers just that message without blocking its siblings.
+     */
+    default String leaseTopicMessagesJson(String topic, String subscriptionName, long limit, long visibilityMs) {
+        throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED);
+    }
+
+    /** Ack one leased message; false if there was no un-acked delivery to ack. */
+    default boolean ackMessage(String topic, String subscriptionName, String messageId) {
+        throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED);
+    }
+
+    /** Nack one leased message (redeliver now); false if there was no un-acked delivery to nack. */
+    default boolean nackMessage(String topic, String subscriptionName, String messageId) {
+        throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED);
+    }
+
     /** A JSON array of per-log-subscription lag snapshots. */
     default String topicLogStatsJson() {
         throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED);
diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/PerMessageAckTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/PerMessageAckTest.java
new file mode 100644
index 00000000..0dcbe96c
--- /dev/null
+++ b/sdks/java/src/test/java/org/byteveda/taskito/core/PerMessageAckTest.java
@@ -0,0 +1,123 @@
+package org.byteveda.taskito.core;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import org.byteveda.taskito.Taskito;
+import org.byteveda.taskito.model.TopicMessage;
+import org.byteveda.taskito.serialization.JsonSerializer;
+import org.byteveda.taskito.serialization.Serializer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Per-message ack: lease / ack / nack / redelivery over a log topic. Unlike the
+ * S28 cursor read, each message is leased and tracked individually, so a nack or
+ * an expired lease redelivers just that message. The core lease algorithm is
+ * unit-tested in Rust; this drives the Java surface end-to-end.
+ */
+class PerMessageAckTest {
+
+    // leaseTopic returns raw payload bytes; decode them with the queue's default
+    // serializer to recover the published value.
+    private static final Serializer SERIALIZER = new JsonSerializer();
+
+    private Taskito open(Path dir) {
+        return Taskito.builder()
+                .backend("sqlite")
+                .url(dir.resolve("t.db").toString())
+                .open();
+    }
+
+    @Test
+    void leaseSkipsInFlightMessages(@TempDir Path dir) {
+        try (Taskito queue = open(dir)) {
+            queue.subscribeLog("events", "c");
+            for (int i = 0; i < 3; i++) {
+                queue.publish("events", String.valueOf(i));
+            }
+
+            // First lease takes the two oldest; both are in-flight for 30s.
+            List<TopicMessage> first = queue.leaseTopic("events", "c", 2, Duration.ofSeconds(30));
+            assertEquals(List.of("0", "1"), decodeAll(first));
+
+            // Second lease sees only the still-available third message.
+            assertEquals(List.of("2"), decodeAll(queue.leaseTopic("events", "c")));
+        }
+    }
+
+    @Test
+    void nackRedeliversButAckDoesNot(@TempDir Path dir) {
+        try (Taskito queue = open(dir)) {
+            queue.subscribeLog("events", "c");
+            queue.publish("events", "a");
+            queue.publish("events", "b");
+
+            List<TopicMessage> leased = queue.leaseTopic("events", "c");
+            assertEquals(List.of("a", "b"), decodeAll(leased));
+
+            // Ack the first (done forever), nack the second (redeliver now).
+            assertTrue(queue.ackMessage("events", "c", leased.get(0).id));
+            assertTrue(queue.nackMessage("events", "c", leased.get(1).id));
+
+            assertEquals(List.of("b"), decodeAll(queue.leaseTopic("events", "c")));
+        }
+    }
+
+    @Test
+    void ackAndNackReturnFalseWithoutLease(@TempDir Path dir) {
+        try (Taskito queue = open(dir)) {
+            queue.subscribeLog("events", "c");
+            queue.publish("events", "x");
+
+            // No lease taken yet, so there is no un-acked delivery to ack or nack.
+            String id = queue.readTopic("events", "c").get(0).id;
+            assertFalse(queue.ackMessage("events", "c", id));
+            assertFalse(queue.nackMessage("events", "c", id));
+        }
+    }
+
+    @Test
+    @Timeout(30)
+    void expiredLeaseRedelivers(@TempDir Path dir) throws InterruptedException {
+        try (Taskito queue = open(dir)) {
+            queue.subscribeLog("events", "c");
+            queue.publish("events", "x");
+
+            Duration visibility = Duration.ofMillis(100);
+            // Leased now; a second immediate lease sees nothing (still in-flight).
+            assertEquals(List.of("x"), decodeAll(queue.leaseTopic("events", "c", 100, visibility)));
+            assertTrue(queue.leaseTopic("events", "c", 100, visibility).isEmpty());
+
+            // Poll until the lease expires and the un-acked message is available again.
+            assertEquals(List.of("x"), pollForLease(queue, visibility));
+        }
+    }
+
+    /** Poll leaseTopic until it yields a message or the deadline passes (redelivery timing). */
+    private static List<String> pollForLease(Taskito queue, Duration visibility) throws InterruptedException {
+        long deadline = System.currentTimeMillis() + 5_000;
+        while (System.currentTimeMillis() < deadline) {
+            List<TopicMessage> leased = queue.leaseTopic("events", "c", 100, visibility);
+            if (!leased.isEmpty()) {
+                return decodeAll(leased);
+            }
+            Thread.sleep(20);
+        }
+        return List.of();
+    }
+
+    private static List<String> decodeAll(List<TopicMessage> msgs) {
+        List<String> out = new ArrayList<>(msgs.size());
+        for (TopicMessage msg : msgs) {
+            out.add(SERIALIZER.deserialize(msg.payload, String.class));
+        }
+        return out;
+    }
+}
diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts
index fcef17c6..5d60b162 100644
--- a/sdks/node/src/queue.ts
+++ b/sdks/node/src/queue.ts
@@ -27,6 +27,7 @@ import { Lock, type LockOptions } from "./locks";
 import type { EnqueueContext, Middleware } from "./middleware";
 import {
   JsQueue,
+  type JsTopicMessage,
   type EnqueueOptions as NativeEnqueueOptions,
   type NativeQueue,
   type OpenOptions,
@@ -652,13 +653,7 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
    */
   async readTopic(topic: string, name: string, limit = 100): Promise<TopicMessage[]> {
     const messages = await this.native.readTopicMessages(topic, name, limit);
-    return messages.map((msg) => ({
-      id: msg.id,
-      args: deserializeCall(this.serializer, msg.payload),
-      metadata: msg.metadata === undefined ? undefined : JSON.parse(msg.metadata),
-      notes: msg.notes === undefined ? undefined : JSON.parse(msg.notes),
-      createdAt: msg.createdAt,
-    }));
+    return messages.map((msg) => this.decodeTopicMessage(msg));
   }
 
   /**
@@ -670,6 +665,56 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
     return this.native.ackTopicCursor(topic, name, cursor);
   }
 
+  /**
+   * Lease up to `opts.limit` (default 100) messages for **per-message**
+   * consumption. Unlike `readTopic`'s cursor, each message is leased for
+   * `opts.visibility` seconds (default 30) and tracked individually: `ackMessage`
+   * it when done, or `nackMessage` to redeliver it now. A lease that expires
+   * un-acked is redelivered, so one poison message no longer blocks its siblings.
+   * In-flight (leased, un-expired) messages are skipped; oldest first.
+   */
+  async leaseTopic(
+    topic: string,
+    name: string,
+    opts?: { limit?: number; visibility?: number },
+  ): Promise<TopicMessage[]> {
+    const visibilityMs = Math.round((opts?.visibility ?? 30) * 1000);
+    const messages = await this.native.leaseTopicMessages(
+      topic,
+      name,
+      opts?.limit ?? 100,
+      visibilityMs,
+    );
+    return messages.map((msg) => this.decodeTopicMessage(msg));
+  }
+
+  /**
+   * Ack one leased message — the delivery is done and never redelivered. Resolves
+   * false when there was no un-acked delivery to ack.
+   */
+  ackMessage(topic: string, name: string, messageId: string): Promise<boolean> {
+    return this.native.ackMessage(topic, name, messageId);
+  }
+
+  /**
+   * Nack one leased message — make it available for redelivery now (vs waiting out
+   * the visibility timeout). Resolves false when there was no un-acked delivery.
+   */
+  nackMessage(topic: string, name: string, messageId: string): Promise<boolean> {
+    return this.native.nackMessage(topic, name, messageId);
+  }
+
+  /** Decode a native log message into a {@link TopicMessage} (shared by read/lease). */
+  private decodeTopicMessage(msg: JsTopicMessage): TopicMessage {
+    return {
+      id: msg.id,
+      args: deserializeCall(this.serializer, msg.payload),
+      metadata: msg.metadata === undefined ? undefined : JSON.parse(msg.metadata),
+      notes: msg.notes === undefined ? undefined : JSON.parse(msg.notes),
+      createdAt: msg.createdAt,
+    };
+  }
+
   /** Lag snapshot per log subscription. */
   topicLogStats(): Promise<TopicLogStat[]> {
     return this.native.topicLogStats();
diff --git a/sdks/node/test/core/per-message-ack.test.ts b/sdks/node/test/core/per-message-ack.test.ts
new file mode 100644
index 00000000..3451827a
--- /dev/null
+++ b/sdks/node/test/core/per-message-ack.test.ts
@@ -0,0 +1,79 @@
+// Per-message ack: lease / ack / nack / redelivery over a log topic. Unlike the
+// S28 cursor read, each message is leased and tracked individually, so a nack or
+// an expired lease redelivers just that message without blocking its siblings.
+
+import { mkdtempSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { expect, it } from "vitest";
+import { Queue } from "../../src/index";
+
+function newQueue(): Queue {
+  return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-per-msg-ack-")), "q.db") });
+}
+
+/** Narrow an array element to non-undefined (strict indexed access). */
+function must<T>(value: T | undefined): T {
+  if (value === undefined) throw new Error("expected a value");
+  return value;
+}
+
+it("leases available messages and skips in-flight ones", async () => {
+  const queue = newQueue();
+  await queue.subscribeLog("events", "c");
+  for (let i = 0; i < 3; i++) await queue.publish("events", [i]);
+
+  // First lease takes the two oldest; they are now in-flight for 30s.
+  const first = await queue.leaseTopic("events", "c", { limit: 2 });
+  expect(first.map((m) => m.args[0])).toEqual([0, 1]);
+
+  // Second lease sees only the still-available third message.
+  const second = await queue.leaseTopic("events", "c");
+  expect(second.map((m) => m.args[0])).toEqual([2]);
+});
+
+it("redelivers a nacked message but not an acked one", async () => {
+  const queue = newQueue();
+  await queue.subscribeLog("events", "c");
+  await queue.publish("events", ["a"]);
+  await queue.publish("events", ["b"]);
+
+  const leased = await queue.leaseTopic("events", "c");
+  expect(leased.map((m) => m.args[0])).toEqual(["a", "b"]);
+
+  // Ack the first (done forever), nack the second (redeliver now).
+  expect(await queue.ackMessage("events", "c", must(leased[0]).id)).toBe(true);
+  expect(await queue.nackMessage("events", "c", must(leased[1]).id)).toBe(true);
+
+  const redelivered = await queue.leaseTopic("events", "c");
+  expect(redelivered.map((m) => m.args[0])).toEqual(["b"]);
+});
+
+it("redelivers after a short visibility lease expires", async () => {
+  const queue = newQueue();
+  await queue.subscribeLog("events", "c");
+  await queue.publish("events", ["x"]);
+
+  // Lease with a 100ms visibility; it is in-flight immediately after.
+  expect(
+    (await queue.leaseTopic("events", "c", { visibility: 0.1 })).map((m) => m.args[0]),
+  ).toEqual(["x"]);
+  expect(await queue.leaseTopic("events", "c", { visibility: 0.1 })).toEqual([]);
+
+  // Once the lease expires the un-acked message becomes available again.
+  await new Promise((r) => setTimeout(r, 150));
+  expect(
+    (await queue.leaseTopic("events", "c", { visibility: 0.1 })).map((m) => m.args[0]),
+  ).toEqual(["x"]);
+});
+
+it("returns false when there is nothing to ack or nack", async () => {
+  const queue = newQueue();
+  await queue.subscribeLog("events", "c");
+  await queue.publish("events", ["x"]);
+
+  // No lease has been taken, so the message id is unknown to ack/nack.
+  const msg = must((await queue.readTopic("events", "c"))[0]);
+  expect(await queue.ackMessage("events", "c", msg.id)).toBe(false);
+  expect(await queue.nackMessage("events", "c", msg.id)).toBe(false);
+});
diff --git a/sdks/python/taskito/_taskito.pyi b/sdks/python/taskito/_taskito.pyi
index 9fbff4ad..b9f7c7be 100644
--- a/sdks/python/taskito/_taskito.pyi
+++ b/sdks/python/taskito/_taskito.pyi
@@ -245,6 +245,11 @@ class PyQueue:
         self, topic: str, subscription_name: str, limit: int
     ) -> list[tuple[str, bytes, str | None, str | None, int]]: ...
     def ack_topic_cursor(self, topic: str, subscription_name: str, cursor: str) -> bool: ...
+    def lease_topic_messages(
+        self, topic: str, subscription_name: str, limit: int, visibility_ms: int
+    ) -> list[tuple[str, bytes, str | None, str | None, int]]: ...
+    def ack_message(self, topic: str, subscription_name: str, message_id: str) -> bool: ...
+    def nack_message(self, topic: str, subscription_name: str, message_id: str) -> bool: ...
     def topic_log_stats(
         self,
     ) -> list[tuple[str, str, str | None, int, int | None]]: ...
diff --git a/sdks/python/taskito/mixins/pubsub.py b/sdks/python/taskito/mixins/pubsub.py
index 94fbd404..da791490 100644
--- a/sdks/python/taskito/mixins/pubsub.py
+++ b/sdks/python/taskito/mixins/pubsub.py
@@ -230,10 +230,15 @@ def read_topic(self, topic: str, name: str, limit: int = 100) -> list[TopicMessa
         queue serializer. Returns an empty list when the consumer is caught up.
         Delivery is at-least-once: process, then :meth:`ack_topic` the last id.
         """
+        return self._decode_topic_messages(self._inner.read_topic_messages(topic, name, limit))
+
+    def _decode_topic_messages(
+        self, rows: list[tuple[str, bytes, str | None, str | None, int]]
+    ) -> list[TopicMessage]:
+        """Decode native ``(id, payload, metadata, notes, created_at)`` rows into
+        :class:`TopicMessage`, unpacking the payload with the queue serializer."""
         messages = []
-        for msg_id, payload, metadata, notes, created_at in self._inner.read_topic_messages(
-            topic, name, limit
-        ):
+        for msg_id, payload, metadata, notes, created_at in rows:
             args, kwargs = self._serializer.loads(payload)
             messages.append(
                 TopicMessage(
@@ -255,6 +260,33 @@ def ack_topic(self, topic: str, name: str, cursor: str) -> bool:
         """
         return self._inner.ack_topic_cursor(topic, name, cursor)
 
+    def lease_topic(
+        self, topic: str, name: str, limit: int = 100, visibility: float = 30.0
+    ) -> list[TopicMessage]:
+        """Lease up to ``limit`` messages for **per-message** consumption.
+
+        Unlike :meth:`read_topic`'s cursor, each returned message is leased for
+        ``visibility`` seconds and tracked individually: :meth:`ack_message` it
+        when done, or :meth:`nack_message` to redeliver it now. A lease that
+        expires un-acked is redelivered, so one poison message no longer blocks
+        its siblings. Oldest first; in-flight (leased, un-expired) messages are
+        skipped.
+        """
+        return self._decode_topic_messages(
+            self._inner.lease_topic_messages(topic, name, limit, int(visibility * 1000))
+        )
+
+    def ack_message(self, topic: str, name: str, message_id: str) -> bool:
+        """Ack one leased message — the delivery is done and never redelivered.
+        Returns False when there was no un-acked delivery to ack."""
+        return self._inner.ack_message(topic, name, message_id)
+
+    def nack_message(self, topic: str, name: str, message_id: str) -> bool:
+        """Nack one leased message — make it available for redelivery now (vs
+        waiting out the visibility timeout). Returns False when there was no
+        un-acked delivery to nack."""
+        return self._inner.nack_message(topic, name, message_id)
+
     def topic_log_stats(self) -> list[dict[str, Any]]:
         """Lag snapshot per log subscription.
 
diff --git a/sdks/python/tests/core/test_pubsub_log.py b/sdks/python/tests/core/test_pubsub_log.py
index 9342e8c4..8a95270c 100644
--- a/sdks/python/tests/core/test_pubsub_log.py
+++ b/sdks/python/tests/core/test_pubsub_log.py
@@ -1,6 +1,7 @@
 """Log topics (S28): one stored message per publish, pulled via a cursor."""
 
 import threading
+import time
 from typing import Any
 
 from taskito import Queue, TopicMessage
@@ -228,3 +229,34 @@ def test_list_declared_topics_and_retention_round_trip(self, queue: Queue) -> No
         topics = {t["name"]: t for t in queue.list_declared_topics()}
         assert topics["orders"]["retention_ms"] == 2000
         assert len(queue.list_declared_topics()) == 2
+
+
+class TestPerMessageAck:
+    def test_lease_ack_nack(self, queue: Queue) -> None:
+        queue.subscribe_log("events", "w")
+        for i in range(3):
+            queue.publish("events", i)
+
+        # Lease 2 with a long visibility; a second lease sees only msg 2.
+        batch = queue.lease_topic("events", "w", limit=2, visibility=60)
+        assert [m.args[0] for m in batch] == [0, 1]
+        assert [m.args[0] for m in queue.lease_topic("events", "w", visibility=60)] == [2]
+
+        # Ack 0 (done); nack 1 (redeliver now). Acking 0 again is a no-op.
+        assert queue.ack_message("events", "w", batch[0].id) is True
+        assert queue.nack_message("events", "w", batch[1].id) is True
+        assert queue.ack_message("events", "w", batch[0].id) is False
+
+        # Only the nacked msg 1 comes back (0 acked, 2 in-flight).
+        assert [m.args[0] for m in queue.lease_topic("events", "w", visibility=60)] == [1]
+
+    def test_visibility_timeout_redelivers(self, queue: Queue) -> None:
+        queue.subscribe_log("events", "w")
+        queue.publish("events", 1)
+
+        assert [m.args[0] for m in queue.lease_topic("events", "w", visibility=0.1)] == [1]
+        # Within the window it is not re-leased.
+        assert queue.lease_topic("events", "w", visibility=0.1) == []
+        time.sleep(0.15)
+        # Past the timeout, an un-acked message redelivers.
+        assert [m.args[0] for m in queue.lease_topic("events", "w", visibility=0.1)] == [1]