feat(broker): dead-letter queue for failed deliveries + redeliver verb + persisted dedup cache

CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `agent-relay-broker` retains terminally-failed deliveries (retry cap exhausted or recipient gone) in a persisted, capped dead-letter queue instead of discarding them, emitting `dead_letter_added`/`dead_letter_redelivered` broker events and exposing `GET /api/dead-letters` and `POST /api/dead-letters/redeliver`.
+- `agent-relay local deadletters` lists dead-letter deliveries and `agent-relay local redeliver <id|--all>` requeues them through the normal delivery path with a reset retry count; `@agent-relay/harness-driver` adds matching `getDeadLetters()`/`redeliverDeadLetters()` client methods.
+- `agent-relay-broker` persists the inbound-event dedup cache alongside pending deliveries and reloads it on startup (dropping expired entries), so a crash + restart no longer re-injects duplicates when the persisted pending file replays.
 - `@agent-relay/sdk` wires the durable delivery surface to the Relaycast backend: `inbox.list`, `inbox.subscribe`, `inbox.ack/fail/defer`, and `deliveries.ack/fail/defer` now use the hosted delivery ledger, agent-scoped capabilities report `serverDeliveryState: true`, and `DeliveryRunner` works against Relaycast-backed inbox items.
 
 ### Changed

crates/broker/src/dedup.rs

@@ -1,14 +1,27 @@
 use std::{
     collections::{HashMap, VecDeque},
+    path::Path,
     time::{Duration, Instant},
 };
 
+use serde::{Deserialize, Serialize};
+
+/// Wall-clock snapshot of one dedup entry for crash recovery. `Instant`s
+/// cannot be serialized, so insertion times are persisted as unix millis and
+/// converted back relative to load time.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PersistedDedupEntry {
+    pub key: String,
+    pub inserted_at_ms: u64,
+}
+
 #[derive(Debug)]
 pub struct DedupCache {
     ttl: Duration,
     max_entries: usize,
     seen: HashMap<String, Instant>,
     order: VecDeque<(String, Instant)>,
+    dirty: bool,
 }
 
 impl DedupCache {
@@ -18,6 +31,7 @@ impl DedupCache {
             max_entries,
             seen: HashMap::new(),
             order: VecDeque::new(),
+            dirty: false,
         }
     }
 
@@ -27,6 +41,7 @@ impl DedupCache {
             return false;
         }
 
+        self.dirty = true;
         self.seen.insert(id.to_string(), now);
         self.order.push_back((id.to_string(), now));
 
@@ -51,11 +66,14 @@ impl DedupCache {
             }
             self.order.pop_front();
             self.seen.remove(&id);
+            self.dirty = true;
         }
     }
 
     pub fn remove(&mut self, id: &str) {
-        self.seen.remove(id);
+        if self.seen.remove(id).is_some() {
+            self.dirty = true;
+        }
         self.order.retain(|(key, _)| key != id);
     }
 
