diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/summary.md
new file mode 100644
index 000000000..17b30d27a
--- /dev/null
+++ b/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/summary.md
@@ -0,0 +1,40 @@
+# Trajectory: Fix manual-flush Relaycast ACK durability
+
+> **Status:** ✅ Completed
+> **Task:** #1241
+> **Confidence:** 93%
+> **Started:** July 10, 2026 at 10:15 AM
+> **Completed:** July 10, 2026 at 10:15 AM
+
+---
+
+## Summary
+
+Fixed #1241 by separating received and ACKed delivery cursors, retaining Relaycast receipts through manual flush, rejecting full queues without eviction, and ACKing only successfully injected FIFO prefixes. Verified focused cursor, replay, overflow, success, and failure regressions; cargo fmt check passed; full broker library suite passed 672 tests with 4 ignored.
+
+**Approach:** Standard approach
+
+---
+
+## Key Decisions
+
+### Split delivery progress into received and ACKed cursors
+
+- **Chose:** Split delivery progress into received and ACKed cursors
+- **Reasoning:** Manual-flush must accept multiple contiguous Relaycast sequences without cumulatively acknowledging volatile queue entries; duplicates and gaps therefore report only the ACKed cursor, while successful enqueue advances received state.
+
+### Reject the newest delivery when the manual queue is full
+
+- **Chose:** Reject the newest delivery when the manual queue is full
+- **Reasoning:** Evicting an already-held delivery could discard the only actionable copy after a later cumulative ACK. Atomic rejection leaves the FIFO unchanged, does not advance received or ACKed state, and preserves Relaycast ownership for replay.
+
+---
+
+## Chapters
+
+### 1. Work
+
+_Agent: default_
+
+- Split delivery progress into received and ACKed cursors: Split delivery progress into received and ACKed cursors
+- Reject the newest delivery when the manual queue is full: Reject the newest delivery when the manual queue is full
diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/trajectory.json b/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/trajectory.json
new file mode 100644
index 000000000..19416c81b
--- /dev/null
+++ b/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/trajectory.json
@@ -0,0 +1,69 @@
+{
+  "id": "traj_cbx77hjzvdaz",
+  "version": 1,
+  "task": {
+    "title": "Fix manual-flush Relaycast ACK durability",
+    "source": {
+      "system": "plain",
+      "id": "#1241"
+    }
+  },
+  "status": "completed",
+  "startedAt": "2026-07-10T14:15:30.303Z",
+  "completedAt": "2026-07-10T14:15:39.949Z",
+  "agents": [
+    {
+      "name": "default",
+      "role": "lead",
+      "joinedAt": "2026-07-10T14:15:35.758Z"
+    }
+  ],
+  "chapters": [
+    {
+      "id": "chap_xdlfc8rczkfx",
+      "title": "Work",
+      "agentName": "default",
+      "startedAt": "2026-07-10T14:15:35.758Z",
+      "endedAt": "2026-07-10T14:15:39.949Z",
+      "events": [
+        {
+          "ts": 1783692935759,
+          "type": "decision",
+          "content": "Split delivery progress into received and ACKed cursors: Split delivery progress into received and ACKed cursors",
+          "raw": {
+            "question": "Split delivery progress into received and ACKed cursors",
+            "chosen": "Split delivery progress into received and ACKed cursors",
+            "alternatives": [],
+            "reasoning": "Manual-flush must accept multiple contiguous Relaycast sequences without cumulatively acknowledging volatile queue entries; duplicates and gaps therefore report only the ACKed cursor, while successful enqueue advances received state."
+          },
+          "significance": "high"
+        },
+        {
+          "ts": 1783692935760,
+          "type": "decision",
+          "content": "Reject the newest delivery when the manual queue is full: Reject the newest delivery when the manual queue is full",
+          "raw": {
+            "question": "Reject the newest delivery when the manual queue is full",
+            "chosen": "Reject the newest delivery when the manual queue is full",
+            "alternatives": [],
+            "reasoning": "Evicting an already-held delivery could discard the only actionable copy after a later cumulative ACK. Atomic rejection leaves the FIFO unchanged, does not advance received or ACKed state, and preserves Relaycast ownership for replay."
+          },
+          "significance": "high"
+        }
+      ]
+    }
+  ],
+  "retrospective": {
+    "summary": "Fixed #1241 by separating received and ACKed delivery cursors, retaining Relaycast receipts through manual flush, rejecting full queues without eviction, and ACKing only successfully injected FIFO prefixes. Verified focused cursor, replay, overflow, success, and failure regressions; cargo fmt check passed; full broker library suite passed 672 tests with 4 ignored.",
+    "approach": "Standard approach",
+    "confidence": 0.93
+  },
+  "commits": [],
+  "filesChanged": [],
+  "projectId": "AgentWorkforce/relay",
+  "tags": [],
+  "_trace": {
+    "startRef": "3cfe0bf69bb24ad966b58ed288813bb3a7ef1a73",
+    "endRef": "3cfe0bf69bb24ad966b58ed288813bb3a7ef1a73"
+  }
+}
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 985dd2570..b42bd98c0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -101,6 +101,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- `agent-relay-broker` resumes Relaycast mailbox delivery at the server's authoritative per-agent ACK cursor after a broker restart, preserving strict gap detection and legacy node compatibility.
+- `agent-relay-broker` no longer acknowledges Relaycast `manual_flush` deliveries while they exist only in volatile memory; flushes ACK only an injected FIFO prefix, and full queues reject new deliveries without evicting held messages.
 - PTY snapshots (`view`/`drive`/`passthrough` attach, `GET /api/spawned/{name}/snapshot` ansi format) now capture and replay terminal modes — alt-screen, cursor visibility, application cursor keys, bracketed paste, mouse reporting, autowrap, and keypad — so attaching to a TUI no longer leaves the client terminal mis-configured (stray cursor, misbehaving arrows, broken paste). Each mode is re-emitted in both directions so an attach after a crashed session heals a terminal left in the wrong state.
 - `agent-relay view`/`drive`/`passthrough`: detaching now emits a conservative terminal reset (leave alt-screen, show cursor, disable mouse reporting + bracketed paste + application cursor keys, reset scroll region) on TTY stdout, so a driven session's replayed snapshot and live stream can't leave your shell in a broken terminal state.
 - `agent-relay drive`/`passthrough`: a `Ctrl+C` during attach setup no longer strands the worker's inbound delivery mode — an interrupt in that window can't leave the worker stuck in `manual_flush` (drive) or cancel an explicit `agent message hold` (passthrough).
