From 2ecc744b30bf5eb2ce1c7ca5bffcc07985470611 Mon Sep 17 00:00:00 2001
From: Paolo Di Tommaso <paolo.ditommaso@gmail.com>
Date: Sat, 25 Jul 2026 15:32:21 +0200
Subject: [PATCH] Rework work queue around invocation leases

---
 lib-cmd-queue-redis/README.md                 |  66 ++---
 lib-cmd-queue-redis/changelog.txt             |   9 +
 .../io/seqera/data/command/CommandConfig.java |  18 +-
 .../io/seqera/data/command/CommandQueue.java  |   7 +-
 .../data/command/CommandServiceImpl.java      | 124 +++++++--
 .../io/seqera/data/command/CommandState.java  |   7 +
 .../io/seqera/data/command/CommandStatus.java |   5 +
 .../data/command/store/CommandStateStore.java |  21 ++
 .../command/store/CommandStateStoreImpl.java  |  38 +++
 .../command/CommandServiceRedisE2ETest.groovy |  93 +++++++
 .../store/CommandAttemptStoreTest.groovy      |  88 +++++++
 .../command/TestRedisWorkQueueConfig.java     |  35 +++
 .../state/impl/LocalStateProvider.groovy      |  24 ++
 .../state/impl/RedisStateProvider.groovy      |  33 +++
 .../store/state/impl/StateProvider.groovy     |  10 +
 .../state/impl/LocalStateProviderTest.groovy  |  13 +
 .../state/impl/RedisStateProviderTest.groovy  |  12 +
 .../data/workqueue/redis/RedisWorkQueue.java  |  45 ++--
 .../workqueue/redis/RedisWorkQueueConfig.java |   7 +-
 .../workqueue/AsyncWorkQueueRedisTest.groovy  |  75 ++++--
 .../seqera/data/workqueue/TunableQueue.groovy |   2 +-
 lib-data-workqueue/README.md                  |  55 ++--
 lib-data-workqueue/changelog.txt              |   8 +
 .../data/workqueue/AbstractWorkQueue.java     | 237 ++++++++----------
 .../seqera/data/workqueue/LocalWorkQueue.java |   7 +-
 .../io/seqera/data/workqueue/WorkQueue.java   |  36 ++-
 .../workqueue/AsyncWorkQueueLocalTest.groovy  |  17 +-
 .../seqera/data/workqueue/TunableQueue.groovy |   2 +-
 28 files changed, 807 insertions(+), 287 deletions(-)
 create mode 100644 lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandServiceRedisE2ETest.groovy
 create mode 100644 lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/store/CommandAttemptStoreTest.groovy
 create mode 100644 lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestRedisWorkQueueConfig.java

