Message Delivery Reliability Fixes

Problem

Discord messages are dropped — both messages that arrive during harness restarts AND new messages that arrive after the harness is back online.

Root Causes

The harness polls Redis Streams via XREADGROUP using > (new-only) with a random consumer name per restart (harness-{uuid}, see pubsub/redis.rs:56). Four compounding issues cause message loss:

1. No PEL recovery (unacked messages stuck forever)

When the harness reads a message via >, it is assigned to consumer harness-{uuid1}. If the harness restarts before XACK, the message remains in the Pending Entries List (PEL) under the old consumer name. On restart, a new consumer harness-{uuid2} polls with > — which only returns messages never delivered to any consumer in the group. PEL messages are skipped. There is no XAUTOCLAIM in the Rust harness to reclaim them.

The Python outbox services already implement XAUTOCLAIM (see pkgs/bridge/redis_provider.py:127, DEFAULT_RECOVER_IDLE_MS = 30_000), but the Rust harness does not.

2. No startup backlog drain

XGROUP CREATE uses id 0-0, meaning the group’s last-delivered-id starts at the beginning of the stream. But once any consumer in the group has read a message (advancing the last-delivered-id), messages that arrived between the last successful poll and the restart may have already been “delivered” to the old consumer — they’re in the PEL, not “new”. The harness never reads with 0 (which would return PEL messages for the current consumer), so it never drains the backlog on startup.

3. Fire-and-forget XACK

handlers.rs:25:

#![allow(unused)]
fn main() {
"pubsub_ack" | "pubsub_reply" | "pubsub_reaction" => Some(true),
}

XACK command results are silently consumed. If XACK fails (transient Redis error, connection blip), the message stays in the PEL with no retry, no alert, no logging. The next harness restart won’t reclaim it (see issue 1).

4. Only one message processed per idle tick

process_pubsub_messages() (handlers.rs:143) pops one message from the inbox via next_message(), processes it, then returns. If a poll returned 10 messages, 9 remain queued until subsequent idle ticks. With a 5s poll interval and a 30s heartbeat, processing 10 queued messages takes up to 5 minutes — during which the user sees no response and assumes messages were dropped.

Proposed Fixes

Fix 1: Add XAUTOCLAIM to reclaim PEL messages

Files: pubsub/redis.rs, pubsub/types.rs, pubsub/manager.rs

Add a claim_command() method to the MessageSource trait and RedisStreamsSource. This generates an XAUTOCLAIM command:

redis-cli XAUTOCLAIM {stream} {group} {consumer} {min_idle_ms} 0-0 COUNT 10

XAUTOCLAIM claims messages from the PEL that have been idle for at least min_idle_ms, transferring ownership to the current consumer. This recovers messages stranded by crashed/old consumers.

The SourceManager should issue claim_command() periodically (every 60s, matching the Python outbox’s DEFAULT_RECOVER_INTERVAL_S) alongside poll_command().

Changes:

• pubsub/types.rs: Add claim_command() to MessageSource trait with a default impl returning None.
• pubsub/redis.rs: Implement claim_command() on RedisStreamsSource — returns the XAUTOCLAIM shell command. Output is parsed by the same parse_poll_result() since the format is similar (XAUTOCLAIM returns [cursor, [(id, {fields})...], [deleted_ids]]).
• pubsub/manager.rs: Add a claim_commands() method that generates claim commands for all enabled sources. Track last_claim timestamps per source. Add a claim_interval field (default 60s).
• ranch/handlers.rs: Add claim_message_sources() method. Call it from on_idle_tick() alongside poll_message_sources(). Add "pubsub_claim" to handle_platform_command() — reuse the same handler as pubsub_poll since the output format is compatible.

New constant:

#![allow(unused)]
fn main() {
/// Minimum idle time (ms) before a PEL message can be claimed back.
const RECOVER_IDLE_MS: u64 = 30_000;
/// How often to run PEL recovery sweeps (seconds).
const RECOVER_INTERVAL_S: f64 = 60.0;
}

Fix 2: Drain PEL on startup with 0

Files: pubsub/redis.rs, pubsub/manager.rs

Add a first_poll boolean to RedisStreamsSource (or to SourceManager per-source). On the very first poll after source registration, use 0 instead of > as the read ID:

XREADGROUP GROUP {group} {consumer} COUNT 10 STREAMS {stream} '0'

This reads messages in the PEL for this consumer. Since the consumer name is new (random UUID), the PEL for this consumer will be empty on the very first poll — but combined with Fix 1 (XAUTOCLAIM), the first claim sweep will pick up orphaned PEL messages from dead consumers and assign them to the new consumer, and subsequent 0 polls will read them.