diff --git a/crates/broker/src/listen_api.rs b/crates/broker/src/listen_api.rs
index 2d806597e..130ce3e80 100644
--- a/crates/broker/src/listen_api.rs
+++ b/crates/broker/src/listen_api.rs
@@ -2149,10 +2149,10 @@ async fn listen_api_get_pending(
 
 /// `POST /api/spawned/{name}/flush` → `{ "flushed": N }`.
 ///
-/// Drains the queue and injects each message into the worker in order
-/// using the existing fire-and-forget inject path. The inbound delivery mode is
-/// *not* changed — a caller still in `manual_flush` delivery mode will continue
-/// to queue newly-arriving messages.
+/// Injects queued messages into the worker in FIFO order and stops at the first
+/// failure, retaining that message and its suffix for a later attempt. The
+/// inbound delivery mode is *not* changed — a caller still in `manual_flush`
+/// delivery mode will continue to queue newly-arriving messages.
 async fn listen_api_flush_pending(
     axum::extract::State(state): axum::extract::State<ListenApiState>,
     axum::extract::Path(name): axum::extract::Path<String>,
@@ -5064,6 +5064,7 @@ mod auth_tests {
                             mode: MessageInjectionMode::Steer,
                             queued_at_ms: 100,
                             event_id: Some(EventId::new("evt_1")),
+                            relaycast_receipt: None,
                         },
                         PendingRelayMessage {
                             from: "Bob".to_string(),
@@ -5076,6 +5077,7 @@ mod auth_tests {
                             mode: MessageInjectionMode::Wait,
                             queued_at_ms: 200,
                             event_id: None,
+                            relaycast_receipt: None,
                         },
                     ]));
                 }
diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs
index 3f8a1afe8..21366d9db 100644
--- a/crates/broker/src/node_control.rs
+++ b/crates/broker/src/node_control.rs
@@ -22,6 +22,7 @@ use crate::{
         FLEET_WIRE_VERSION,
     },
     protocol::NodeManifest,