@@ -66,13 +84,83 @@ impl DedupCache {
     pub fn is_empty(&self) -> bool {
         self.seen.is_empty()
     }
+
+    /// Return whether the cache was mutated since the last call, clearing the flag.
+    pub fn take_dirty(&mut self) -> bool {
+        std::mem::take(&mut self.dirty)
+    }
+
+    /// Snapshot live entries as wall-clock timestamps (oldest first).
+    /// `now`/`now_ms` must describe the same moment on both clocks.
+    pub fn to_persisted(&self, now: Instant, now_ms: u64) -> Vec<PersistedDedupEntry> {
+        self.order
+            .iter()
+            .map(|(key, inserted_at)| PersistedDedupEntry {
+                key: key.clone(),
+                inserted_at_ms: now_ms
+                    .saturating_sub(now.duration_since(*inserted_at).as_millis() as u64),
+            })
+            .collect()
+    }
+
+    /// Rebuild a cache from persisted entries, dropping anything whose TTL
+    /// already elapsed. `now`/`now_ms` must describe the same moment on both
+    /// clocks.
+    pub fn from_persisted(
+        ttl: Duration,
+        max_entries: usize,
+        entries: Vec<PersistedDedupEntry>,
+        now: Instant,
+        now_ms: u64,
+    ) -> Self {
+        let mut cache = Self::new(ttl, max_entries);
+        let mut entries = entries;
+        entries.sort_by_key(|entry| entry.inserted_at_ms);
+        for entry in entries {
+            let elapsed = Duration::from_millis(now_ms.saturating_sub(entry.inserted_at_ms));
+            if elapsed >= ttl {
+                continue;
+            }
+            // `checked_sub` can fail close to process/system start; treating
+            // the entry as just-inserted only extends its dedup window.
+            let inserted_at = now.checked_sub(elapsed).unwrap_or(now);
+            if cache.seen.contains_key(&entry.key) {
+                continue;
+            }
+            cache.seen.insert(entry.key.clone(), inserted_at);
+            cache.order.push_back((entry.key, inserted_at));
+            while cache.seen.len() > cache.max_entries {
+                if let Some((old_id, _)) = cache.order.pop_front() {
+                    cache.seen.remove(&old_id);
+                }
+            }
+        }
+        cache
+    }
+}
+
+pub(crate) fn unix_now_millis() -> u64 {
+    chrono::Utc::now().timestamp_millis().max(0) as u64
+}
+
+pub(crate) fn save_dedup_cache(path: &Path, cache: &DedupCache) -> anyhow::Result<()> {
+    let persisted = cache.to_persisted(Instant::now(), unix_now_millis());
+    crate::util::fs::write_json_atomic(path, &persisted)
+}
+
+pub(crate) fn load_dedup_cache(path: &Path, ttl: Duration, max_entries: usize) -> DedupCache {
+    let entries: Vec<PersistedDedupEntry> = std::fs::read_to_string(path)
+        .ok()
+        .and_then(|data| serde_json::from_str(&data).ok())
+        .unwrap_or_default();
+    DedupCache::from_persisted(ttl, max_entries, entries, Instant::now(), unix_now_millis())
 }
 
 #[cfg(test)]
 mod tests {
     use std::time::{Duration, Instant};
 
-    use super::DedupCache;
+    use super::{load_dedup_cache, save_dedup_cache, DedupCache, PersistedDedupEntry};
 
     #[test]
     fn drops_duplicates() {
@@ -101,4 +189,81 @@ mod tests {
         // After TTL expires, should be insertable again
         assert!(dedup.insert_if_new("x", now + Duration::from_secs(6)));
     }
+
+    #[test]
+    fn tracks_dirty_on_mutation() {
+        let mut dedup = DedupCache::new(Duration::from_secs(60), 100);
+        assert!(!dedup.take_dirty(), "fresh cache starts clean");
+
+        assert!(dedup.insert_if_new("a", Instant::now()));
+        assert!(dedup.take_dirty(), "new insert marks dirty");
+        assert!(!dedup.take_dirty(), "take_dirty clears the flag");
+
+        assert!(!dedup.insert_if_new("a", Instant::now()));
+        assert!(!dedup.take_dirty(), "duplicate insert is not a mutation");
+
+        dedup.remove("a");
+        assert!(dedup.take_dirty(), "remove marks dirty");
+        dedup.remove("a");
+        assert!(
+            !dedup.take_dirty(),
+            "removing a missing key is not a mutation"
+        );
+    }
+
+    #[test]
+    fn persistence_round_trips_and_drops_expired() {
+        let dir = tempfile::tempdir().expect("tempdir should create");
+        let path = dir.path().join("dedup.json");
+        let ttl = Duration::from_secs(300);
+
+        let mut dedup = DedupCache::new(ttl, 100);
+        let now = Instant::now();
+        dedup.insert_if_new("ws_demo:evt_1", now);
+        dedup.insert_if_new("ws_demo:evt_2", now);
+        save_dedup_cache(&path, &dedup).expect("dedup cache should save");
+
+        let mut reloaded = load_dedup_cache(&path, ttl, 100);
+        assert_eq!(reloaded.len(), 2, "live entries survive a reload");
+        assert!(
+            !reloaded.insert_if_new("ws_demo:evt_1", Instant::now()),
+            "reloaded entries still deduplicate replayed events"
+        );
+        assert!(
+            reloaded.insert_if_new("ws_demo:evt_3", Instant::now()),
+            "new events still insert after reload"
+        );
+    }
+
+    #[test]
+    fn from_persisted_drops_expired_entries() {
+        let ttl = Duration::from_secs(300);
+        let now_ms: u64 = 1_700_000_000_000;
+        let entries = vec![
+            PersistedDedupEntry {
+                key: "fresh".to_string(),
+                inserted_at_ms: now_ms - 1_000,
+            },
+            PersistedDedupEntry {
+                key: "expired".to_string(),
+                inserted_at_ms: now_ms - ttl.as_millis() as u64 - 1,
+            },
+        ];
+
+        let mut cache = DedupCache::from_persisted(ttl, 100, entries, Instant::now(), now_ms);
+        assert_eq!(cache.len(), 1, "expired entries are dropped on load");
+        assert!(!cache.insert_if_new("fresh", Instant::now()));
+        assert!(cache.insert_if_new("expired", Instant::now()));
+    }
+
+    #[test]
+    fn load_missing_or_corrupt_file_yields_empty_cache() {
+        let dir = tempfile::tempdir().expect("tempdir should create");
+        let missing = dir.path().join("missing.json");
+        assert!(load_dedup_cache(&missing, Duration::from_secs(300), 100).is_empty());
+
+        let corrupt = dir.path().join("corrupt.json");
+        std::fs::write(&corrupt, "not json").expect("seed file should write");
+        assert!(load_dedup_cache(&corrupt, Duration::from_secs(300), 100).is_empty());
+    }
 }

crates/broker/src/listen_api.rs

@@ -11,7 +11,9 @@ use std::{
 };
 
 use crate::{
-    ids::{ChannelName, MessageTarget, ThreadId, WorkerName, WorkspaceAlias, WorkspaceId},
+    ids::{
+        ChannelName, DeliveryId, MessageTarget, ThreadId, WorkerName, WorkspaceAlias, WorkspaceId,
+    },
     protocol::{MessageInjectionMode, ResolvedHarnessConfig},
     relaycast::WorkspaceMembershipSummary,
     replay_buffer::ReplayBuffer,
@@ -130,6 +132,18 @@ pub enum ListenApiRequest {
     GetCrashInsights {
         reply: tokio::sync::oneshot::Sender<Result<Value, String>>,
     },
+    /// `GET /api/dead-letters` — list terminally-failed deliveries retained
+    /// in the dead-letter queue.
+    GetDeadLetters {
+        reply: tokio::sync::oneshot::Sender<Result<Value, String>>,
+    },
+    /// `POST /api/dead-letters/redeliver` — requeue dead-letter entries
+    /// through the normal delivery path with a reset retry count. `id: None`
+    /// redelivers every entry whose recipient is currently running.
+    RedeliverDeadLetters {
+        id: Option<DeliveryId>,
+        reply: tokio::sync::oneshot::Sender<Result<Value, String>>,
+    },
     Preflight {
         agents: Vec<PreflightEntry>,
         reply: tokio::sync::oneshot::Sender<Result<Value, String>>,
@@ -401,6 +415,11 @@ fn listen_api_router_with_auth(
             "/api/crash-insights",
             routing::get(listen_api_crash_insights),
         )
+        .route("/api/dead-letters", routing::get(listen_api_dead_letters))
+        .route(
+            "/api/dead-letters/redeliver",
+            routing::post(listen_api_redeliver_dead_letters),
+        )
         .route("/api/preflight", routing::post(listen_api_preflight))
         .route("/api/shutdown", routing::post(listen_api_shutdown))
         .route(
@@ -1887,6 +1906,72 @@ async fn listen_api_crash_insights(
     }
 }
 
+async fn listen_api_dead_letters(
+    axum::extract::State(state): axum::extract::State<ListenApiState>,
+) -> (axum::http::StatusCode, axum::Json<Value>) {
+    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
+    if state
+        .tx
+        .send(ListenApiRequest::GetDeadLetters { reply: reply_tx })
+        .await
+        .is_err()
+    {
+        return internal_error();
+    }
+    match reply_rx.await {
+        Ok(Ok(val)) => (axum::http::StatusCode::OK, axum::Json(val)),
+        Ok(Err(err)) => api_error(axum::http::StatusCode::INTERNAL_SERVER_ERROR, "error", err),
+        Err(_) => internal_error(),
+    }
+}
+
+#[derive(Deserialize, Default)]
+struct RedeliverBody {
+    /// Dead-letter delivery id to requeue; omit (with `all: true`) to
+    /// requeue everything.
+    id: Option<String>,
+    #[serde(default)]
+    all: bool,
+}
+
+async fn listen_api_redeliver_dead_letters(
+    axum::extract::State(state): axum::extract::State<ListenApiState>,
+    axum::Json(body): axum::Json<RedeliverBody>,
+) -> (axum::http::StatusCode, axum::Json<Value>) {
+    let id = match (&body.id, body.all) {
+        (Some(id), false) => Some(DeliveryId::from(id.as_str())),
+        (None, true) => None,
+        _ => {
+            return api_error(
+                axum::http::StatusCode::BAD_REQUEST,
+                "invalid_request",
+                "provide exactly one of 'id' or 'all: true'".to_string(),
+            );
+        }
+    };
+    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
+    if state
+        .tx
+        .send(ListenApiRequest::RedeliverDeadLetters {
+            id,
+            reply: reply_tx,
+        })
+        .await
+        .is_err()
+    {
+        return internal_error();
+    }
+    match reply_rx.await {
+        Ok(Ok(val)) => (axum::http::StatusCode::OK, axum::Json(val)),
+        Ok(Err(err)) => api_error(
+            axum::http::StatusCode::NOT_FOUND,
+            "dead_letter_not_found",
+            err,
+        ),
+        Err(_) => internal_error(),
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Lifecycle
 // ---------------------------------------------------------------------------

crates/broker/src/protocol.rs

@@ -447,6 +447,20 @@ pub enum BrokerEvent {
         #[serde(rename = "lastError")]
         last_error: String,
     },
+    DeadLetterAdded {
+        name: WorkerName,
+        delivery_id: DeliveryId,
+        event_id: EventId,
+        from: String,
+        to: MessageTarget,
+        attempts: u32,
+        reason: String,
+    },
+    DeadLetterRedelivered {
+        name: WorkerName,
+        delivery_id: DeliveryId,
+        event_id: EventId,
+    },
     DeliveryQueued {
         delivery_id: DeliveryId,
         agent: WorkerName,