Alternative approach: Instead of tracking first_poll, just always issue both a 0 poll (for PEL) and an > poll (for new) on each tick. The 0 poll is cheap — it returns immediately if the PEL is empty. This is simpler and more robust than tracking first-poll state.

Changes:

• pubsub/redis.rs: Add a pel_poll_command() method that uses 0 as the stream ID. The main poll_command() continues using >.
• pubsub/manager.rs: poll_commands() issues both poll_command() and pel_poll_command() per source. Use distinct context types ("pubsub_poll" vs "pubsub_pel_poll") so results are handled the same way.
• Track inflight state separately for PEL and new polls to avoid double-issuing.

Fix 3: Verify XACK succeeded

Files: ranch/handlers.rs, pubsub/manager.rs

Change handle_platform_command to actually check XACK results instead of silently returning true:

#![allow(unused)]
fn main() {
"pubsub_ack" => {
    let source = context.get("source").cloned().unwrap_or_default();
    let message_id = context.get("message_id").cloned().unwrap_or_default();
    if exit_code == Some(0) {
        // XACK returns the number of acked entries (0 = nothing acked)
        let count = String::from_utf8_lossy(stdout).trim();
        if count == "0" || count.is_empty() {
            log_warn!(
                self.logger,
                "pubsub: XACK returned 0 for message {} on source '{}' \
                 (already acked or wrong group?)",
                message_id, source
            );
        } else {
            log_debug!(
                self.logger,
                "pubsub: XACK confirmed {} entry/entries on source '{}'",
                count, source
            );
        }
    } else {
        log_warn!(
            self.logger,
            "pubsub: XACK FAILED for message {} on source '{}': exit={:?} stderr={}",
            message_id, source, exit_code,
            String::from_utf8_lossy(stderr).trim()
        );
        // Re-queue the ack for the next tick
        self.source_manager.requeue_ack(&source, &message_id);
    }
    Some(true)
}
}

Changes to manager.rs:

Add a requeue_ack() method that puts a PendingAck back into pending_acks if the XACK command failed, so it will be retried on the next send_pubsub_acks() call.

#![allow(unused)]
fn main() {
pub fn requeue_ack(&mut self, source: &str, message_id: &str) {
    self.pending_acks.push(PendingAck {
        source: source.to_string(),
        message_id: message_id.to_string(),
        processed_at: Instant::now(),
    });
}
}

Also add a failed_acks counter to PendingAck or a wrapper struct to avoid infinite retry loops — after N failed attempts, log an error and drop the ack (the message will be reclaimed by XAUTOCLAIM eventually, providing a second chance at processing).

Fix 4: Process multiple messages per idle tick

Files: ranch/handlers.rs

Change process_pubsub_messages() to loop through all queued messages instead of processing just one:

#![allow(unused)]
fn main() {
pub fn process_pubsub_messages(&mut self) {
    if !matches!(
        self.state.status,
        AgentStatus::WaitingForInput | AgentStatus::Idle | AgentStatus::WaitingForExternal
    ) {
        return;
    }

    while self.source_manager.has_pending() {
        let msg = match self.source_manager.next_message() {
            Some(m) => m,
            None => break,
        };

        log_info!(
            self.logger,
            "Processing message {}/{} from {}: {}",
            // ... index, total, source, preview
            msg.source,
            if msg.content.chars().count() > 50 {
                let preview: String = msg.content.chars().take(50).collect();
                format!("{preview}...")
            } else {
                msg.content.clone()
            }
        );

        // ... existing processing logic ...
    }
}
}

Caveat: Processing multiple messages means multiple LLM calls in sequence. Each process_user_input() sends a prompt to the LLM and the response comes back asynchronously. The harness can’t process the next message until the current LLM response completes.

Alternative (recommended): Instead of a tight loop, process one message per tick but reduce the idle delay when the inbox has queued messages. In heartbeat.rs, plan_next_tick() can check if source_manager.has_pending() and use a fast delay (e.g., 0.5s) to drain the queue quickly:

#![allow(unused)]
fn main() {
} else if has_pending_messages {
    TickPlan {
        delay: delay::ACTIVE,  // 0.3s — fast drain
        rerender: false,
    }
} else {
    // Genuinely idle: back off
    TickPlan {
        delay: idle_delay,
        rerender: false,
    }
}
}

This way messages are processed one per tick but at 0.3s intervals instead of 5-30s, so 10 queued messages drain in ~3 seconds instead of up to 5 minutes.

Summary of Changes

Testing

• Add unit tests for claim_command() in redis.rs (verify command format)
• Add unit test for pel_poll_command() (verify 0 is used instead of >)
• Add unit test for requeue_ack() in manager.rs
• Add integration test: poll with >, simulate crash (no ack), restart, XAUTOCLAIM recovers the message
• Run existing test suite: cargo test -p cowboy-harness