+    types::RelaycastDeliveryReceipt,
 };
 
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(12);
@@ -563,7 +564,8 @@ impl SeenMsgIds {
 #[derive(Debug, Default, Clone, PartialEq, Eq)]
 struct AgentDeliveryCursor {
     agent_name: String,
-    up_to_seq: u64,
+    acked_up_to_seq: u64,
+    received_up_to_seq: u64,
     seen_msg_ids: SeenMsgIds,
 }
 
@@ -680,6 +682,11 @@ impl FleetDeliveryBook {
     }
 
     /// Seed Relaycast's cumulative cursor after identity authority is bound.
+    ///
+    /// The immutable `agent_id` is the key: a later agent reusing the same name
+    /// must not inherit the old identity's cumulative ACK position. The seeded
+    /// cursor initializes `received == acked` at Relaycast's authoritative
+    /// position.
     pub(crate) fn seed_cursor(
         &mut self,
         agent: impl Into<String>,
@@ -696,7 +703,8 @@ impl FleetDeliveryBook {
             agent_id,
             AgentDeliveryCursor {
                 agent_name: agent,
-                up_to_seq,
+                acked_up_to_seq: up_to_seq,
+                received_up_to_seq: up_to_seq,
                 seen_msg_ids: SeenMsgIds::default(),
             },
         );
@@ -706,7 +714,7 @@ impl FleetDeliveryBook {
         self.active_agent_bindings_by_name
             .get(agent)
             .and_then(|binding| self.agents.get(&binding.agent_id))
-            .map_or(0, |cursor| cursor.up_to_seq)
+            .map_or(0, |cursor| cursor.acked_up_to_seq)
     }
 
     pub(crate) fn observe(&self, deliver: &Deliver) -> DeliveryDecision {
@@ -750,11 +758,11 @@ impl FleetDeliveryBook {
             if let Some(cursor) = cursor {
                 if cursor.seen_msg_ids.contains(&deliver.msg_id) {
                     return DeliveryDecision::Duplicate {
-                        up_to_seq: cursor.up_to_seq,
+                        up_to_seq: cursor.acked_up_to_seq,
                     };
                 }
                 return DeliveryDecision::Deliver {
-                    up_to_seq: cursor.up_to_seq,
+                    up_to_seq: cursor.acked_up_to_seq,
                 };
             }
             return DeliveryDecision::Deliver { up_to_seq: 0 };
@@ -768,19 +776,19 @@ impl FleetDeliveryBook {
         };
         if cursor.seen_msg_ids.contains(&deliver.msg_id) {
             return DeliveryDecision::Duplicate {
-                up_to_seq: cursor.up_to_seq,
+                up_to_seq: cursor.acked_up_to_seq,
             };
         }
 
-        if deliver.seq <= cursor.up_to_seq {
+        if deliver.seq <= cursor.received_up_to_seq {
             return DeliveryDecision::Stale {
-                up_to_seq: cursor.up_to_seq,
+                up_to_seq: cursor.acked_up_to_seq,
             };
         }
 
-        if deliver.seq != cursor.up_to_seq.saturating_add(1) {
+        if deliver.seq != cursor.received_up_to_seq.saturating_add(1) {
             return DeliveryDecision::Gap {
-                up_to_seq: cursor.up_to_seq,
+                up_to_seq: cursor.acked_up_to_seq,
             };
         }
 
@@ -789,7 +797,7 @@ impl FleetDeliveryBook {
         }
     }
 
-    pub(crate) fn commit_delivered(&mut self, deliver: &Deliver) -> u64 {
+    pub(crate) fn commit_received(&mut self, deliver: &Deliver) -> u64 {
         if !self.bind_identity(&deliver.agent, &deliver.agent_id, false) {
             return self.active_up_to_seq(&deliver.agent);
         }
@@ -801,17 +809,71 @@ impl FleetDeliveryBook {
                 ..AgentDeliveryCursor::default()
             });
         cursor.agent_name.clone_from(&deliver.agent);
-        // seq:0 fan-out frames never advance the sequence cursor; they are
+        // seq:0 fan-out frames never advance either sequence cursor; they are
         // deduped purely by msg_id.
         if deliver.seq == 0 {
             cursor.seen_msg_ids.insert(&deliver.msg_id);
-            return cursor.up_to_seq;
+            return cursor.received_up_to_seq;
         }
-        if deliver.seq == cursor.up_to_seq.saturating_add(1) {
+        if deliver.seq == cursor.received_up_to_seq.saturating_add(1) {
             cursor.seen_msg_ids.insert(&deliver.msg_id);
-            cursor.up_to_seq = deliver.seq;
+            cursor.received_up_to_seq = deliver.seq;
+        }
+        cursor.received_up_to_seq
+    }
+
+    pub(crate) fn commit_acked_receipt(
+        &mut self,
+        receipt: &RelaycastDeliveryReceipt,
+    ) -> Option<u64> {
+        let cursor = self.agents.get_mut(receipt.agent_id.as_str())?;
+        cursor.agent_name = receipt.agent.to_string();
+        if receipt.seq == 0 {
+            cursor.seen_msg_ids.insert(receipt.msg_id.as_str());
+            return Some(cursor.acked_up_to_seq);
+        }
+        if receipt.seq != cursor.acked_up_to_seq.saturating_add(1)
+            || receipt.seq > cursor.received_up_to_seq
+        {
+            return None;
         }
-        cursor.up_to_seq
+        cursor.acked_up_to_seq = receipt.seq;
+        Some(cursor.acked_up_to_seq)
+    }
+
+    pub(crate) fn can_ack_receipt(&self, receipt: &RelaycastDeliveryReceipt) -> bool {
+        let Some(cursor) = self.agents.get(receipt.agent_id.as_str()) else {
+            return false;
+        };
+        receipt.seq == 0
+            || (receipt.seq == cursor.acked_up_to_seq.saturating_add(1)
+                && receipt.seq <= cursor.received_up_to_seq)
+    }
+
+    pub(crate) fn commit_delivered(&mut self, deliver: &Deliver) -> u64 {
+        self.commit_received(deliver);
+        let receipt = RelaycastDeliveryReceipt {
+            agent: deliver.agent.clone().into(),
+            agent_id: deliver.agent_id.clone().into(),
+            delivery_id: deliver.delivery_id.clone().into(),
+            msg_id: deliver.msg_id.clone().into(),
+            seq: deliver.seq,
+        };
+        self.commit_acked_receipt(&receipt)
+            .unwrap_or_else(|| self.acked_up_to_seq(&deliver.agent_id))
+    }
+
+    pub(crate) fn acked_up_to_seq(&self, agent_id: &str) -> u64 {
+        self.agents
+            .get(agent_id)
+            .map_or(0, |cursor| cursor.acked_up_to_seq)
+    }
+
+    #[cfg(test)]
+    pub(crate) fn received_up_to_seq(&self, agent_id: &str) -> u64 {
+        self.agents
+            .get(agent_id)
+            .map_or(0, |cursor| cursor.received_up_to_seq)
     }
 
     pub(crate) fn remove_agent(&mut self, agent: &str) {
@@ -2319,6 +2381,109 @@ mod tests {
         );
     }
 
+    #[test]
+    fn delivery_book_receives_multiple_sequences_without_acknowledging_them() {
+        let mut book = FleetDeliveryBook::default();
+        let first = Deliver {
+            v: FLEET_WIRE_VERSION,
+            agent: "agent-a".to_string(),
+            agent_id: "agent-a-id".to_string(),
+            delivery_id: "delivery-1".to_string(),
+            msg_id: "msg-1".to_string(),
+            seq: 1,
+            mode: DeliveryMode::Wait,
+            payload: json!({"text": "one"}),
+        };
+        let second = Deliver {
+            delivery_id: "delivery-2".to_string(),
+            msg_id: "msg-2".to_string(),
+            seq: 2,
+            payload: json!({"text": "two"}),
+            ..first.clone()
+        };
+
+        assert_eq!(book.commit_received(&first), 1);
+        assert_eq!(book.acked_up_to_seq("agent-a-id"), 0);
+        assert_eq!(
+            book.observe(&first),
+            DeliveryDecision::Duplicate { up_to_seq: 0 },
+            "a replayed held frame must not be queued twice or ACKed"
+        );
+        assert_eq!(
+            book.observe(&second),
+            DeliveryDecision::Deliver { up_to_seq: 2 }
+        );
+        assert_eq!(book.commit_received(&second), 2);
+        assert_eq!(book.received_up_to_seq("agent-a-id"), 2);
+        assert_eq!(book.acked_up_to_seq("agent-a-id"), 0);
+
+        let second_receipt = RelaycastDeliveryReceipt {
+            agent: "agent-a".into(),
+            agent_id: "agent-a-id".into(),
+            delivery_id: "delivery-2".into(),
+            msg_id: "msg-2".into(),
+            seq: 2,
+        };
+        assert_eq!(
+            book.commit_acked_receipt(&second_receipt),
+            None,
+            "cumulative ACK cannot skip the held first sequence"
+        );
+
+        let first_receipt = RelaycastDeliveryReceipt {
+            agent: "agent-a".into(),
+            agent_id: "agent-a-id".into(),
+            delivery_id: "delivery-1".into(),
+            msg_id: "msg-1".into(),
+            seq: 1,
+        };
+        assert_eq!(book.commit_acked_receipt(&first_receipt), Some(1));
+        assert_eq!(book.commit_acked_receipt(&second_receipt), Some(2));
+    }
+
+    #[test]
+    fn delivery_book_replays_unacked_manual_sequences_after_restart_baseline() {
+        let mut before_restart = FleetDeliveryBook::default();
+        seed_authoritative_cursor(&mut before_restart, "agent-a", "agent-a-id", 42);
+        let first = Deliver {
+            v: FLEET_WIRE_VERSION,
+            agent: "agent-a".to_string(),
+            agent_id: "agent-a-id".to_string(),
+            delivery_id: "delivery-43".to_string(),
+            msg_id: "msg-43".to_string(),
+            seq: 43,
+            mode: DeliveryMode::Wait,
+            payload: json!({"text": "held before restart"}),
+        };
+        let second = Deliver {
+            delivery_id: "delivery-44".to_string(),
+            msg_id: "msg-44".to_string(),
+            seq: 44,
+            ..first.clone()
+        };
+        before_restart.commit_received(&first);
+        before_restart.commit_received(&second);
+        assert_eq!(before_restart.received_up_to_seq("agent-a-id"), 44);
+        assert_eq!(before_restart.acked_up_to_seq("agent-a-id"), 42);
+
+        // Issue #1240 supplies this persisted ACK baseline after a real broker
+        // restart. Received-only state is intentionally volatile so Relaycast
+        // can replay every unacknowledged manual delivery.
+        let mut after_restart = FleetDeliveryBook::default();
+        seed_authoritative_cursor(&mut after_restart, "agent-a", "agent-a-id", 42);
+        assert_eq!(
+            after_restart.observe(&first),
+            DeliveryDecision::Deliver { up_to_seq: 43 }
+        );
+        after_restart.commit_received(&first);
+        assert_eq!(
+            after_restart.observe(&second),
+            DeliveryDecision::Deliver { up_to_seq: 44 }
+        );
+        after_restart.commit_received(&second);
+        assert_eq!(after_restart.acked_up_to_seq("agent-a-id"), 42);
+    }
+
     #[test]
     fn delivery_book_remove_agent_prunes_cursor_and_msg_ids() {
         let mut book = FleetDeliveryBook::default();
diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs
index 4882e3e99..1948b046d 100644
--- a/crates/broker/src/runtime/api.rs
+++ b/crates/broker/src/runtime/api.rs
@@ -1447,95 +1447,126 @@ impl BrokerRuntime {
                 if !workers.has_worker(&name) {
                     let _ = reply.send(Err(DeliveryRouteError::WorkerNotFound(name)));
                 } else {
-                    let entry = delivery_states.entry(name.clone()).or_default();
                     // Compare-and-set guard: when the caller supplied an
                     // `expected_mode` (detach restore) and it no longer matches
                     // the worker's current mode, a concurrent change happened —
                     // no-op and report the current mode with `matched: false`
                     // rather than clobbering it. Closes the restore TOCTOU.
-                    if let Some(expected) = expected_mode {
-                        if entry.mode != expected {
-                            let current = entry.mode;
-                            tracing::info!(
-                                target = "agent_relay::broker",
-                                worker = %name,
-                                expected = expected.as_wire_str(),
-                                current = current.as_wire_str(),
-                                requested = mode.as_wire_str(),
-                                "inbound delivery mode compare-and-set skipped (expected_mode mismatch)"
-                            );
-                            let _ = reply.send(Ok(SetInboundDeliveryModeOk {
-                                mode: current,
-                                flushed: 0,
-                                matched: false,
-                                revision: entry.revision,
-                            }));
-                            return;
+                    {
+                        let entry = delivery_states.entry(name.clone()).or_default();
+                        if let Some(expected) = expected_mode {
+                            if entry.mode != expected {
+                                let current = entry.mode;
+                                let revision = entry.revision;
+                                tracing::info!(
+                                    target = "agent_relay::broker",
+                                    worker = %name,
+                                    expected = expected.as_wire_str(),
+                                    current = current.as_wire_str(),
+                                    requested = mode.as_wire_str(),
+                                    "inbound delivery mode compare-and-set skipped (expected_mode mismatch)"
+                                );
+                                let _ = reply.send(Ok(SetInboundDeliveryModeOk {
+                                    mode: current,
+                                    flushed: 0,
+                                    matched: false,
+                                    revision,
+                                }));
+                                return;
+                            }
                         }
-                    }
-                    if let Some(expected) = expected_revision {
-                        if entry.revision != expected {
-                            let current = entry.mode;
-                            tracing::info!(
-                                target = "agent_relay::broker",
-                                worker = %name,
-                                expected_revision = expected,
-                                current_revision = entry.revision,
-                                requested = mode.as_wire_str(),
-                                "inbound delivery mode compare-and-set skipped (revision mismatch)"
-                            );
-                            let _ = reply.send(Ok(SetInboundDeliveryModeOk {
-                                mode: current,
-                                flushed: 0,
-                                matched: false,
-                                revision: entry.revision,
-                            }));
-                            return;
+                        if let Some(expected) = expected_revision {
+                            if entry.revision != expected {
+                                let current = entry.mode;
+                                let revision = entry.revision;
+                                tracing::info!(
+                                    target = "agent_relay::broker",
+                                    worker = %name,
+                                    expected_revision = expected,
+                                    current_revision = revision,
+                                    requested = mode.as_wire_str(),
+                                    "inbound delivery mode compare-and-set skipped (revision mismatch)"
+                                );
+                                let _ = reply.send(Ok(SetInboundDeliveryModeOk {
+                                    mode: current,
+                                    flushed: 0,
+                                    matched: false,
+                                    revision,
+                                }));
+                                return;
+                            }
                         }
                     }
-                    let previous = entry.set_mode(mode);
-                    let to_flush: Vec<PendingRelayMessage> = if previous
-                        == InboundDeliveryMode::ManualFlush
-                        && mode == InboundDeliveryMode::AutoInject
-                    {
-                        entry.drain_pending()
-                    } else {
-                        Vec::new()
-                    };
-                    let flushed = to_flush.len();
-                    if !to_flush.is_empty() {
+                    let previous = delivery_states.entry(name.clone()).or_default().mode;
+                    let transition_requires_flush = previous == InboundDeliveryMode::ManualFlush
+                        && mode == InboundDeliveryMode::AutoInject;
+                    // Deferred-ACK flush: inject and ACK only the contiguous FIFO
+                    // prefix, stopping at the first not-yet-ACKable receipt or
+                    // failed injection so held frames are never silently ACKed.
+                    let flush_result = if transition_requires_flush {
                         tracing::info!(
                             target = "agent_relay::broker",
                             worker = %name,
-                            drained = flushed,
                             "draining pending queue on manual_flush → auto_inject transition"
                         );
-                    }
-                    for queued in to_flush {
-                        inject_pending_relay_message(
+                        super::fleet::flush_pending_relay_messages(
+                            delivery_states,
                             workers,
-                            pending_deliveries,
+                            fleet_delivery_book,
+                            fleet_control_tx,
                             &name,
-                            &queued,
                             delivery_retry_interval,
                         )
-                        .await;
+                        .await
+                    } else {
+                        super::fleet::FlushPendingRelayResult::default()
+                    };
+                    let flushed = flush_result.flushed;
+                    if let Some(error) = flush_result.failure.as_deref() {
+                        tracing::warn!(
+                            target = "agent_relay::broker",
+                            worker = %name,
+                            flushed,
+                            error,
+                            "stopped delivery-mode transition at failed pending message"
+                        );
+                    }
+                    // A partial flush leaves the worker in manual_flush so the
+                    // held frames are retried rather than silently ACKed.
+                    let actual_mode = if transition_requires_flush && flush_result.failure.is_some()
+                    {
+                        InboundDeliveryMode::ManualFlush
+                    } else {
+                        mode
+                    };
+                    let revision = {
+                        let entry = delivery_states.entry(name.clone()).or_default();
+                        entry.set_mode(actual_mode);
+                        entry.revision
+                    };
+                    if flushed > 0 {
+                        tracing::info!(
+                            target = "agent_relay::broker",
+                            worker = %name,
+                            drained = flushed,
+                            "drained pending queue on delivery-mode transition"
+                        );
                     }
                     tracing::info!(
                         target = "agent_relay::broker",
                         worker = %name,
                         previous_mode = previous.as_wire_str(),
-                        mode = mode.as_wire_str(),
+                        mode = actual_mode.as_wire_str(),
                         flushed,
                         "inbound delivery mode updated"
                     );
-                    if previous != mode {
+                    if previous != actual_mode {
                         // Toggle the worker-side interactive hold across a
                         // manual_flush boundary so worker automation (pending
                         // injections, auto-enter, prompt auto-responders) can't
                         // splice into a human's drive. Only PTY workers run that
                         // automation; headless workers don't handle the frame.
-                        let entered_manual = mode == InboundDeliveryMode::ManualFlush;
+                        let entered_manual = actual_mode == InboundDeliveryMode::ManualFlush;
                         let left_manual = previous == InboundDeliveryMode::ManualFlush;
                         if (entered_manual || left_manual)
                             && workers
@@ -1568,7 +1599,7 @@ impl BrokerRuntime {
                                 "kind":"agent_inbound_delivery_mode_changed",
                                 "name":&name,
                                 "previous_mode":previous.as_wire_str(),
-                                "mode":mode.as_wire_str(),
+                                "mode":actual_mode.as_wire_str(),
                             }),
                         )
                         .await;
@@ -1586,10 +1617,10 @@ impl BrokerRuntime {
                         .await;
                     }
                     let _ = reply.send(Ok(SetInboundDeliveryModeOk {
-                        mode,
+                        mode: actual_mode,
                         flushed,
                         matched: true,
-                        revision: entry.revision,
+                        revision,
                     }));
                 }
             }
@@ -1608,11 +1639,16 @@ impl BrokerRuntime {
                 if !workers.has_worker(&name) {
                     let _ = reply.send(Err(DeliveryRouteError::WorkerNotFound(name)));
                 } else {
-                    let to_flush: Vec<PendingRelayMessage> = delivery_states
-                        .get_mut(&name)
-                        .map(|state| state.drain_pending())
-                        .unwrap_or_default();
-                    let flushed = to_flush.len();
+                    let flush_result = super::fleet::flush_pending_relay_messages(
+                        delivery_states,
+                        workers,
+                        fleet_delivery_book,
+                        fleet_control_tx,
+                        &name,
+                        delivery_retry_interval,
+                    )
+                    .await;
+                    let flushed = flush_result.flushed;
                     if flushed > 0 {
                         tracing::info!(
                             target = "agent_relay::broker",
@@ -1621,15 +1657,14 @@ impl BrokerRuntime {
                             "flushing pending queue on explicit /flush"
                         );
                     }
-                    for queued in to_flush {
-                        inject_pending_relay_message(
-                            workers,
-                            pending_deliveries,
-                            &name,
-                            &queued,
-                            delivery_retry_interval,
-                        )
-                        .await;
+                    if let Some(error) = flush_result.failure.as_deref() {
+                        tracing::warn!(
+                            target = "agent_relay::broker",
+                            worker = %name,
+                            flushed,
+                            error,
+                            "stopped explicit flush at failed pending message"
+                        );
                     }
                     if flushed > 0 {
                         let _ = send_event(
diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs
index 3dc4c4c44..37545fc08 100644
--- a/crates/broker/src/runtime/delivery.rs
+++ b/crates/broker/src/runtime/delivery.rs
@@ -400,6 +400,7 @@ fn emit_delivery_read_ack_telemetry(
 pub(crate) enum InboundQueueOutcome {
     Queued,
     DrainNow(Vec<PendingRelayMessage>),
+    RejectedFull,
     WorkerMissing,
 }
 
@@ -446,6 +447,7 @@ pub(crate) struct InboundContext<'a> {
     pub(super) priority: u8,
     pub(super) mode: MessageInjectionMode,
     pub(super) event_id: Option<&'a str>,
+    pub(super) relaycast_receipt: Option<crate::types::RelaycastDeliveryReceipt>,
 }
 
 /// Queue an inbound relay message through the per-worker [`InboundDeliveryMode`].
@@ -475,6 +477,21 @@ pub(crate) fn queue_inbound_for_delivery_mode(
     let state = delivery_states
         .entry(WorkerName::from(worker_name))
         .or_default();
+    if state.pending.len() >= crate::types::MAX_PENDING_PER_WORKER {
+        tracing::warn!(
+            target = "agent_relay::broker",
+            worker = %worker_name,
+            from = %ctx.from,
+            mode = state.mode.as_wire_str(),
+            queue_len = state.pending.len(),
+            max_pending = crate::types::MAX_PENDING_PER_WORKER,
+            "pending queue full - rejecting newest message"
+        );
+        return InboundQueueResult {
+            outcome: InboundQueueOutcome::RejectedFull,
+            evicted_from: None,
+        };
+    }
     let should_drain = state.should_drain_immediately();
     let queued_at_ms = chrono::Utc::now().timestamp_millis().max(0) as u64;
     let msg = PendingRelayMessage {
@@ -488,6 +505,7 @@ pub(crate) fn queue_inbound_for_delivery_mode(
         mode: ctx.mode,
         queued_at_ms,
         event_id: ctx.event_id.map(EventId::from),
+        relaycast_receipt: ctx.relaycast_receipt,
     };
     let evicted_from = match state.accept_inbound(msg) {
         InboundDeliveryDispatch::Queued { queue_len } => {
@@ -579,38 +597,46 @@ pub(crate) async fn try_inject_pending_relay_message(
     }
 }
 
-/// Inject a previously-queued pending relay message into the worker via
-/// the existing `queue_and_try_delivery_raw` path. Used by the
-/// `/api/spawned/{name}/flush` handler and by the auto-drain on a
-/// `manual_flush → auto_inject` transition. Failures are logged but not
-/// propagated — the broker treats `flush` as best-effort fire-and-forget
-/// the same way `/api/send` does for individual targets.
-pub(crate) async fn inject_pending_relay_message(
+/// Attempt one PTY injection without transferring ownership to the broker's
+/// retry queue. Manual-flush callers keep the original message at the head of
+/// their FIFO on failure, along with its Relaycast receipt, so retrying cannot
+/// race a second broker-owned copy of the same delivery.
+pub(crate) async fn try_inject_pending_relay_message_once(
     workers: &mut WorkerRegistry,
-    pending_deliveries: &mut HashMap<DeliveryId, PendingDelivery>,
     worker_name: &str,
     msg: &PendingRelayMessage,
     retry_interval: Duration,
-) {
-    let event_id = msg.event_id.as_deref().unwrap_or("");
-    if let Err(error) = try_inject_pending_relay_message(
-        workers,
-        pending_deliveries,
-        worker_name,
-        msg,
-        retry_interval,
-    )
-    .await
-    {
-        tracing::warn!(
-            target = "agent_relay::broker",
-            worker = %worker_name,
-            from = %msg.from,
-            event_id = %event_id,
-            error = %error,
-            "failed to inject pending relay message during flush"
-        );
-    }
+) -> Result<()> {
+    let event_id = msg
+        .event_id
+        .clone()
+        .unwrap_or_else(|| EventId::new(format!("flush_{}", Uuid::new_v4().simple())));
+    let delivery_id = msg
+        .relaycast_receipt
+        .as_ref()
+        .map(|receipt| receipt.delivery_id.clone())
+        .unwrap_or_else(|| DeliveryId::new(format!("del_{}", Uuid::new_v4().simple())));
+    let delivery = RelayDelivery {
+        delivery_id,
+        event_id,
+        workspace_id: msg.workspace_id.clone(),
+        workspace_alias: msg.workspace_alias.clone(),
+        from: msg.from.clone(),
+        target: msg.target.clone(),
+        body: msg.body.clone(),
+        thread_id: msg.thread_id.clone(),
+        priority: Some(msg.priority),
+        injection_mode: msg.mode.clone(),
+    };
+
+    timeout(retry_interval, workers.deliver(worker_name, delivery))
+        .await
+        .map_err(|_| {
+            anyhow::anyhow!(
+                "pending relay delivery timed out after {}ms",
+                retry_interval.as_millis()
+            )
+        })?
 }
 
 #[allow(clippy::too_many_arguments)]
diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs
index a0b076642..4320cd570 100644
--- a/crates/broker/src/runtime/fleet.rs
+++ b/crates/broker/src/runtime/fleet.rs
@@ -10,6 +10,12 @@ use crate::{
 
 const FLEET_AGENT_REGISTER_TIMEOUT: Duration = Duration::from_secs(30);
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum FleetDeliverySurfaceOutcome {
+    Acknowledge,
+    HoldForManualFlush,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum FleetDeliveryPlan {
     Surface,
@@ -64,7 +70,13 @@ impl BrokerRuntime {
         let decision = self.fleet_delivery_book.observe(&deliver);
         let up_to_seq = match plan_fleet_delivery(decision) {
             FleetDeliveryPlan::Surface => match self.surface_fleet_deliver(&deliver).await {
-                Ok(()) => self.fleet_delivery_book.commit_delivered(&deliver),
+                Ok(FleetDeliverySurfaceOutcome::Acknowledge) => {
+                    self.fleet_delivery_book.commit_delivered(&deliver)
+                }
+                Ok(FleetDeliverySurfaceOutcome::HoldForManualFlush) => {
+                    self.fleet_delivery_book.commit_received(&deliver);
+                    return;
+                }
                 Err(error) => {
                     tracing::warn!(
                         target = "relay_broker::fleet",
@@ -102,10 +114,13 @@ impl BrokerRuntime {
     /// Surface a node `deliver` frame by branching on its payload `type`:
     /// message-class events inject into the recipient worker's PTY; reaction /
     /// read receipts are acked with a tracing log only (PTY surfacing deferred).
-    /// An `Ok` return means the delivery may be committed and acked; an `Err`
-    /// means injection failed and the ack must be withheld so the engine
-    /// redelivers.
-    async fn surface_fleet_deliver(&mut self, deliver: &Deliver) -> Result<(), anyhow::Error> {
+    /// `Acknowledge` means the delivery crossed the PTY injection boundary;
+    /// `HoldForManualFlush` means it was received into the volatile FIFO but
+    /// remains owned by Relaycast until a later successful flush.
+    async fn surface_fleet_deliver(
+        &mut self,
+        deliver: &Deliver,
+    ) -> Result<FleetDeliverySurfaceOutcome, anyhow::Error> {
         let payload_type = deliver
             .payload
             .get("type")
@@ -170,6 +185,13 @@ impl BrokerRuntime {
                         priority,
                         mode: injection_mode,
                         event_id: Some(&deliver.msg_id),
+                        relaycast_receipt: Some(RelaycastDeliveryReceipt {
+                            agent: WorkerName::from(&deliver.agent),
+                            agent_id: AgentId::from(&deliver.agent_id),
+                            delivery_id: DeliveryId::from(&deliver.delivery_id),
+                            msg_id: EventId::from(&deliver.msg_id),
+                            seq: deliver.seq,
+                        }),
                     },
                 );
                 if let Some(dropped_from) = &queue_result.evicted_from {
@@ -207,7 +229,7 @@ impl BrokerRuntime {
                             }),
                         )
                         .await;
-                        Ok(())
+                        Ok(FleetDeliverySurfaceOutcome::HoldForManualFlush)
                     }
                     InboundQueueOutcome::DrainNow(to_drain) => {
                         // Mirrors the HTTP send path: drain may surface older
@@ -243,11 +265,18 @@ impl BrokerRuntime {
                                 }
                             }
                         }
-                        current_result
+                        current_result.map(|()| FleetDeliverySurfaceOutcome::Acknowledge)
                     }
+                    InboundQueueOutcome::RejectedFull => anyhow::bail!(
+                        "manual delivery queue is full for '{}'; retaining Relaycast ownership",
+                        deliver.agent
+                    ),
                     InboundQueueOutcome::WorkerMissing => {
                         let relay_delivery = self.fleet_relay_delivery(deliver);
-                        self.workers.deliver(&deliver.agent, relay_delivery).await
+                        self.workers
+                            .deliver(&deliver.agent, relay_delivery)
+                            .await
+                            .map(|()| FleetDeliverySurfaceOutcome::Acknowledge)
                     }
                 }
             }
@@ -260,7 +289,7 @@ impl BrokerRuntime {
                     payload_type = %payload_type,
                     "acking node receipt/reaction delivery without PTY surfacing (deferred)"
                 );
-                Ok(())
+                Ok(FleetDeliverySurfaceOutcome::Acknowledge)
             }
             FleetDeliverySurfacing::AckUnknown => {
                 tracing::warn!(
@@ -270,7 +299,7 @@ impl BrokerRuntime {
                     payload_type = %payload_type,
                     "acking unrecognized node delivery payload type without surfacing"
                 );
-                Ok(())
+                Ok(FleetDeliverySurfaceOutcome::Acknowledge)
             }
         }
     }
@@ -514,6 +543,87 @@ impl BrokerRuntime {
     }
 }
 
+#[derive(Debug, Default, PartialEq, Eq)]
+pub(super) struct FlushPendingRelayResult {
+    pub(super) flushed: usize,
+    pub(super) failure: Option<String>,
+}
+
+/// Inject a worker's held queue in FIFO order. A failed item and every item
+/// behind it remain queued. Relaycast ACKs advance only after the corresponding
+/// PTY write succeeds, so the emitted cursor is always an injected prefix.
+pub(super) async fn flush_pending_relay_messages(
+    delivery_states: &mut HashMap<WorkerName, InboundDeliveryState>,
+    workers: &mut WorkerRegistry,
+    fleet_delivery_book: &mut FleetDeliveryBook,
+    fleet_control_tx: &mpsc::Sender<FleetControlCommand>,
+    worker_name: &WorkerName,
+    retry_interval: Duration,
+) -> FlushPendingRelayResult {
+    let mut result = FlushPendingRelayResult::default();
+
+    loop {
+        let next = delivery_states
+            .get(worker_name)
+            .and_then(|state| state.pending.front())
+            .cloned();
+        let Some(queued) = next else {
+            break;
+        };
+
+        if let Some(receipt) = queued.relaycast_receipt.as_ref() {
+            if !fleet_delivery_book.can_ack_receipt(receipt) {
+                result.failure = Some(format!(
+                    "delivery sequence {} for '{}' is not the next ACKable receipt",
+                    receipt.seq, receipt.agent
+                ));
+                break;
+            }
+        }
+
+        if let Err(error) =
+            try_inject_pending_relay_message_once(workers, worker_name, &queued, retry_interval)
+                .await
+        {
+            result.failure = Some(error.to_string());
+            break;
+        }
+
+        if let Some(receipt) = queued.relaycast_receipt.as_ref() {
+            let Some(up_to_seq) = fleet_delivery_book.commit_acked_receipt(receipt) else {
+                result.failure = Some(format!(
+                    "delivery sequence {} for '{}' could not advance the ACK cursor",
+                    receipt.seq, receipt.agent
+                ));
+                break;
+            };
+            if let Err(error) = fleet_control_tx
+                .send(FleetControlCommand::Send(delivery_ack(
+                    receipt.agent.to_string(),
+                    up_to_seq,
+                )))
+                .await
+            {
+                tracing::warn!(
+                    target = "relay_broker::fleet",
+                    agent = %receipt.agent,
+                    up_to_seq,
+                    error = %error,
+                    "failed to enqueue delivery ACK after manual flush"
+                );
+            }
+        }
+
+        let removed = delivery_states
+            .get_mut(worker_name)
+            .and_then(|state| state.pending.pop_front());
+        debug_assert_eq!(removed.as_ref(), Some(&queued));
+        result.flushed += 1;
+    }
+
+    result
+}
+
 /// Bind an agent to this node by sending node-control `agent.register` and
 /// awaiting the engine reply with the minted agent token. This is the single
 /// "register agent via node" step both the `/api/spawn` path and the node
diff --git a/crates/broker/src/runtime/mod.rs b/crates/broker/src/runtime/mod.rs
index 424e8fe7a..db0cef560 100644
--- a/crates/broker/src/runtime/mod.rs
+++ b/crates/broker/src/runtime/mod.rs
@@ -44,7 +44,7 @@ use crate::{
     telemetry::{ActionSource, TelemetryClient, TelemetryEvent},
     types::{
         AgentResultMcpConfig, InboundDeliveryDispatch, InboundDeliveryMode, InboundDeliveryState,
-        PendingRelayMessage,
+        PendingRelayMessage, RelaycastDeliveryReceipt,
     },
 };
 
diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs
index da82d96fd..f4da6b358 100644
--- a/crates/broker/src/runtime/tests.rs
+++ b/crates/broker/src/runtime/tests.rs
@@ -6,10 +6,12 @@ use std::{
     time::{Duration, Instant},
 };
 
+use crate::fleet_wire::{BrokerToRelaycast, Deliver, DeliveryMode, FLEET_WIRE_VERSION};
 use crate::ids::{
     AgentId, ChannelName, DeliveryId, EventId, MessageTarget, WorkerName, WorkspaceAlias,
     WorkspaceId,
 };
+use crate::node_control::{FleetControlCommand, FleetDeliveryBook};
 use crate::protocol::{
     AgentSpec, BrokerEvent, DeliveryReadAckStatus, HarnessReleasePolicy, HeadlessHarnessConfig,
     HeadlessHarnessDriver, MessageInjectionMode, RelayDelivery, ResolvedHarnessConfig,
@@ -51,7 +53,9 @@ use crate::dedup::DedupCache;
 use crate::relaycast::{
     format_worker_preregistration_error, RelaycastHttpClient, RelaycastRegistrationError, WsControl,
 };
-use crate::types::{InboundDeliveryMode, InboundDeliveryState};
+use crate::types::{
+    InboundDeliveryMode, InboundDeliveryState, PendingRelayMessage, RelaycastDeliveryReceipt,
+};
 use relaycast::ObserverScope;
 
 fn env_test_lock() -> &'static Mutex<()> {
@@ -126,6 +130,42 @@ fn inbound_ctx<'a>(event_id: &'a str) -> InboundContext<'a> {
         priority: 1,
         mode: MessageInjectionMode::Steer,
         event_id: Some(event_id),
+        relaycast_receipt: None,
+    }
+}
+
+fn fleet_deliver(seq: u64) -> Deliver {
+    Deliver {
+        v: FLEET_WIRE_VERSION,
+        agent: "worker-a".to_string(),
+        agent_id: "agent-worker-a".to_string(),
+        delivery_id: format!("delivery-{seq}"),
+        msg_id: format!("message-{seq}"),
+        seq,
+        mode: DeliveryMode::Wait,
+        payload: json!({"type": "message.created", "text": format!("message {seq}")}),
+    }
+}
+
+fn held_fleet_message(deliver: &Deliver) -> PendingRelayMessage {
+    PendingRelayMessage {
+        from: "Alice".to_string(),
+        body: format!("message {}", deliver.seq),
+        target: MessageTarget::new("worker-a"),
+        thread_id: None,
+        workspace_id: Some(WorkspaceId::new("ws_demo")),
+        workspace_alias: Some(WorkspaceAlias::new("Demo")),
+        priority: 2,
+        mode: MessageInjectionMode::Wait,
+        queued_at_ms: super::unix_timestamp_millis(),
+        event_id: Some(EventId::from(&deliver.msg_id)),
+        relaycast_receipt: Some(RelaycastDeliveryReceipt {
+            agent: WorkerName::from(&deliver.agent),
+            agent_id: AgentId::from(&deliver.agent_id),
+            delivery_id: DeliveryId::from(&deliver.delivery_id),
+            msg_id: EventId::from(&deliver.msg_id),
+            seq: deliver.seq,
+        }),
     }
 }
 
@@ -246,7 +286,7 @@ async fn inbound_queue_worker_missing_does_not_create_state() {
 }
 
 #[tokio::test]
-async fn inbound_queue_eviction_surfaces_dropped_message() {
+async fn inbound_queue_rejects_overflow_without_evicting_held_message() {
     let worker_name = "worker-a";
     let workers = make_worker_registry_with_worker(worker_name).await;
     let mut delivery_states = HashMap::from([(
@@ -264,32 +304,120 @@ async fn inbound_queue_eviction_surfaces_dropped_message() {
         assert_eq!(result.evicted_from, None);
     }
 
-    let result = queue_inbound_for_delivery_mode(
-        &mut delivery_states,
-        &workers,
-        worker_name,
-        inbound_ctx("evt_overflow"),
-    );
+    let before = delivery_states
+        .get(worker_name)
+        .expect("state should exist")
+        .pending_snapshot();
+    let rejected_deliver = fleet_deliver(1);
+    let mut rejected_ctx = inbound_ctx("message-1");
+    rejected_ctx.relaycast_receipt = held_fleet_message(&rejected_deliver).relaycast_receipt;
+    let result =
+        queue_inbound_for_delivery_mode(&mut delivery_states, &workers, worker_name, rejected_ctx);
 
-    assert_eq!(result.outcome, InboundQueueOutcome::Queued);
-    assert_eq!(
-        result.evicted_from.as_deref(),
-        Some("Alice"),
-        "hitting the per-worker cap must surface the evicted sender so callers can emit delivery_dropped"
-    );
+    assert_eq!(result.outcome, InboundQueueOutcome::RejectedFull);
+    assert_eq!(result.evicted_from, None);
     assert_eq!(
         delivery_states
             .get(worker_name)
             .expect("state should exist")
-            .pending_snapshot()
-            .len(),
-        crate::types::MAX_PENDING_PER_WORKER,
-        "queue stays at the cap after eviction"
+            .pending_snapshot(),
+        before,
+        "a full queue must remain byte-for-byte unchanged"
     );
+    let delivery_book = FleetDeliveryBook::default();
+    assert_eq!(delivery_book.received_up_to_seq("agent-worker-a"), 0);
+    assert_eq!(delivery_book.acked_up_to_seq("agent-worker-a"), 0);
 
     cleanup_worker_registry(workers).await;
 }
 
+#[tokio::test]
+async fn manual_flush_injects_and_acks_multiple_sequences_in_fifo_order() {
+    let worker_name = WorkerName::from("worker-a");
+    let mut workers = make_worker_registry_with_worker(&worker_name).await;
+    let first = fleet_deliver(1);
+    let second = fleet_deliver(2);
+    let first_message = held_fleet_message(&first);
+    let second_message = held_fleet_message(&second);
+    let mut state = InboundDeliveryState::new(InboundDeliveryMode::ManualFlush);
+    state.accept_inbound(first_message);
+    state.accept_inbound(second_message);
+    let mut delivery_states = HashMap::from([(worker_name.clone(), state)]);
+    let mut delivery_book = FleetDeliveryBook::default();
+    delivery_book.commit_received(&first);
+    delivery_book.commit_received(&second);
+    let (fleet_control_tx, mut fleet_control_rx) = mpsc::channel(4);
+
+    let result = super::fleet::flush_pending_relay_messages(
+        &mut delivery_states,
+        &mut workers,
+        &mut delivery_book,
+        &fleet_control_tx,
+        &worker_name,
+        Duration::from_secs(1),
+    )
+    .await;
+
+    assert_eq!(result.flushed, 2);
+    assert_eq!(result.failure, None);
+    assert!(delivery_states[&worker_name].pending.is_empty());
+    assert_eq!(delivery_book.received_up_to_seq("agent-worker-a"), 2);
+    assert_eq!(delivery_book.acked_up_to_seq("agent-worker-a"), 2);
+    for expected_seq in [1, 2] {
+        match fleet_control_rx.recv().await {
+            Some(FleetControlCommand::Send(BrokerToRelaycast::DeliveryAck(ack))) => {
+                assert_eq!(ack.agent, worker_name);
+                assert_eq!(ack.up_to_seq, expected_seq);
+            }
+            other => panic!("expected delivery ACK {expected_seq}, got {other:?}"),
+        }
+    }
+    assert!(fleet_control_rx.try_recv().is_err());
+
+    cleanup_worker_registry(workers).await;
+}
+
+#[tokio::test]
+async fn manual_flush_failure_retains_failed_message_and_suffix_without_ack() {
+    let worker_name = WorkerName::from("worker-a");
+    let (worker_event_tx, _worker_event_rx) = mpsc::channel::<WorkerEvent>(4);
+    let mut workers = WorkerRegistry::new(
+        worker_event_tx,
+        Vec::new(),
+        PathBuf::from("/tmp/agent-relay-broker-tests"),
+        Instant::now(),
+    );
+    let first = fleet_deliver(1);
+    let second = fleet_deliver(2);
+    let expected = vec![held_fleet_message(&first), held_fleet_message(&second)];
+    let mut state = InboundDeliveryState::new(InboundDeliveryMode::ManualFlush);
+    for message in expected.iter().cloned() {
+        state.accept_inbound(message);
+    }
+    let mut delivery_states = HashMap::from([(worker_name.clone(), state)]);
+    let mut delivery_book = FleetDeliveryBook::default();
+    delivery_book.commit_received(&first);
+    delivery_book.commit_received(&second);
+    let (fleet_control_tx, mut fleet_control_rx) = mpsc::channel(4);
+
+    let result = super::fleet::flush_pending_relay_messages(
+        &mut delivery_states,
+        &mut workers,
+        &mut delivery_book,
+        &fleet_control_tx,
+        &worker_name,
+        Duration::from_millis(50),
+    )
+    .await;
+
+    assert_eq!(result.flushed, 0);
+    assert!(result.failure.is_some());
+    assert_eq!(delivery_states[&worker_name].pending_snapshot(), expected);
+    assert_eq!(delivery_book.received_up_to_seq("agent-worker-a"), 2);
+    assert_eq!(delivery_book.acked_up_to_seq("agent-worker-a"), 0);
+    assert!(fleet_control_rx.try_recv().is_err());
+}
+
 fn make_pending_delivery(delivery_id: &str, worker: &str) -> PendingDelivery {
     PendingDelivery {
         worker_name: WorkerName::from(worker),
diff --git a/crates/broker/src/types.rs b/crates/broker/src/types.rs
index 2702c687d..279814150 100644
--- a/crates/broker/src/types.rs
+++ b/crates/broker/src/types.rs
@@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
 use crate::ids::{
-    AgentId, EventId, MessageTarget, ThreadId, WorkerName, WorkspaceAlias, WorkspaceId,
+    AgentId, DeliveryId, EventId, MessageTarget, ThreadId, WorkerName, WorkspaceAlias, WorkspaceId,
 };
 use crate::protocol::MessageInjectionMode;
 
@@ -93,6 +93,20 @@ pub struct PendingRelayMessage {
     /// telemetry / dedup parity with the auto-inject path.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub event_id: Option<EventId>,
+    /// Relaycast delivery metadata for messages received over node control.
+    /// Manual-flush messages retain this receipt until PTY injection so the
+    /// broker can advance the cumulative ACK only across an injected prefix.
+    #[serde(skip)]
+    pub relaycast_receipt: Option<RelaycastDeliveryReceipt>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RelaycastDeliveryReceipt {
+    pub agent: WorkerName,
+    pub agent_id: AgentId,
+    pub delivery_id: DeliveryId,
+    pub msg_id: EventId,
+    pub seq: u64,
 }
 
 fn default_priority() -> u8 {
@@ -343,6 +357,7 @@ mod inbound_delivery_tests {
             mode: MessageInjectionMode::Wait,
             queued_at_ms: 0,
             event_id: None,
+            relaycast_receipt: None,
         }
     }
 
@@ -400,6 +415,7 @@ mod inbound_delivery_tests {
             mode: MessageInjectionMode::Steer,
             queued_at_ms: 123_456,
             event_id: Some(EventId::new("evt_xyz")),
+            relaycast_receipt: None,
         };
         let mut state = InboundDeliveryState::new(InboundDeliveryMode::ManualFlush);
         state.accept_inbound(queued.clone());
diff --git a/specs/fleet-delivery.md b/specs/fleet-delivery.md
index bb9219aa4..0a21f82e6 100644
--- a/specs/fleet-delivery.md
+++ b/specs/fleet-delivery.md
@@ -137,7 +137,7 @@ Consequence: persistence _across process death_ is a **resumable-only** property
 ### 8.4 Where state lives
 
 - **Relaycast** holds all durable state (source of truth): mailboxes, agent records (`resumable`, `session_ref`, origin node), locations, node registry. Must survive Relaycast restarts.
-- **Broker** keeps only in-memory per-session state: `seq` cursor, dedup set, local pending-injection queue. **No disk needed for delivery durability.**
+- **Broker** keeps only in-memory per-session state: separate contiguous `received` and cumulative `acked` sequence cursors, a dedup set, and the local pending-injection queue. **No disk needed for delivery durability.**
   - Uplink blip, broker alive → cursor/dedup survive → clean replay, no duplicates.
   - Broker process dies → its child agents die too. When a resumable identity re-registers, Relaycast returns that identity's authoritative cumulative ACK cursor before replaying pending delivery, so the fresh broker can continue at `cursor + 1` without accepting an arbitrary first sequence.
 
@@ -157,6 +157,14 @@ the legacy fresh-session rule (`seq == 1`) and continues rejecting gaps. Relayca
 omits the field for nodes that did not advertise the capability so older brokers
 with strict reply decoders remain compatible.
 
+The authoritative cursor initializes `received == acked`. Auto-inject advances
+both only after PTY injection. `manual_flush` may advance `received` as it accepts
+multiple contiguous frames into its bounded FIFO, but leaves `acked` unchanged;
+duplicates and gaps report only the `acked` cursor. A flush advances `acked` only
+across the successfully injected FIFO prefix, and a full FIFO rejects the newest
+frame without advancing either cursor. If the broker dies, received-only state is
+discarded and Relaycast replays from its authoritative ACK cursor on registration.
+
 ### 8.5 One durable store
 
 The per-agent mailbox **subsumes** any per-node replay buffer: node-disconnect replay is just "redeliver this node's agents' unacked mail on reconnect." The per-location `seq` + ack is the at-least-once transport on top of the mailbox.