diff --git a/lib-cmd-queue-redis/README.md b/lib-cmd-queue-redis/README.md
index 438786b8..51b150b0 100644
--- a/lib-cmd-queue-redis/README.md
+++ b/lib-cmd-queue-redis/README.md
@@ -16,9 +16,9 @@ dependencies {
 
 - Fire-and-forget command submission
 - Typed parameters and results with JSON serialization
-- Status transitions: `PENDING` → `PROCESSING` → `SUCCEEDED`/`FAILED`/`CANCELLED`
+- Status transitions: `PENDING` → `SUBMITTING` → `PROCESSING` → terminal
 - Non-blocking, concurrent handler execution on virtual threads (no per-command timeout)
-- Periodic status checking for async commands via in-process re-polling
+- Durable status checking via Redis Stream PEL redelivery
 - Command cancellation support
 - Persistent storage using Redis or in-memory backend
 
@@ -162,20 +162,19 @@ just transport; the store is the source of truth.
         ▼
   ┌──────────────┐  save() ┌──────────────────────────────────────────────┐
   │ CommandState │◀────────┤              CommandServiceImpl                │
-  │    store     │  find() │  processCommand(msg): load state, then         │──▶ execute()
-  │ (Redis/mem)  │────────▶│  dispatch the handler to a worker pool         │    checkStatus()
-  │              │         │  (off the dispatcher thread; virtual threads): │
-  └──────────────┘         │    • terminal     → ack (remove from queue)    │
-        ▲                  │    • processing() → keep lease, re-poll after  │
-        │ getState/Result  │                     pollInterval (in-process)  │
+  │    store     │  find() │  acquire renewable command-attempt guard,      │──▶ execute()
+  │ (Redis/mem)  │────────▶│  load state, invoke one handler operation:     │    checkStatus()
+  │              │         │    • terminal     → persist, then ack          │
+  └──────────────┘         │    • processing() → persist, return to PEL     │
+        ▲                  │    • duplicate    → guard rejects invocation   │
+        │ getState/Result  │                                                │
         │                  └───────────────────────┬────────────────────────┘
         │                       submit(msg)         │  addConsumer(processCommand)
         │                                           ▼
         │                     ┌────────────────────────────────────────────┐
         └─────────────────────│  CommandQueue (Redis work queue / in-mem.)  │
                               │  = AbstractWorkQueue: dispatcher + worker   │
-                              │  pool + heartbeat lease → exactly one live  │
-                              │  runner per command, no timeout             │
+                              │  pool + invocation heartbeat, at-least-once │
                               └────────────────────────────────────────────┘
 ```
 
@@ -196,9 +195,10 @@ implementations (useful for tests and single-node setups).
 
 ### Submit path
 
-`submit()` is fire-and-forget: it persists a `PENDING` `CommandState` to the
-store, then enqueues a `CommandMsg`, and returns the command id immediately. No
-handler runs on the caller's thread.
+`submit()` is fire-and-forget and idempotent by command id: only the first creation
+persists `PENDING`. A retry for an existing non-terminal command enqueues another
+`CommandMsg`, repairing the state-write/queue-offer failure window; the attempt guard
+collapses duplicate deliveries during processing.
 
 ### Processing loop
 
@@ -209,43 +209,43 @@ returns immediately, so a slow handler never blocks intake. The consumer returns
 boolean:
 
 - **`true`** → terminal; the message is acknowledged and removed.
-- **`false`** → not yet terminal; the command **keeps its lease** and `processCommand`
-  is re-invoked in-process after `pollInterval` (the poll loop for long-running commands).
+- **`false`** → not yet terminal; the invocation releases its thread and permit, and
+  the Stream PEL makes it eligible again after visibility timeout.
 
 For each delivery, `processCommand` loads the state and decides:
 
 1. **State missing or already terminal** → `true`; nothing to do (another replica finished it, or it was cancelled).
 2. **No handler registered** → mark `FAILED`, `true`.
-3. **State is `PENDING`** → run `handler.execute()`. Terminal result → apply, `true`;
-   `processing()` → mark `PROCESSING`, `false` (re-polled after `pollInterval`).
-4. **State is `PROCESSING`** → run `handler.checkStatus()`. Terminal → `true`;
+3. **State is `PENDING`** → persist `SUBMITTING`, then run `handler.execute()` using the
+   stable command id for external idempotency. Terminal result → apply, `true`;
+   `processing()` → mark `PROCESSING`, `false`.
+4. **State is `SUBMITTING`** → repeat idempotent `execute()` after an interrupted submission.
+5. **State is `PROCESSING`** → run `handler.checkStatus()`. Terminal → `true`;
    `processing()` → `false` (re-polled again).
 
 A quick command finishes in one delivery; a slow or external one flips to `PROCESSING`
-and is driven to completion by repeated `checkStatus()` calls at `pollInterval`
-cadence. Handler exceptions transition the command to `FAILED` and ack. There is no
-per-command timeout and no per-command lock — the handler runs to completion on a
-virtual thread, and the underlying work queue's per-message lease guarantees a single
-concurrent runner across replicas (see
+and is driven to completion by repeated `checkStatus()` calls at Stream visibility-timeout
+cadence. Handler exceptions remain retryable. A renewable, owner-checked attempt record
+in `CommandStateStore` prevents duplicate at-least-once deliveries from concurrently
+invoking or persisting the same command (see
 [`lib-data-workqueue-redis`](../lib-data-workqueue-redis/README.md)).
 
 ### Multi-replica behaviour
 
-The work queue's per-message lease (a heartbeated Redis consumer-group entry) ensures a
-command is processed by exactly one live replica at a time; if that replica dies, the
-lease lapses and a peer reclaims the command. The store is the shared source of truth,
-so the terminal-state check keeps processing idempotent. Delivery is **at-least-once**,
-so `execute()`/`checkStatus()` should be idempotent.
+The work queue provides at-least-once transport and heartbeats only while one handler call
+runs. The command store provides a second renewable attempt guard and terminal-state check.
+`execute()` must use `command.id()` as an external idempotency key wherever supported;
+no local state can make an uncooperative external side effect exactly once.
 
 ## Command Status Flow
 
 ```
-submit() ──▶ PENDING ──pickup──▶ PROCESSING ─┬─success──▶ SUCCEEDED
-                                             ├─error────▶ FAILED
-                                             └─cancel───▶ CANCELLED
+submit() ──▶ PENDING ──pickup──▶ SUBMITTING ──▶ PROCESSING ─┬─success──▶ SUCCEEDED
+                                                            ├─error────▶ FAILED
+                                                            └─cancel───▶ CANCELLED
 
-(new state persists as PENDING/PROCESSING; legacy SUBMITTED/RUNNING entries still decode.
-Upgrading across this rename requires a zero-overlap rollout — see changelog 0.7.0.)
+(new state persists as PENDING/SUBMITTING/PROCESSING; legacy SUBMITTED/RUNNING entries
+still decode. SUBMITTING is a new wire value; see the changelog before a mixed-version rollout.)
 ```
 
 ## Testing
diff --git a/lib-cmd-queue-redis/changelog.txt b/lib-cmd-queue-redis/changelog.txt
index 9e2bb4b2..454c589d 100644
--- a/lib-cmd-queue-redis/changelog.txt
+++ b/lib-cmd-queue-redis/changelog.txt
@@ -1,5 +1,14 @@
 # lib-cmd-queue-redis changelog
 
+Next
+- Add renewable, owner-checked command attempt guards in CommandStateStore. Duplicate
+  at-least-once deliveries cannot concurrently invoke or persist the same command.
+- Add SUBMITTING state before the initial external call. Retries use the stable command id
+  as their idempotency key until PROCESSING is durably recorded.
+- Make submit idempotent by command id and serialize cancellation with handler attempts.
+- Command retries now come from the Redis Stream PEL after visibility timeout; no command
+  lifecycle remains scheduled in AbstractWorkQueue.
+
 0.7.0 - 17 Jul 2026
 - Adopt the lib-data-workqueue / lib-data-workqueue-redis modules; no longer depends on
   lib-data-stream-redis (which remains available as a standalone module).
diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandConfig.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandConfig.java
index af64da15..675c3116 100644
--- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandConfig.java
+++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandConfig.java
@@ -45,12 +45,22 @@ default Duration stateTtl() {
     }
 
     /**
-     * Maximum number of commands that may be in flight on a single instance at once.
-     * Handlers run on virtual threads, so this is a memory/heartbeat ceiling (the underlying
-     * work queue's in-flight semaphore), not a thread-pool size; commands beyond it wait in
-     * the queue (backpressure). Effective minimum is 1.
+     * Maximum number of handler invocations that may run concurrently on one instance.
+     * A non-terminal command releases its permit between Stream deliveries.
      */
     default int concurrency() {
         return 1000;
     }
+
+    /**
+     * Distributed state-guard lifetime for one handler invocation. The guard is renewed
+     * periodically while the invocation runs.
+     */
+    default Duration attemptLease() {
+        return Duration.ofMinutes(5);
+    }
+
+    default Duration attemptHeartbeat() {
+        return attemptLease().dividedBy(3);
+    }
 }
diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandQueue.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandQueue.java
index 8a60b89d..f35ca3c9 100644
--- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandQueue.java
+++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandQueue.java
@@ -79,11 +79,8 @@ protected Duration pollInterval() {
     }
 
     /**
-     * Maximum number of commands in flight on this instance at once, from
-     * {@link CommandConfig#concurrency()}. Handlers run on virtual threads, so this is a
-     * memory/heartbeat ceiling rather than a thread count; commands beyond it wait in the
-     * queue (backpressure). Cross-replica single-runner exclusion is provided by the
-     * per-message lease, so no per-command lock is required.
+     * Maximum number of command-handler invocations running on this instance at once.
+     * Live commands do not retain permits between Stream deliveries.
      */
     @Override
     protected int concurrency() {
diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandServiceImpl.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandServiceImpl.java
index 1082bc3e..0d89dc6b 100644
--- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandServiceImpl.java
+++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandServiceImpl.java
@@ -18,8 +18,13 @@
 
 import java.util.Map;
 import java.util.Optional;
+import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 
 import io.micronaut.scheduling.TaskExecutors;
 import io.seqera.data.command.store.CommandStateStore;
@@ -35,13 +40,14 @@
  *
  * <p>Processing runs on the shared worker pool of the underlying message stream, so
  * neither {@code execute()} nor {@code checkStatus()} blocks the dispatcher loop and no
- * per-command timeout is needed. Cross-replica single-runner exclusion comes from the
- * stream's per-message lease.
+ * per-command timeout is needed. Cross-replica single-runner exclusion and stale-writer
+ * fencing come from the command attempt guard in the state store.
  *
  * <p>Processing flow:
  * <ul>
  *   <li>If command is already PROCESSING → call checkStatus()</li>
- *   <li>If command is still PENDING → call execute()</li>
+ *   <li>If command is PENDING → persist SUBMITTING, then call execute()</li>
+ *   <li>If command is SUBMITTING after redelivery → repeat execute() with the stable command ID</li>
  *   <li>If result is PROCESSING → mark as PROCESSING and return false (re-polled later)</li>
  *   <li>If result is terminal → apply result and return true (message removed from queue)</li>
  *   <li>If the handler throws → return false so the message is retried; a throw is treated as
@@ -60,12 +66,21 @@ public class CommandServiceImpl implements CommandService {
     @Inject
     private CommandQueue queue;
 
+    @Inject
+    private CommandConfig config;
+
     @Inject
     @Named(TaskExecutors.BLOCKING)
     private ExecutorService blockingExecutor;
 
     private final Map<String, CommandRegistration<?, ?>> handlers = new ConcurrentHashMap<>();
 
+    private final ScheduledExecutorService attemptHeartbeat = Executors.newSingleThreadScheduledExecutor(r -> {
+        final Thread thread = new Thread(r, "command-attempt-heartbeat");
+        thread.setDaemon(true);
+        return thread;
+    });
+
     private volatile boolean started = false;
 
     @Override
@@ -75,7 +90,7 @@ public void start() {
             return;
         }
         started = true;
-        // run handlers on the shared Micronaut BLOCKING (virtual-thread) executor
+        // run handlers on the shared Micronaut BLOCKING executor
         queue.withHandlerExecutor(blockingExecutor);
         queue.addConsumer(this::processCommand);
         log.info("Command service started - consuming commands");
@@ -97,8 +112,14 @@ public <P> String submit(Command<P> command) {
         final var state = CommandState.submitted(command.id(), command.type(), command.params());
 
         // Persist to storage and submit to queue
-        store.save(state);
-        queue.submit(CommandMsg.of(command.id(), command.type()));
+        final boolean created = store.create(state);
+        final CommandState existing = created ? state : store.findById(command.id()).orElse(null);
+        // Re-enqueue a pre-existing non-terminal command so a caller retry can repair the
+        // state-created/queue-offer failure window. Duplicate messages are fenced by the
+        // command attempt guard and terminal-state check.
+        if (created || (existing != null && !existing.status().isTerminal())) {
+            queue.submit(CommandMsg.of(command.id(), command.type()));
+        }
 
         log.debug("Command submitted: id={}, type={}", command.id(), command.type());
         return command.id();
@@ -119,18 +140,24 @@ public <R> Optional<R> getResult(String commandId, Class<R> resultType) {
 
     @Override
     public boolean cancel(String commandId) {
-        final var state = store.findById(commandId).orElse(null);
-        if (state == null) {
+        final String owner = UUID.randomUUID().toString();
+        if (!store.tryAcquireAttempt(commandId, owner, config.attemptLease())) {
             return false;
         }
-
-        if (state.status().isTerminal()) {
-            return false;
+        try {
+            final var state = store.findById(commandId).orElse(null);
+            if (state == null || state.status().isTerminal()) {
+                return false;
+            }
+            if (!store.saveOwned(state.cancelled(), owner)) {
+                return false;
+            }
+            log.info("Command cancelled: id={}", commandId);
+            return true;
+        }
+        finally {
+            store.releaseAttempt(commandId, owner);
         }
-
-        store.save(state.cancelled());
-        log.info("Command cancelled: id={}", commandId);
-        return true;
     }
 
     @Override
@@ -185,6 +212,37 @@ public P params() {
      * @return true to acknowledge (remove from queue), false to retry later
      */
     private boolean processCommand(CommandMsg msg) {
+        final String owner = UUID.randomUUID().toString();
+        if (!store.tryAcquireAttempt(msg.commandId(), owner, config.attemptLease())) {
+            log.debug("Command attempt already active, retaining delivery: id={}", msg.commandId());
+            return false;
+        }
+        final ScheduledFuture<?> renewer = attemptHeartbeat.scheduleAtFixedRate(
+                () -> renewAttempt(msg.commandId(), owner),
+                config.attemptHeartbeat().toMillis(),
+                config.attemptHeartbeat().toMillis(),
+                TimeUnit.MILLISECONDS);
+        try {
+            return processCommandOwned(msg, owner);
+        }
+        finally {
+            renewer.cancel(false);
+            store.releaseAttempt(msg.commandId(), owner);
+        }
+    }
+
+    private void renewAttempt(String commandId, String owner) {
+        try {
+            if (!store.renewAttempt(commandId, owner, config.attemptLease())) {
+                log.warn("Command attempt ownership lost while handler is active: id={}", commandId);
+            }
+        }
+        catch (Throwable e) {
+            log.warn("Unable to renew command attempt: id={}; cause={}", commandId, e.getMessage());
+        }
+    }
+
+    private boolean processCommandOwned(CommandMsg msg, String owner) {
         // Step 1: Load command state from persistent storage
         var state = store.findById(msg.commandId()).orElse(null);
         if (state == null) {
@@ -204,14 +262,14 @@ private boolean processCommand(CommandMsg msg) {
         final var registration = getHandler(state.type());
         if (registration == null) {
             log.error("No handler for command type: {}", state.type());
-            store.save(state.failed("No handler for type: " + state.type()));
+            saveOwned(state.failed("No handler for type: " + state.type()), owner);
             return true;
         }
 
         // Step 4: Delegate to the type-capturing helper method
         // This pattern allows Java to infer concrete type parameters (P, R) from the
         // CommandRegistration, enabling type-safe handler invocation without raw types.
-        return processCommandWithHandler(msg, state, registration);
+        return processCommandWithHandler(msg, state, registration, owner);
     }
 
     /**
@@ -223,7 +281,8 @@ private boolean processCommand(CommandMsg msg) {
      * <p>Runs directly on the shared worker pool thread (no timeout, no extra executor):
      * <ol>
      *   <li>If command is already PROCESSING → call {@code checkStatus()} to poll for completion</li>
-     *   <li>If command is still PENDING → call {@code execute()}</li>
+     *   <li>If command is PENDING → persist SUBMITTING, then call {@code execute()}</li>
+     *   <li>If command is SUBMITTING → repeat {@code execute()} with the stable command ID</li>
      *   <li>If result status is PROCESSING → mark PROCESSING and return false (re-polled later)</li>
      *   <li>If result status is terminal → update state and return true (done)</li>
      * </ol>
@@ -238,21 +297,26 @@ private boolean processCommand(CommandMsg msg) {
     private <P, R> boolean processCommandWithHandler(
             CommandMsg msg,
             CommandState state,
-            CommandRegistration<P, R> registration) {
+            CommandRegistration<P, R> registration,
+            String owner) {
 
         // Reconstruct the typed Command object from persisted state
         // Uses Class.cast() internally for type-safe conversion
+        if (state.status() == CommandStatus.PENDING) {
+            state = state.submitting();
+            saveOwned(state, owner);
+        }
         final Command<P> command = toCommand(state, registration);
         final CommandHandler<P, R> handler = registration.handler();
 
         try {
-            // Branch on the command status. Only PENDING and PROCESSING are reachable here
+            // Branch on the command status. Only SUBMITTING and PROCESSING are reachable here
             // (processCommand already acked terminal states); any other value is a bug or a
             // newly-added status and must fail loudly rather than be silently executed. Both
             // execute() and checkStatus() run on the shared worker pool, so a slow handler
             // does not block the loop.
             final CommandResult<R> result = switch (state.status()) {
-                case PENDING -> handler.execute(command);
+                case SUBMITTING -> handler.execute(command);
                 case PROCESSING -> handler.checkStatus(command, state);
                 default -> throw new IllegalStateException("Unexpected command status: " + state.status() + " - id=" + state.id());
             };
@@ -262,11 +326,11 @@ private <P, R> boolean processCommandWithHandler(
                 // Handler explicitly returned PROCESSING (e.g., async job not yet complete)
                 // Ensure state reflects PROCESSING status for accurate reporting
                 if (state.status() != CommandStatus.PROCESSING) {
-                    store.save(state.started());
+                    saveOwned(state.started(), owner);
                 } else if (state.errorsCount() > 0) {
                     // Recovered after one or more transient errors — reset the streak. Single write,
                     // and only when there is something to reset, so healthy re-polls stay write-free.
-                    store.save(state.clearErrors());
+                    saveOwned(state.clearErrors(), owner);
                 }
                 return false; // Keep in queue - re-polled and will call checkStatus()
             }
@@ -274,7 +338,7 @@ private <P, R> boolean processCommandWithHandler(
             // Terminal result (SUCCEEDED, FAILED, or CANCELLED)
             // Apply the result to transition to terminal state
             final CommandState newState = state.applyResult(result);
-            store.save(newState);
+            saveOwned(newState, owner);
             log.debug("Command completed: id={}, status={}", state.id(), newState.status());
             return true; // Remove from queue - processing complete
 
@@ -288,7 +352,7 @@ private <P, R> boolean processCommandWithHandler(
             // non-terminal, stranding the work. Deciding a command has *permanently* failed is
             // delegated to the domain layer that owns the entity state (see seqeralabs/sched#712).
             log.error("Command processing errored, will retry: id={}", msg.commandId(), e);
-            recordError(state, e);
+            recordError(state, e, owner);
             return false; // Keep in queue - redelivered / re-polled
         }
     }
@@ -299,11 +363,17 @@ private <P, R> boolean processCommandWithHandler(
      * command that stays retryable. A failure to persist this must not change control flow: the
      * command is kept in the queue and retried regardless.
      */
-    private void recordError(CommandState state, Exception e) {
+    private void recordError(CommandState state, Exception e, String owner) {
         try {
-            store.save(state.withError(e.getMessage() != null ? e.getMessage() : e.toString()));
+            saveOwned(state.withError(e.getMessage() != null ? e.getMessage() : e.toString()), owner);
         } catch (Exception fail) {
             log.warn("Failed to record command error state: id={}", state.id(), fail);
         }
     }
+
+    private void saveOwned(CommandState state, String owner) {
+        if (!store.saveOwned(state, owner)) {
+            throw new IllegalStateException("Command attempt ownership lost: id=" + state.id());
+        }
+    }
 }
diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java
index f4ca3898..da5d431a 100644
--- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java
+++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java
@@ -87,6 +87,13 @@ public CommandState started() {
         );
     }
 
+    public CommandState submitting() {
+        return new CommandState(
+                id, type, CommandStatus.SUBMITTING, params,
+                result, error, 0, createdAt, startedAt, Instant.now(), completedAt
+        );
+    }
+
     /**
      * Transition to SUCCEEDED status with result.
      */
diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java
index 400c25ab..077002b7 100644
--- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java
+++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java
@@ -31,6 +31,11 @@ public enum CommandStatus {
     /** In the queue, awaiting first processing (legacy wire name: {@code "SUBMITTED"}). */
     @JsonAlias("SUBMITTED")
     PENDING,
+    /**
+     * Initial external submission may have started. Redelivery repeats execute with the
+     * stable command id as idempotency key until PROCESSING is durably recorded.
+     */
+    SUBMITTING,
     /** Being processed by a handler (legacy wire name: {@code "RUNNING"}). */
     @JsonAlias("RUNNING")
     PROCESSING,
diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStore.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStore.java
index 0209c4ac..54f48dcd 100644
--- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStore.java
+++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStore.java
@@ -16,6 +16,7 @@
  */
 package io.seqera.data.command.store;
 
+import java.time.Duration;
 import java.util.Optional;
 
 import io.seqera.data.command.CommandState;
@@ -34,6 +35,26 @@ public interface CommandStateStore {
      */
     void save(CommandState state);
 
+    /**
+     * Persist a command only if its id is not already present.
+     */
+    boolean create(CommandState state);
+
+    boolean tryAcquireAttempt(String commandId, String owner, Duration ttl);
+
+    boolean renewAttempt(String commandId, String owner, Duration ttl);
+
+    boolean isAttemptOwner(String commandId, String owner);
+
+    boolean releaseAttempt(String commandId, String owner);
+
+    /**
+     * Persist state only if the supplied attempt still owns the command lease.
+     *
+     * @return true when saved; false when the attempt has lost ownership
+     */
+    boolean saveOwned(CommandState state, String owner);
+
     /**
      * Find a command state by ID.
      *
diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStoreImpl.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStoreImpl.java
index 43c55d79..98f0dd25 100644
--- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStoreImpl.java
+++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStoreImpl.java
@@ -38,11 +38,14 @@
 public class CommandStateStoreImpl extends AbstractStateStore<CommandState> implements CommandStateStore {
 
     private static final String PREFIX = "cmd-state/v1";
+    private static final String ATTEMPT_PREFIX = "cmd-attempt/v1:";
 
     private final Duration ttl;
+    private final StateProvider<String, String> provider;
 
     public CommandStateStoreImpl(StateProvider<String, String> provider, StringEncodingStrategy<CommandState> encodingStrategy, Duration ttl) {
         super(provider, encodingStrategy);
+        this.provider = provider;
         this.ttl = ttl;
     }
 
@@ -66,4 +69,39 @@ public void save(CommandState state) {
         put(state.id(), state);
     }
 
+    @Override
+    public boolean create(CommandState state) {
+        return putIfAbsent(state.id(), state, ttl);
+    }
+
+    @Override
+    public boolean tryAcquireAttempt(String commandId, String owner, Duration lease) {
+        return provider.putIfAbsent(ATTEMPT_PREFIX + commandId, owner, lease);
+    }
+
+    @Override
+    public boolean renewAttempt(String commandId, String owner, Duration lease) {
+        return provider.compareAndSet(ATTEMPT_PREFIX + commandId, owner, owner, lease);
+    }
+
+    @Override
+    public boolean isAttemptOwner(String commandId, String owner) {
+        return owner.equals(provider.get(ATTEMPT_PREFIX + commandId));
+    }
+
+    @Override
+    public boolean releaseAttempt(String commandId, String owner) {
+        return provider.compareAndDelete(ATTEMPT_PREFIX + commandId, owner);
+    }
+
+    @Override
+    public boolean saveOwned(CommandState state, String owner) {
+        return provider.putIfOwner(
+                ATTEMPT_PREFIX + state.id(),
+                owner,
+                key0(state.id()),
+                serialize(state),
+                ttl);
+    }
+
 }
diff --git a/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandServiceRedisE2ETest.groovy b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandServiceRedisE2ETest.groovy
new file mode 100644
index 00000000..794bec25
--- /dev/null
+++ b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandServiceRedisE2ETest.groovy
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2026, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.seqera.data.command
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+
+import io.micronaut.context.ApplicationContext
+import io.seqera.fixtures.redis.RedisTestContainer
+import spock.lang.Specification
+import spock.util.concurrent.PollingConditions
+
+class CommandServiceRedisE2ETest extends Specification implements RedisTestContainer {
+
+    def 'should execute a long command once across two replicas and persist terminal state'() {
+        given:
+        def properties = ['redis.uri': "redis://${redisHostName}:${redisPort}".toString()]
+        def ctxA = ApplicationContext.run(properties, 'test', 'redis')
+        def ctxB = ApplicationContext.run(properties, 'test', 'redis')
+        def serviceA = ctxA.getBean(CommandService)
+        def serviceB = ctxB.getBean(CommandService)
+        def calls = new AtomicInteger()
+        def started = new CountDownLatch(1)
+        def release = new CountDownLatch(1)
+        def handler = new RedisOnceHandler(calls, started, release)
+        def id = "redis-e2e-${UUID.randomUUID()}"
+        def command = new TestCommand(id, 'redis-once', new TestParams(1, 'redis'))
+
+        when:
+        serviceA.registerHandler(handler)
+        serviceB.registerHandler(handler)
+        serviceA.start()
+        serviceB.start()
+        serviceA.submit(command)
+        serviceB.submit(command) // duplicate submission is transport-level at-least-once
+
+        then:
+        started.await(5, TimeUnit.SECONDS)
+
+        when: 'the handler remains active for longer than the one-second claim timeout'
+        sleep 2_500
+
+        then: 'neither the Stream lease nor the command-state guard permits overlap'
+        calls.get() == 1
+
+        when:
+        release.countDown()
+
+        then:
+        new PollingConditions(timeout: 8).eventually {
+            assert serviceA.getState(id).orElseThrow().status() == CommandStatus.SUCCEEDED
+        }
+        calls.get() == 1
+
+        cleanup:
+        release.countDown()
+        serviceA?.stop()
+        serviceB?.stop()
+        ctxA?.close()
+        ctxB?.close()
+    }
+}
+
+class RedisOnceHandler implements CommandHandler<TestParams, TestResult> {
+    private final AtomicInteger calls
+    private final CountDownLatch started
+    private final CountDownLatch release
+
+    RedisOnceHandler(AtomicInteger calls, CountDownLatch started, CountDownLatch release) {
+        this.calls = calls
+        this.started = started
+        this.release = release
+    }
+
+    @Override
+    String type() { 'redis-once' }
+
+    @Override
+    CommandResult<TestResult> execute(Command<TestParams> command) {
+        calls.incrementAndGet()
+        started.countDown()
+        release.await(10, TimeUnit.SECONDS)
+        return CommandResult.success(new TestResult('done', command.params().value))
+    }
+}
diff --git a/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/store/CommandAttemptStoreTest.groovy b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/store/CommandAttemptStoreTest.groovy
new file mode 100644
index 00000000..40d3b1e3
--- /dev/null
+++ b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/store/CommandAttemptStoreTest.groovy
@@ -0,0 +1,88 @@
+/*
+ * Copyright 2026, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.seqera.data.command.store
+
+import java.time.Duration
+
+import io.seqera.data.command.CommandState
+import io.seqera.data.store.state.impl.LocalStateProvider
+import io.seqera.serde.jackson.JacksonEncodingStrategy
+import spock.lang.Specification
+
+class CommandAttemptStoreTest extends Specification {
+
+    private CommandStateStoreImpl newStore() {
+        new CommandStateStoreImpl(
+                new LocalStateProvider(),
+                new JacksonEncodingStrategy<CommandState>() {},
+                Duration.ofHours(1))
+    }
+
+    def 'should create command state idempotently'() {
+        given:
+        def store = newStore()
+        def state = CommandState.submitted('c1', 'test', [value: 1])
+
+        expect:
+        store.create(state)
+        !store.create(state)
+    }
+
+    def 'should fence command attempts by owner and renew their lease'() {
+        given:
+        def store = newStore()
+
+        expect:
+        store.tryAcquireAttempt('c1', 'owner-a', Duration.ofMillis(200))
+        !store.tryAcquireAttempt('c1', 'owner-b', Duration.ofMillis(200))
+        store.isAttemptOwner('c1', 'owner-a')
+
+        and:
+        store.renewAttempt('c1', 'owner-a', Duration.ofSeconds(1))
+        !store.renewAttempt('c1', 'owner-b', Duration.ofSeconds(1))
+        !store.releaseAttempt('c1', 'owner-b')
+        store.releaseAttempt('c1', 'owner-a')
+        store.tryAcquireAttempt('c1', 'owner-b', Duration.ofSeconds(1))
+    }
+
+    def 'should recover an attempt after its owner lease expires'() {
+        given:
+        def store = newStore()
+
+        expect:
+        store.tryAcquireAttempt('c1', 'dead-owner', Duration.ofMillis(50))
+
+        when:
+        sleep 100
+
+        then:
+        store.tryAcquireAttempt('c1', 'live-owner', Duration.ofSeconds(1))
+    }
+
+    def 'should reject a stale owner state update'() {
+        given:
+        def store = newStore()
+        def submitted = CommandState.submitted('c1', 'test', [value: 1])
+        store.create(submitted)
+        store.tryAcquireAttempt('c1', 'owner-a', Duration.ofSeconds(1))
+
+        expect:
+        !store.saveOwned(submitted.failed('stale'), 'owner-b')
+        store.findById('c1').get() == submitted
+        store.saveOwned(submitted.submitting(), 'owner-a')
+        store.findById('c1').get().status().name() == 'SUBMITTING'
+    }
+}
diff --git a/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestRedisWorkQueueConfig.java b/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestRedisWorkQueueConfig.java
new file mode 100644
index 00000000..9839d245
--- /dev/null
+++ b/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestRedisWorkQueueConfig.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2026, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.seqera.data.command;
+
+import java.time.Duration;
+
+import io.micronaut.context.annotation.Requires;
+import io.seqera.data.workqueue.redis.RedisWorkQueueConfig;
+import jakarta.inject.Singleton;
+
+@Singleton
+@Requires(env = "test")
+public class TestRedisWorkQueueConfig implements RedisWorkQueueConfig {
+    @Override
+    public String getDefaultConsumerGroupName() {
+        return "command-queue-test";
+    }
+
+    @Override
+    public Duration getVisibilityTimeout() {
+        return Duration.ofSeconds(1);
+    }
+
+    @Override
+    public Duration getConsumerWarnTimeout() {
+        return Duration.ofSeconds(5);
+    }
+}
diff --git a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/LocalStateProvider.groovy b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/LocalStateProvider.groovy
index 64b96ab4..6b8cd633 100644
--- a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/LocalStateProvider.groovy
+++ b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/LocalStateProvider.groovy
@@ -94,6 +94,30 @@ class LocalStateProvider implements StateProvider<String,String> {
         return putIfAbsent0(key, value, ttl) == null
     }
 
+    @Override
+    synchronized boolean compareAndSet(String key, String expected, String value, Duration ttl) {
+        if( get(key) != expected )
+            return false
+        store.put(key, new Entry<>(value, ttl))
+        return true
+    }
+
+    @Override
+    synchronized boolean compareAndDelete(String key, String expected) {
+        if( get(key) != expected )
+            return false
+        store.remove(key)
+        return true
+    }
+
+    @Override
+    synchronized boolean putIfOwner(String ownerKey, String owner, String key, String value, Duration ttl) {
+        if( get(ownerKey) != owner )
+            return false
+        store.put(key, new Entry<>(value, ttl))
+        return true
+    }
+
     @Override
     synchronized CountResult<String> putJsonIfAbsentAndIncreaseCount(String key, String json, Duration ttl, CountParams counterKey, String luaScript) {
         final counter = counterKey.key + '/' + counterKey.field
diff --git a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/RedisStateProvider.groovy b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/RedisStateProvider.groovy
index a381035f..87e00a61 100644
--- a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/RedisStateProvider.groovy
+++ b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/RedisStateProvider.groovy
@@ -39,6 +39,15 @@ import redis.clients.jedis.params.SetParams
 @CompileStatic
 class RedisStateProvider implements StateProvider<String,String> {
 
+    private static final String COMPARE_SET_SCRIPT =
+            "if redis.call('GET', KEYS[1]) == ARGV[1] then redis.call('SET', KEYS[1], ARGV[2], 'PX', ARGV[3]); return 1 else return 0 end"
+
+    private static final String COMPARE_DELETE_SCRIPT =
+            "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end"
+
+    private static final String PUT_IF_OWNER_SCRIPT =
+            "if redis.call('GET', KEYS[1]) == ARGV[1] then redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3]); return 1 else return 0 end"
+
     @Inject
     private JedisPool pool
 
@@ -80,6 +89,30 @@ class RedisStateProvider implements StateProvider<String,String> {
         }
     }
 
+    @Override
+    boolean compareAndSet(String key, String expected, String value, Duration ttl) {
+        try( Jedis conn=pool.getResource() ) {
+            final result = conn.eval(COMPARE_SET_SCRIPT, 1, key, expected, value, ttl.toMillis().toString())
+            return result == 1L
+        }
+    }
+
+    @Override
+    boolean compareAndDelete(String key, String expected) {
+        try( Jedis conn=pool.getResource() ) {
+            final result = conn.eval(COMPARE_DELETE_SCRIPT, 1, key, expected)
+            return result == 1L
+        }
+    }
+
+    @Override
+    boolean putIfOwner(String ownerKey, String owner, String key, String value, Duration ttl) {
+        try( Jedis conn=pool.getResource() ) {
+            final result = conn.eval(PUT_IF_OWNER_SCRIPT, 2, ownerKey, key, owner, value, ttl.toMillis().toString())
+            return result == 1L
+        }
+    }
+
     /*
      * Set a value only the specified key does not exists, if the value can be set
      * the counter identified by the key provided via 'KEYS[2]' is incremented by 1,
diff --git a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/StateProvider.groovy b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/StateProvider.groovy
index a528a71d..e09295e7 100644
--- a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/StateProvider.groovy
+++ b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/StateProvider.groovy
@@ -29,6 +29,16 @@ import io.seqera.data.store.state.StateStore
  */
 interface StateProvider<K,V> extends StateStore<K,V> {
 
+    boolean compareAndSet(K key, V expected, V value, Duration ttl)
+
+    boolean compareAndDelete(K key, V expected)
+
+    /**
+     * Store a value only while an ownership key still contains the expected owner.
+     * The ownership check and value update must be atomic.
+     */
+    boolean putIfOwner(K ownerKey, V owner, K key, V value, Duration ttl)
+
     /**
      * Store a value in the cache only if does not exist. If the operation is successful
      * the counter identified by the key specified is incremented by 1 and the counter (new)
diff --git a/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/LocalStateProviderTest.groovy b/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/LocalStateProviderTest.groovy
index 32ec8257..628e842b 100644
--- a/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/LocalStateProviderTest.groovy
+++ b/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/LocalStateProviderTest.groovy
@@ -179,4 +179,17 @@ class LocalStateProviderTest extends Specification {
         provider.get(k) == null
     }
 
+    def 'should update a value only for the current owner'() {
+        given:
+        def ownerKey = UUID.randomUUID().toString()
+        def valueKey = UUID.randomUUID().toString()
+        provider.put(ownerKey, 'owner-a', Duration.ofSeconds(1))
+
+        expect:
+        !provider.putIfOwner(ownerKey, 'owner-b', valueKey, 'stale', Duration.ofSeconds(1))
+        provider.get(valueKey) == null
+        provider.putIfOwner(ownerKey, 'owner-a', valueKey, 'current', Duration.ofSeconds(1))
+        provider.get(valueKey) == 'current'
+    }
+
 }
diff --git a/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/RedisStateProviderTest.groovy b/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/RedisStateProviderTest.groovy
index 13ab40d9..410700d3 100644
--- a/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/RedisStateProviderTest.groovy
+++ b/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/RedisStateProviderTest.groovy
@@ -191,5 +191,17 @@ class RedisStateProviderTest extends Specification implements RedisTestContainer
         result.count == 3
     }
 
+    def 'should update a value only for the current owner'() {
+        given:
+        def ownerKey = UUID.randomUUID().toString()
+        def valueKey = UUID.randomUUID().toString()
+        provider.put(ownerKey, 'owner-a', Duration.ofSeconds(1))
+
+        expect:
+        !provider.putIfOwner(ownerKey, 'owner-b', valueKey, 'stale', Duration.ofSeconds(1))
+        provider.get(valueKey) == null
+        provider.putIfOwner(ownerKey, 'owner-a', valueKey, 'current', Duration.ofSeconds(1))
+        provider.get(valueKey) == 'current'
+    }
 
 }
diff --git a/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueue.java b/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueue.java
index ba56d62c..7f79be18 100644
--- a/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueue.java
+++ b/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueue.java
@@ -18,6 +18,7 @@
 package io.seqera.data.workqueue.redis;
 
 import java.time.Duration;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -37,7 +38,6 @@
 import redis.clients.jedis.StreamEntryID;
 import redis.clients.jedis.exceptions.JedisDataException;
 import redis.clients.jedis.params.XAutoClaimParams;
-import redis.clients.jedis.params.XClaimParams;
 import redis.clients.jedis.params.XReadGroupParams;
 import redis.clients.jedis.resps.StreamEntry;
 
@@ -86,6 +86,17 @@ public class RedisWorkQueue implements WorkQueue<String> {
 
     private static final String DATA_FIELD = "data";
 
+    /**
+     * Touch a pending entry only if this consumer still owns it. Without the ownership
+     * check, a stale handler could XCLAIM the entry back after a peer reclaimed it.
+     * The script uses Stream/PEL commands only.
+     */
+    private static final String RENEW_IF_OWNER_SCRIPT =
+            "local p=redis.call('XPENDING',KEYS[1],ARGV[1],ARGV[2],ARGV[2],1); " +
+            "if #p==1 and p[1][2]==ARGV[3] then " +
+            "return redis.call('XCLAIM',KEYS[1],ARGV[1],ARGV[3],0,ARGV[2],'JUSTID') " +
+            "else return {} end";
+
     @Inject
     private JedisPool pool;
 
@@ -149,12 +160,20 @@ public void offer(String queueId, String message) {
      * The returned lease id is the Redis {@link StreamEntryID} of the delivered entry.
      */
     @Override
-    public Lease<String> receive(String queueId) {
+    public Lease<String> receiveNew(String queueId) {
         try (Jedis jedis = pool.getResource()) {
-            StreamEntry entry = claimMessage(jedis, queueId);
+            final StreamEntry entry = readMessage(jedis, queueId);
             if (entry == null) {
-                entry = readMessage(jedis, queueId);
+                return null;
             }
+            return new Lease<>(entry.getID().toString(), entry.getFields().get(DATA_FIELD));
+        }
+    }
+
+    @Override
+    public Lease<String> reclaim(String queueId) {
+        try (Jedis jedis = pool.getResource()) {
+            final StreamEntry entry = claimMessage(jedis, queueId);
             if (entry == null) {
                 return null;
             }
@@ -173,13 +192,10 @@ public Lease<String> receive(String queueId) {
     @Override
     public void renewLease(String queueId, String leaseId) {
         try (Jedis jedis = pool.getResource()) {
-            jedis.xclaimJustId(
-                    queueId,
-                    config.getDefaultConsumerGroupName(),
-                    consumerName,
-                    0L,
-                    XClaimParams.xClaimParams(),
-                    new StreamEntryID(leaseId));
+            jedis.eval(
+                    RENEW_IF_OWNER_SCRIPT,
+                    Collections.singletonList(queueId),
+                    List.of(config.getDefaultConsumerGroupName(), leaseId, consumerName));
         }
     }
 
@@ -205,12 +221,13 @@ public void ack(String queueId, String leaseId) {
     /**
      * {@inheritDoc}
      *
-     * <p>No-op: the entry remains in the pending-entries list and becomes reclaimable
-     * by a peer consumer once its idle time exceeds the visibility timeout.
+     * <p>Touches the entry one final time and leaves it in the pending-entries list.
+     * It becomes reclaimable after a full visibility timeout without retaining any
+     * Java-side lifecycle state.
      */
     @Override
     public void release(String queueId, String leaseId) {
-        // no-op: entry stays in the PEL, reclaimable after the visibility timeout
+        renewLease(queueId, leaseId);
     }
 
     /**
diff --git a/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueueConfig.java b/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueueConfig.java
index 57207a2e..18088572 100644
--- a/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueueConfig.java
+++ b/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueueConfig.java
@@ -119,10 +119,9 @@ default long getHeartbeatIntervalMillis() {
     }
 
     /**
-     * Returns the upper bound on a single {@code accept()} invocation before its
-     * lease is released (safety valve). This bounds one handler invocation, not the
-     * total lease lifetime; past this bound the heartbeat daemon stops renewing the
-     * lease so it becomes reclaimable. Defaults to {@code 15m}.
+     * Returns the duration after which a still-running {@code accept()} invocation is
+     * reported as stalled. Renewal continues while the handler is active to prevent
+     * timeout-driven overlap. Defaults to {@code 15m}.
      *
      * @return the maximum single-invocation processing time duration
      */
diff --git a/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueRedisTest.groovy b/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueRedisTest.groovy
index 7c28ab06..8325d73c 100644
--- a/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueRedisTest.groovy
+++ b/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueRedisTest.groovy
@@ -19,6 +19,7 @@ package io.seqera.data.workqueue
 
 import java.time.Duration
 import java.util.concurrent.CountDownLatch
+import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicInteger
 
@@ -47,6 +48,48 @@ class AsyncWorkQueueRedisTest extends Specification implements RedisTestContaine
         return ApplicationContext.run('test', 'redis')
     }
 
+    def 'should admit all live messages while concurrency only bounds active invocations'() {
+        given:
+        def ctx = newContext()
+        def target = ctx.getBean(RedisWorkQueue)
+        def queue = new TunableQueue(target,
+                concurrency: 2,
+                pollInterval: Duration.ofMillis(50),
+                heartbeatInterval: Duration.ofMillis(300))
+        def id = "queue-${LongRndKey.rndHex()}"
+        def firstSeen = ConcurrentHashMap.<String>newKeySet()
+        def admitted = new CountDownLatch(20)
+        def active = new AtomicInteger()
+        def maxActive = new AtomicInteger()
+
+        when:
+        queue.addConsumer(id, { msg ->
+            def n = active.incrementAndGet()
+            maxActive.accumulateAndGet(n, Math::max)
+            try {
+                if (firstSeen.add(msg)) {
+                    admitted.countDown()
+                }
+                return false
+            }
+            finally {
+                active.decrementAndGet()
+            }
+        })
+        20.times { queue.offer(id, "task-$it".toString()) }
+
+        then: 'every task gets an invocation even though all remain non-terminal'
+        admitted.await(8, TimeUnit.SECONDS)
+        firstSeen.size() == 20
+
+        and: 'only handler calls, not live tasks, are bounded'
+        maxActive.get() <= 2
+
+        cleanup:
+        queue.close()
+        ctx.stop()
+    }
+
     // single instance — a handler running longer than visibility-timeout is not
     // reclaimed by this instance's own poll while it is alive/heartbeated
     def 'should not reclaim live work within a single instance' () {
@@ -165,9 +208,8 @@ class AsyncWorkQueueRedisTest extends Specification implements RedisTestContaine
         ctxLive.stop()
     }
 
-    // a single invocation exceeding max-processing-time has its lease released
-    // (stops being renewed), so the message is reclaimed and re-delivered
-    def 'should release the lease of an invocation exceeding max-processing-time' () {
+    // max-processing-time is diagnostic only: releasing a live handler would overlap it.
+    def 'should continue renewing beyond max-processing-time without overlapping execution' () {
         given:
         def ctx = newContext()
         def target = ctx.getBean(RedisWorkQueue)
@@ -179,25 +221,28 @@ class AsyncWorkQueueRedisTest extends Specification implements RedisTestContaine
         def id = "queue-${LongRndKey.rndHex()}"
         def calls = new AtomicInteger()
         def hang = new CountDownLatch(1)
-        def redelivered = new CountDownLatch(1)
+        def completed = new CountDownLatch(1)
 
         when:
         queue.addConsumer(id, { msg ->
-            def n = calls.incrementAndGet()
-            if (n == 1) {
-                // first invocation hangs past max-processing-time (1s) -> lease released
-                hang.await()
-                return true
-            }
-            // the re-delivered invocation completes normally
-            redelivered.countDown()
+            calls.incrementAndGet()
+            hang.await()
+            completed.countDown()
             return true
         })
         queue.offer(id, 'hung')
 
-        then: 'the stalled invocation is evicted and the message is re-delivered'
-        redelivered.await(10, TimeUnit.SECONDS)
-        calls.get() >= 2
+        then: 'the live invocation remains the sole owner well beyond both timeouts'
+        sleep 3_000
+        calls.get() == 1
+
+        when:
+        hang.countDown()
+
+        then:
+        completed.await(5, TimeUnit.SECONDS)
+        sleep 1_500
+        calls.get() == 1
 
         cleanup:
         hang.countDown()
diff --git a/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy b/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
index 4e2523a9..431fbaf7 100644
--- a/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
+++ b/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
@@ -24,7 +24,7 @@ import io.seqera.serde.encode.StringEncodingStrategy
 /**
  * A {@link AbstractWorkQueue} used by the async-processing tests. It carries a
  * String payload (identity encoding) and exposes the async knobs — concurrency,
- * poll interval, heartbeat interval and max-processing-time — as constructor options
+ * poll interval, heartbeat interval and handler-warning threshold — as constructor options
  * so each test can tune them independently.
  *
  * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
diff --git a/lib-data-workqueue/README.md b/lib-data-workqueue/README.md
index 8a1abc7f..58886304 100644
--- a/lib-data-workqueue/README.md
+++ b/lib-data-workqueue/README.md
@@ -56,7 +56,7 @@ stream key, e.g. `cmd-queue/v1`).
 |---|---|---|---|---|
 | `seqera.workqueue.entries` | Gauge | — | entries | Current queue backlog (Redis `XLEN`, polled at scrape time). |
 | `seqera.workqueue.messages` | Counter | `outcome` | messages | Total messages processed per outcome. |
-| `seqera.workqueue.processing` | Timer | `outcome` | seconds | Per-entry processing time. Includes the full lifecycle from the underlying `queue.consume(...)` entry through the consumer's `accept` and the Redis acknowledge/delete. Published as a Prometheus histogram (with buckets) so quantiles can be aggregated server-side across replicas via `histogram_quantile()`. |
+| `seqera.workqueue.processing` | Timer | `outcome` | seconds | Per-handler-invocation processing time. Published as a Prometheus histogram (with buckets) so quantiles can be aggregated server-side across replicas via `histogram_quantile()`. |
 
 The `outcome` tag takes one of three values:
 
@@ -154,11 +154,9 @@ workQueue.addConsumer("user-activity", new ActivityConsumer())
 
 ## Architecture
 
-`AbstractWorkQueue` runs handlers **asynchronously and concurrently** while
-guaranteeing that a given message is processed by exactly one *live* consumer at a
-time. A message is owned by its consumer for as long as the handler keeps working —
-independent of how long that takes — and ownership is relinquished only when the work
-finishes or the consumer dies.
+`AbstractWorkQueue` is an at-least-once transport. `concurrency()` bounds active
+handler invocations, not the number of live messages. A non-terminal message consumes
+no Java thread or semaphore permit between invocations.
 
 ```
   offer(msg)                                    ┌──────────────────────────────┐
@@ -171,11 +169,11 @@ finishes or the consumer dies.
  │  group)  │◀──────────── heartbeat daemon ────┤     (never runs it inline)    │
  │          │      every visibility-timeout/3    │                               │
  │          │   ack (XACK + XDEL)                │  worker (executor thread)     │
- │          │◀──────────── on terminal ─────────┤   accept(msg):                │
- └──────────┘                                   │    ├─ true  → ack + free slot │
-      ▲                                         │    └─ false → keep lease,     │
-      │ reclaimed by a peer only if the owner   │       re-run after pollInterval│
-      │ dies (heartbeat stops → idle > visibility-timeout) via the re-poll sched │
+ │          │◀──────────── on terminal ─────────┤   one accept(msg) call:       │
+ └──────────┘                                   │    ├─ true  → ack             │
+      ▲                                         │    └─ false → leave in PEL    │
+      │ retry/dead-owner reclaim after          │  then always free the slot    │
+      │ idle > visibility-timeout               │                               │
       └─────────────────────────────────────────────────────────────────────────┘
 ```
 
@@ -185,27 +183,23 @@ finishes or the consumer dies.
    handler; it hands each message to a worker executor and moves on. Handlers run on the
    executor supplied via `withHandlerExecutor(...)` — **mandatory, no default** (Micronaut
    consumers inject the `@Named(BLOCKING)` executor). A `Semaphore` sized by
-   `concurrency()` bounds how many messages are in flight at once (backpressure: excess
-   messages stay in the queue).
+   `concurrency()` bounds how many handler calls run at once.
 
 2. **Heartbeat lease (single live runner + safe long handlers).** While a message is in
    flight, a daemon renews its Redis consumer-group entry (`XCLAIM … JUSTID`) every
    `visibility-timeout / 3`, pinning its idle time near zero so no peer's `XAUTOCLAIM` can
    reclaim it — no matter how long the handler runs. If the owning process dies, the
    heartbeat stops, idle time crosses the visibility timeout, and a peer reclaims the message
-   (real dead-consumer failover). A `max-processing-time` safety valve stops renewing a
-   single invocation that runs pathologically long (logged as *stalled*), without
-   interrupting its thread.
-
-3. **In-process re-poll for not-yet-terminal work.** When a handler returns `false` (work
-   in progress), the message keeps its lease and the handler is **re-invoked in-process**
-   after `pollInterval` via a scheduler — Redis is not re-read. This makes the re-poll
-   cadence independent of `visibility-timeout` (which then governs only failover). The next
-   invocation is scheduled only after the previous one returns, so a given message is
-   never processed by two overlapping invocations.
-
-Delivery is **at-least-once** (a crash/pause beyond `visibility-timeout`, or the
-`max-processing-time` valve, can hand a still-running message to a peer), so consumers
+   (real dead-consumer failover). `max-processing-time` is a warning threshold only:
+   renewal continues while the handler thread is active, preventing timeout-driven overlap.
+
+3. **Stream-owned retry.** When a handler returns `false`, the entry is touched once and
+   left in the Redis PEL. Its thread and permit are released immediately. A later
+   `XAUTOCLAIM` starts the next invocation after `visibility-timeout`. New delivery and
+   expired reclaim are selected with a weighted policy so retries cannot freeze intake.
+
+Delivery is **at-least-once** (a crash, partition, or pause beyond
+`visibility-timeout` can hand a still-running message to a peer), so consumers
 must be idempotent. The in-memory `LocalWorkQueue` has no pending-entries list, so it
 has no lease/heartbeat (renewLease is a no-op); it still benefits from async, concurrent dispatch.
 
@@ -213,11 +207,12 @@ has no lease/heartbeat (renewLease is a no-op); it still benefits from async, co
 
 | Knob | Where | Default | Governs |
 |---|---|---|---|
-| `pollInterval()` | `AbstractWorkQueue` | — (subclass) | Idle backoff **and** in-process re-poll cadence |
-| `concurrency()` | `AbstractWorkQueue` | `1` | Max in-flight messages (semaphore ceiling) |
-| `getVisibilityTimeout()` | `RedisWorkQueueConfig` | — | Dead-consumer failover window |
+| `pollInterval()` | `AbstractWorkQueue` | — (subclass) | Idle dispatcher backoff |
+| `concurrency()` | `AbstractWorkQueue` | `1` | Max concurrent handler invocations |
+| `newToRetryRatio()` | `AbstractWorkQueue` | `3` | Fair selection between new and expired entries |
+| `getVisibilityTimeout()` | `RedisWorkQueueConfig` | — | Retry cadence and dead-consumer failover |
 | `getHeartbeatInterval()` | `RedisWorkQueueConfig` | `visibility-timeout / 3` | Lease renewal cadence |
-| `getMaxProcessingTime()` | `RedisWorkQueueConfig` | `15m` | Upper bound on a single `accept()` before its lease is released |
+| `getMaxProcessingTime()` | `RedisWorkQueueConfig` | `15m` | Long-handler warning threshold |
 
 ## Testing
 
diff --git a/lib-data-workqueue/changelog.txt b/lib-data-workqueue/changelog.txt
index b7755cf5..0920bb3f 100644
--- a/lib-data-workqueue/changelog.txt
+++ b/lib-data-workqueue/changelog.txt
@@ -1,5 +1,13 @@
 # lib-data-workqueue changelog
 
+Next
+- Concurrency now bounds active handler invocations rather than live message lifecycles.
+- Non-terminal entries are returned to the backend after each invocation; Redis entries
+  remain in the PEL and are reclaimed after visibility timeout.
+- Heartbeats exist only while a handler runs, allowing execution beyond visibility timeout.
+- max-processing-time is diagnostic and never releases a still-running handler.
+- Split new delivery from expired reclaim and select them fairly so retries cannot starve intake.
+
 1.0.0 - 11 Jul 2026
 - Initial release. This module is the redis-free split of lib-data-stream-redis 1.6.0,
   renamed to reflect the reliable work-queue semantics it actually implements (competing
diff --git a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/AbstractWorkQueue.java b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/AbstractWorkQueue.java
index 760f0fc3..42662ced 100644
--- a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/AbstractWorkQueue.java
+++ b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/AbstractWorkQueue.java
@@ -127,10 +127,8 @@ public abstract class AbstractWorkQueue<M> implements Closeable {
     private final String name0;
 
     /**
-     * A message picked up from a queue and held while it is processed. The
-     * {@code queueId} + {@code leaseId} pair identifies the delivered entry; the
-     * {@code message} is kept so a not-yet-terminal command can be re-invoked in-process
-     * (Model B) without re-reading it from the queue.
+     * One active handler invocation. It exists only while {@code accept()} is running,
+     * never for the full lifecycle of a non-terminal message.
      */
     private record InFlight(String queueId, String leaseId, String message) {
         String key() {
@@ -139,17 +137,14 @@ String key() {
     }
 
     /**
-     * Leases held from pickup to terminal/crash; every entry is heartbeated by the
-     * daemon so an alive consumer is never reclaimed. Keyed by {@code queueId|leaseId}.
+     * Handler invocations currently running. Every entry is heartbeated until that one
+     * invocation returns, allowing handler time to exceed the queue visibility timeout.
      */
-    private final Map<String, InFlight> inFlight = new ConcurrentHashMap<>();
+    private final Map<String, InFlight> active = new ConcurrentHashMap<>();
 
-    /**
-     * Subset of {@link #inFlight} whose {@code accept()} invocation is running right now,
-     * mapped to the wall-clock millis at which that invocation started. Used by the
-     * heartbeat daemon to enforce {@code max-processing-time} on a single invocation.
-     */
-    private final Map<String, Long> active = new ConcurrentHashMap<>();
+    private final Map<String, Long> activeSince = new ConcurrentHashMap<>();
+
+    private final java.util.Set<String> warned = ConcurrentHashMap.newKeySet();
 
     /**
      * Executor that runs the message handlers. Supplied by the consumer via
@@ -161,25 +156,21 @@ String key() {
     private volatile ExecutorService pool;
 
     /**
-     * Gates new intake: a permit is acquired when a lease is picked up and held for the
-     * whole lease lifetime (across re-polls), released on terminal ack / eviction /
-     * release. This bounds concurrent handlers to {@link #concurrency()} and reserves
-     * capacity so in-flight commands' re-polls are never starved by new intake.
+     * Bounds handler invocations, not live messages. A permit is held only from delivery
+     * until one {@code accept()} call returns.
      */
     private volatile Semaphore slots;
 
     /**
-     * Schedules delayed re-poll re-submissions for not-yet-terminal commands (Model B).
-     */
-    private volatile ScheduledExecutorService scheduler;
-
-    /**
-     * Renews every in-flight lease on a fixed cadence so an alive consumer keeps ownership.
+     * Renews every active invocation on a fixed cadence so a handler may safely run longer
+     * than the visibility timeout.
      */
     private volatile ScheduledExecutorService heartbeat;
 
     private volatile boolean closed;
 
+    private final AtomicInteger deliverySequence = new AtomicInteger();
+
     /**
      * Constructs a new queue without metrics instrumentation. Behavior is identical
      * to passing {@link NoopQueueMetrics#INSTANCE} to {@link #AbstractWorkQueue(WorkQueue, QueueMetrics)}.
@@ -225,21 +216,28 @@ protected Thread createListenerThread() {
     /**
      * @return
      *      The time interval to await before trying to read again the queue
-     *      when no more entries are available. Also the cadence at which a
-     *      not-yet-terminal command is re-invoked in-process (Model B).
+     *      when no more entries are available. Redis retry cadence is governed
+     *      by the visibility timeout of the pending Stream entry.
      */
     protected abstract Duration pollInterval();
 
     /**
      * @return
-     *      The maximum number of message handlers that may run concurrently on this
-     *      instance (the worker pool size). Defaults to {@code 1}; subclasses may
-     *      override to enable parallel processing.
+     *      The maximum number of handler invocations that may run concurrently on this
+     *      instance. A non-terminal message releases its permit after each invocation.
      */
     protected int concurrency() {
         return 1;
     }
 
+    /**
+     * Number of new-message delivery opportunities for each expired-message reclaim.
+     * Both paths fall back to the other when empty, so neither intake nor retry can starve.
+     */
+    protected int newToRetryRatio() {
+        return 3;
+    }
+
     /**
      * @return
      *      How often in-flight leases are renewed so an alive consumer keeps ownership
@@ -254,9 +252,9 @@ protected Duration heartbeatInterval() {
 
     /**
      * @return
-     *      The upper bound on a single {@code accept()} invocation before its lease is
-     *      released (safety valve); it does not interrupt the handler thread. Defaults
-     *      to {@code 15m}.
+     *      Duration after which a still-running invocation is reported as stalled.
+     *      This is an observability threshold only: the lease continues to be renewed,
+     *      because releasing it while the handler runs would create overlapping execution.
      */
     protected Duration maxProcessingTime() {
         final Duration d = queue.maxProcessingTime();
@@ -325,17 +323,15 @@ public void addConsumer(String queueId, MessageConsumer<M> consumer) {
     }
 
     /**
-     * Lazily create the worker pool, the re-poll scheduler, the heartbeat daemon and the
-     * capacity gate, then start the dispatcher thread. Invoked once, when the first
+     * Lazily create the heartbeat daemon and invocation-capacity gate, then start the
+     * dispatcher thread. Invoked once, when the first
      * consumer is registered.
      */
     private void startProcessing() {
         // a handler executor must be supplied via withHandlerExecutor() before processing starts
         Objects.requireNonNull(pool, "Handler executor not set - call withHandlerExecutor() before addConsumer()");
-        // 'slots' — not the executor — bounds how many commands may be in flight at once;
-        // the cap is a memory/heartbeat ceiling, independent of the executor's threading model.
+        // slots bound active handler calls, never the number of live queue entries
         this.slots = new Semaphore(Math.max(1, concurrency()));
-        this.scheduler = new ScheduledThreadPoolExecutor(1, daemonFactory(name() + "-repoll-" + count.get()));
         this.heartbeat = new ScheduledThreadPoolExecutor(1, daemonFactory(name() + "-heartbeat-" + count.get()));
         final long hb = heartbeatInterval().toMillis();
         this.heartbeat.scheduleAtFixedRate(this::heartbeatTick, hb, hb, TimeUnit.MILLISECONDS);
@@ -383,9 +379,8 @@ protected boolean processMessage(String msg, MessageConsumer<M> consumer) {
 
     /**
      * The dispatcher loop (runs on the listener thread). It never runs a handler itself:
-     * for every queue that has free pool capacity it polls one message (without acking)
-     * and submits its processing to the worker pool, then sleeps for {@link #pollInterval()}
-     * when nothing was polled this cycle.
+     * for every queue that has free invocation capacity it selects fairly between new
+     * messages and expired pending messages, then submits one handler call to the pool.
      */
     protected void processMessages() {
         log.trace("Work queue - starting dispatcher thread");
@@ -393,9 +388,7 @@ protected void processMessages() {
             try {
                 boolean polled = false;
                 for (Map.Entry<String, MessageConsumer<M>> entry : listeners.entrySet()) {
-                    // poll a queue only when a worker slot is free (backpressure); the
-                    // permit is held for the whole lease lifetime so re-polls of in-flight
-                    // commands are never starved by new intake
+                    // A permit is held for one handler invocation only.
                     if (!slots.tryAcquire()) {
                         break;
                     }
@@ -425,8 +418,8 @@ protected void processMessages() {
     }
 
     /**
-     * Poll a single queue (a worker permit has already been acquired by the caller) and,
-     * if a message is available, register it as in-flight and submit it to the pool.
+     * Poll a single queue (an invocation permit has already been acquired by the caller)
+     * and submit one delivery to the pool.
      * If nothing is available the permit is released and {@code false} is returned.
      *
      * @return {@code true} if a message was polled and submitted, {@code false} otherwise
@@ -434,21 +427,22 @@ protected void processMessages() {
     private boolean dispatchOne(String queueId) {
         boolean submitted = false;
         try {
-            final WorkQueue.Lease<String> lease = queue.receive(queueId);
+            final WorkQueue.Lease<String> lease = receiveFair(queueId);
             if (lease == null) {
                 metrics.recordOutcome(metrics.startSample(), queueId, Outcome.EMPTY);
                 return false;
             }
             final var e = new InFlight(queueId, lease.id(), lease.message());
-            // Guard against self-reclaim: if the heartbeat falls behind by more than the
-            // visibility timeout, this instance's own receive() (XAUTOCLAIM) can re-deliver an
-            // entry it is already processing. The reclaim only refreshed the lease idle time, so
-            // keep the live in-flight entry and drop the duplicate — otherwise a second handler
-            // runs concurrently and its permit leaks (the original remove() returns null).
-            if (inFlight.putIfAbsent(e.key(), e) != null) {
+            // Guard against a local self-reclaim after a delayed heartbeat.
+            if (active.putIfAbsent(e.key(), e) != null) {
                 return false;   // 'submitted' stays false → finally releases this permit
             }
-            submitRun(e);
+            activeSince.put(e.key(), System.currentTimeMillis());
+            if (!submitRun(e)) {
+                activeSince.remove(e.key());
+                active.remove(e.key());
+                return false;
+            }
             submitted = true;
             return true;
         }
@@ -461,33 +455,52 @@ private boolean dispatchOne(String queueId) {
         }
     }
 
+    private WorkQueue.Lease<String> receiveFair(String queueId) {
+        final int ratio = Math.max(1, newToRetryRatio());
+        final boolean preferNew = Math.floorMod(deliverySequence.getAndIncrement(), ratio + 1) < ratio;
+        WorkQueue.Lease<String> result = preferNew
+                ? queue.receiveNew(queueId)
+                : queue.reclaim(queueId);
+        if (result == null) {
+            result = preferNew
+                    ? queue.reclaim(queueId)
+                    : queue.receiveNew(queueId);
+        }
+        return result;
+    }
+
     /**
      * Submit the processing of an in-flight lease to the worker pool. Swallows the
      * rejection that occurs when the pool is being shut down.
      */
-    private void submitRun(InFlight e) {
+    private boolean submitRun(InFlight e) {
         try {
             pool.execute(() -> run(e));
+            return true;
         }
         catch (RejectedExecutionException ex) {
             log.debug("Work queue - worker pool rejected task for entry={} (shutting down)", e.key());
+            return false;
         }
     }
 
     /**
-     * Runs a single {@code accept()} invocation on a worker thread. On {@code true}
-     * (terminal) it acks the message and drops the lease; on {@code false} (Model B,
-     * not-yet-terminal) it keeps the lease in-flight and schedules the next invocation
-     * after {@link #pollInterval()} — strictly serial per command, since the next
-     * invocation is scheduled only after this one returned.
+     * Runs exactly one {@code accept()} invocation. A terminal result is acknowledged;
+     * a non-terminal result is touched once and left in the PEL for reclaim after the
+     * visibility timeout. Either outcome releases the invocation permit.
      */
     private void run(InFlight e) {
-        final boolean accepted = invokeHandler(e);
-        if (accepted) {
-            acknowledge(e);
+        try {
+            final boolean accepted = invokeHandler(e);
+            if (accepted) {
+                acknowledge(e);
+            }
+            else if (!closed) {
+                retryLater(e);
+            }
         }
-        else if (shouldRepoll(e)) {
-            scheduleRepoll(e);
+        finally {
+            finishAttempt(e);
         }
     }
 
@@ -501,7 +514,6 @@ private boolean invokeHandler(InFlight e) {
         final long sample = metrics.startSample();
         boolean accepted = false;
         Outcome outcome = Outcome.ACTIVE;
-        active.put(e.key(), System.currentTimeMillis());
         try {
             accepted = processMessage(e.message(), consumer);
             outcome = accepted ? Outcome.PROCESSED : Outcome.ACTIVE;
@@ -511,7 +523,6 @@ private boolean invokeHandler(InFlight e) {
             log.error("Work queue - error processing entry={} - cause: {}", e.key(), t.getMessage(), t);
         }
         finally {
-            active.remove(e.key());
             metrics.recordOutcome(sample, e.queueId(), outcome);
         }
         return accepted;
@@ -525,78 +536,62 @@ private void acknowledge(InFlight e) {
         catch (Throwable t) {
             log.error("Work queue - error acking entry={} - cause: {}", e.key(), t.getMessage(), t);
         }
-        finally {
-            releaseLease(e.key());
-        }
-    }
-
-    /** Whether a not-yet-terminal command should be re-polled: still owned and not shutting down. */
-    private boolean shouldRepoll(InFlight e) {
-        return !closed && inFlight.containsKey(e.key());
     }
 
     /**
-     * Keep the lease (the heartbeat keeps renewing it, so no reclaim/migration) and schedule
-     * the next in-process invocation after {@link #pollInterval()} — strictly serial, since
-     * it is scheduled only after the previous invocation returned.
+     * Reset the delivery idle time after a non-terminal result, then leave the entry in
+     * the Stream PEL. It becomes eligible for another invocation after visibility timeout.
      */
-    private void scheduleRepoll(InFlight e) {
+    private void retryLater(InFlight e) {
         try {
-            scheduler.schedule(() -> submitRun(e), pollInterval().toMillis(), TimeUnit.MILLISECONDS);
+            queue.release(e.queueId(), e.leaseId());
         }
-        catch (RejectedExecutionException ex) {
-            log.debug("Work queue - re-poll scheduler rejected entry={} (shutting down)", e.key());
+        catch (Throwable t) {
+            log.warn("Work queue - error touching retry entry={} - cause: {}", e.key(), t.getMessage());
         }
     }
 
-    /**
-     * Drop a lease from the in-flight set and free its capacity permit. This pair is the
-     * single invariant "a permit is held iff its key is in-flight"; returns {@code true} if
-     * this call performed the removal (so callers can log only a real eviction).
-     */
-    private boolean releaseLease(String key) {
-        if (inFlight.remove(key) != null) {
+    private void finishAttempt(InFlight e) {
+        final String key = e.key();
+        activeSince.remove(key);
+        warned.remove(key);
+        if (active.remove(key) != null) {
             slots.release();
-            return true;
         }
-        return false;
+        if (closed && active.isEmpty() && heartbeat != null) {
+            heartbeat.shutdown();
+        }
     }
 
     /**
-     * Heartbeat tick: renew every in-flight lease so an alive consumer keeps ownership,
-     * and release the lease of any single invocation that has exceeded
-     * {@link #maxProcessingTime()} (safety valve; does not interrupt the handler thread).
+     * Heartbeat every currently executing invocation. Long-running handlers are warned
+     * about, but never evicted while their thread remains active.
      */
     private void heartbeatTick() {
         final long now = System.currentTimeMillis();
         final long maxMillis = maxProcessingTime().toMillis();
-        for (InFlight e : inFlight.values()) {
+        for (InFlight e : active.values()) {
             final String key = e.key();
-            final long start = active.getOrDefault(key, now);
+            final long start = activeSince.getOrDefault(key, now);
             if (now - start > maxMillis) {
-                // a single invocation is stalled beyond the bound: stop renewing so the
-                // lease becomes reclaimable, and free its capacity permit
-                if (releaseLease(key)) {
-                    log.warn("Work queue - releasing lease of stalled entry={} after {} - reclaimable after visibility timeout",
+                if (warned.add(key)) {
+                    log.warn("Work queue - handler still active for entry={} after {}; continuing lease renewal to prevent overlap",
                             key, Duration.ofMillis(now - start));
                 }
             }
-            else {
-                try {
-                    queue.renewLease(e.queueId(), e.leaseId());
-                }
-                catch (Throwable t) {
-                    // swallow transient errors; the next tick retries
-                    log.warn("Work queue - error renewing lease for entry={} - cause: {}", key, t.getMessage());
-                }
+            try {
+                queue.renewLease(e.queueId(), e.leaseId());
+            }
+            catch (Throwable t) {
+                // swallow transient errors; the next tick retries
+                log.warn("Work queue - error renewing lease for entry={} - cause: {}", key, t.getMessage());
             }
         }
     }
 
     /**
-     * Shutdown orderly the queue: stop the dispatcher, cancel pending re-polls, drain
-     * the worker pool so active handlers finish and ack, release any remaining leases so
-     * they are redelivered, and finally stop the heartbeat daemon.
+     * Stop intake. Active handlers remain heartbeated until they finish; close never makes
+     * an entry reclaimable while its handler thread is still executing.
      */
     @Override
     public void close() {
@@ -612,26 +607,10 @@ public void close() {
         catch (Exception e) {
             log.debug("Unexpected error while terminating {} - cause: {}", name0, e.getMessage());
         }
-        // 2. cancel pending scheduled re-polls
-        if (scheduler != null) {
-            scheduler.shutdownNow();
-        }
-        // 3. the handler executor is shared / container-managed — not shut down here;
-        //    any active handler finishes on its own (short-lived) and acks
-        // 4. release any lease still held so it is redelivered without waiting for lapse
-        for (InFlight e : inFlight.values()) {
-            if (inFlight.remove(e.key()) != null) {
-                try {
-                    queue.release(e.queueId(), e.leaseId());
-                }
-                catch (Throwable t) {
-                    log.debug("Work queue - error releasing entry={} on shutdown - cause: {}", e.key(), t.getMessage());
-                }
-            }
-        }
-        // 5. stop the heartbeat daemon last (any remaining leases lapse -> peers reclaim)
-        if (heartbeat != null) {
-            heartbeat.shutdownNow();
+        // The handler executor is shared/container-managed. Keep the heartbeat daemon alive
+        // until every active invocation completes; finishAttempt shuts it down at that point.
+        if (heartbeat != null && active.isEmpty()) {
+            heartbeat.shutdown();
         }
     }
 
diff --git a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/LocalWorkQueue.java b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/LocalWorkQueue.java
index 5c1ea54c..0288b577 100644
--- a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/LocalWorkQueue.java
+++ b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/LocalWorkQueue.java
@@ -90,7 +90,7 @@ public void offer(String queueId, String message) {
      * the lease id is simply the message value itself (used to re-offer it on release).
      */
     @Override
-    public Lease<String> receive(String queueId) {
+    public Lease<String> receiveNew(String queueId) {
         final var message = delegate
                 .get(queueId)
                 .poll();
@@ -100,6 +100,11 @@ public Lease<String> receive(String queueId) {
         return new Lease<>(message, message);
     }
 
+    @Override
+    public Lease<String> reclaim(String queueId) {
+        return null;
+    }
+
     /**
      * {@inheritDoc}
      *
diff --git a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/WorkQueue.java b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/WorkQueue.java
index d1d4e63f..5fd7215a 100644
--- a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/WorkQueue.java
+++ b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/WorkQueue.java
@@ -162,15 +162,26 @@ default boolean consume(String queueId, MessageConsumer<M> consumer) {
     record Lease<M>(String id, M message) {}
 
     /**
-     * Receives one message (either newly delivered or reclaimed from a stalled consumer)
-     * <strong>without acknowledging</strong> it. The caller becomes responsible for
-     * eventually calling {@link #ack(String, String)} once processing terminates, or
-     * {@link #release(String, String)} to hand it back for later redelivery.
-     *
-     * @param queueId the unique identifier of the source queue; must not be null or empty
-     * @return a {@link Lease} for the delivered message, or {@code null} if none is available
+     * Receives one message that has never previously been delivered to the consumer group.
+     * The caller is responsible for acknowledging it or leaving it pending for retry.
+     */
+    Lease<M> receiveNew(String queueId);
+
+    /**
+     * Reclaims one pending message whose owner has stopped renewing it for longer than the
+     * implementation's visibility timeout.
      */
-    Lease<M> receive(String queueId);
+    Lease<M> reclaim(String queueId);
+
+    /**
+     * Receives one message, preferring an expired pending entry for compatibility with the
+     * original synchronous API. Asynchronous dispatchers should select explicitly between
+     * {@link #receiveNew(String)} and {@link #reclaim(String)} to provide intake fairness.
+     */
+    default Lease<M> receive(String queueId) {
+        final Lease<M> reclaimed = reclaim(queueId);
+        return reclaimed != null ? reclaimed : receiveNew(queueId);
+    }
 
     /**
      * Resets the idle time of the given lease (heartbeat), so that an alive consumer
@@ -216,12 +227,11 @@ default Duration heartbeatInterval() {
     }
 
     /**
-     * Upper bound on a single {@code accept()} invocation before its lease is released
-     * (safety valve); it does not interrupt the handler thread. Returns {@code null}
-     * when the implementation has no lease concept, in which case the caller uses its
-     * own default.
+     * Observability threshold after which an active {@code accept()} invocation is reported
+     * as stalled. It never expires or releases the lease while the handler is still running.
+     * Returns {@code null} when the implementation supplies no threshold.
      *
-     * @return the maximum single-invocation processing time, or {@code null}
+     * @return the warning threshold for one handler invocation, or {@code null}
      */
     default Duration maxProcessingTime() {
         return null;
diff --git a/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueLocalTest.groovy b/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueLocalTest.groovy
index 8884c125..04412d47 100644
--- a/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueLocalTest.groovy
+++ b/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueLocalTest.groovy
@@ -31,7 +31,7 @@ import spock.util.concurrent.PollingConditions
  * Async-processing behaviour of {@link AbstractWorkQueue} exercised over the
  * in-memory {@link LocalWorkQueue} backend, so these run WITHOUT Docker.
  *
- * Covers: non-blocking dispatch, concurrency, re-poll cadence, serial-per-command,
+ * Covers: non-blocking dispatch, invocation concurrency, local retry, serial-per-message,
  * backpressure, and concurrency==1 default.
  *
  * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
@@ -87,8 +87,8 @@ class AsyncWorkQueueLocalTest extends Specification {
         queue.close()
     }
 
-    // a not-yet-terminal command is re-invoked at ~pollInterval (Model B)
-    def 'should re-poll a not-yet-terminal command at poll interval' () {
+    // Local queues have no PEL/visibility timeout, so RETRY is immediately re-offered.
+    def 'should release an invocation slot before retrying a non-terminal message' () {
         given:
         def poll = Duration.ofMillis(300)
         def target = new LocalWorkQueue()
@@ -103,16 +103,12 @@ class AsyncWorkQueueLocalTest extends Specification {
             return timestamps.size() >= 5
         })
         queue.offer(id, 'running')
+        queue.offer(id, 'other')
 
         then:
         new PollingConditions(timeout: 10).eventually {
-            assert timestamps.size() == 5
+            assert timestamps.size() >= 5
         }
-        and:
-        def times = timestamps.toList()
-        def gaps = (1..<times.size()).collect { times[it] - times[it-1] }
-        // each re-poll gap is ~pollInterval (allow generous slack for scheduling)
-        gaps.every { it >= 150 && it <= 1_500 }
 
         cleanup:
         queue.close()
@@ -219,7 +215,8 @@ class AsyncWorkQueueLocalTest extends Specification {
         def target = [
                 init      : { String q -> },
                 offer     : { String q, String m -> },
-                receive   : { String q -> deliveries.getAndIncrement() < 2 ? new WorkQueue.Lease<String>('dup-id', 'payload') : null },
+                receiveNew: { String q -> deliveries.getAndIncrement() < 2 ? new WorkQueue.Lease<String>('dup-id', 'payload') : null },
+                reclaim   : { String q -> null },
                 renewLease: { String q, String id -> },
                 ack       : { String q, String id -> acks.incrementAndGet() },
                 release   : { String q, String id -> },
diff --git a/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy b/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
index 4e2523a9..431fbaf7 100644
--- a/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
+++ b/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
@@ -24,7 +24,7 @@ import io.seqera.serde.encode.StringEncodingStrategy
 /**
  * A {@link AbstractWorkQueue} used by the async-processing tests. It carries a
  * String payload (identity encoding) and exposes the async knobs — concurrency,
- * poll interval, heartbeat interval and max-processing-time — as constructor options
+ * poll interval, heartbeat interval and handler-warning threshold — as constructor options
  * so each test can tune them independently.
  *
  * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>