fix: emit one Queue terminal response

src/api/runtime_ingress/domain_frame_dispatcher.rs

@@ -460,8 +460,19 @@ impl DomainFrameDispatcher<'_> {
         domain: DispatchDomain,
         router: &crate::runtime::Router,
         error: &crate::runtime::router::RouteError,
+        reply_claim: &crate::runtime::envelope::ReplyClaim,
     ) -> IngressDecision {
         obs::counter_inc(obs::METRIC_INGRESS_DOMAIN_DISPATCH_TIMEOUTS);
+        if domain == DispatchDomain::Queue && !reply_claim.try_claim() {
+            warn!(
+                session_id = session_id,
+                domain = domain.as_str(),
+                error = %error,
+                outcome = "domain-response-won",
+                "Ingress: domain dispatch timed out after its terminal response was claimed"
+            );
+            return IngressDecision::Accept;
+        }
         warn!(
             session_id = session_id,
             domain = domain.as_str(),
@@ -618,6 +629,7 @@ impl DomainFrameDispatcher<'_> {
                     source: source.clone(),
                     destination: addr.clone(),
                 });
+            let reply_claim = envelope.reply_claim();
 
             debug!(
                 session_id = session_id,
@@ -671,6 +683,7 @@ impl DomainFrameDispatcher<'_> {
                         domain,
                         router,
                         &error,
+                        &reply_claim,
                     ));
                 }
                 Err(error) => {

src/api/runtime_ingress/tests/domain_backpressure.rs

@@ -216,6 +216,91 @@ impl MailboxSink for AlwaysTimingOutSink {
     }
 }
 
+struct QueueReplyThenTimeoutSink {
+    router: Arc<crate::runtime::Router>,
+    session_id: u64,
+}
+
+impl MailboxSink for QueueReplyThenTimeoutSink {
+    fn deliver(&self, envelope: Envelope) -> Result<(), DeliveryError> {
+        assert!(envelope.try_claim_reply(), "queue response claim");
+        let response = envelope
+            .try_reply_to(FrameContext::new(
+                self.session_id,
+                ChannelId::Pub,
+                crate::protocol::tlv::MessageType::new(200),
+                Bytes::from_static(&[0]),
+                RouteFamily::new(1),
+            ))
+            .expect("queue response envelope");
+        self.router.route(response).expect("route queue response");
+        Err(DeliveryError::Timeout)
+    }
+
+    fn deliver_high_priority(&self, envelope: Envelope) -> Result<(), DeliveryError> {
+        self.deliver(envelope)
+    }
+}
+
+#[test]
+fn should_not_emit_a_second_queue_terminal_response_after_the_domain_replied() {
+    // Arrange
+    // A Queue command can finish at the same instant its mailbox reply wait
+    // expires. Its response and ingress' indeterminate timeout compete for one
+    // terminal-response slot; emitting both shifts the client's per-type FIFO
+    // and can make a later accepted enqueue look retryably rejected.
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    let router = Arc::new(crate::runtime::Router::new());
+    let session_id = 6_500;
+    let client_frames = Arc::new(Mutex::new(Vec::<FrameContext>::new()));
+    router.register(
+        crate::runtime::routing::RouteAddress::new(
+            RouteFamily::new(1),
+            crate::runtime::routing::Route::new(format!("inbox://session/{session_id}")),
+        ),
+        Arc::new(CapturingInboxSink {
+            frames: client_frames.clone(),
+        }) as Arc<dyn MailboxSink>,
+    );
+    router.register_domain_pattern(
+        "queue",
+        Arc::new(QueueReplyThenTimeoutSink {
+            router: router.clone(),
+            session_id,
+        }),
+    );
+    let ingress = RuntimeIngress::new(false).with_router(router);
+    let (_, payload) = crate::benchkit::extract_single_tlv_field(
+        &crate::benchkit::build_queue_enqueue("queue://test/app/jobs", b"job"),
+    );
+
+    // Act
+    let decision = rt.block_on(async {
+        ingress
+            .on_open(make_session_info(session_id, TransportKind::Tcp))
+            .await
+            .unwrap();
+        ingress
+            .on_frame(
+                session_id,
+                ChannelId::Pub,
+                crate::protocol::tlv::MessageType::new(200),
+                payload,
+            )
+            .await
+    });
+
+    // Assert
+    assert_eq!(decision, IngressDecision::Accept);
+    let frames = client_frames.lock().unwrap();
+    assert_eq!(
+        frames.len(),
+        1,
+        "one request must produce exactly one terminal response"
+    );
+    assert_eq!(frames[0].payload.as_ref(), &[0]);
+}
+
 #[test]
 fn should_surface_sustained_high_lane_domain_mailbox_backpressure_for_each_domain() {
     // Arrange

src/domains/queue/sink/ingress.rs

@@ -217,7 +217,7 @@ impl QueueDomainCore {
             crate::domains::queue::QueueResponse::ReceivedRouted { messages } if messages.is_empty()
         ) {
             if let Some(wait_seconds) = wait_seconds.filter(|seconds| *seconds > 0) {
-                if let Some(source) = envelope.source() {
+                if envelope.source().is_some() {
                     let mut message = pending_message;
                     if let crate::domains::queue::protocol::QueueMessage::Receive {
                         wait_seconds,
@@ -230,11 +230,7 @@ impl QueueDomainCore {
                         .checked_add(Duration::from_secs(wait_seconds))
                         .unwrap_or_else(Instant::now);
                     self.pending_reserves.lock().push_back(PendingQueueReserve {
-                        envelope: Envelope::from_route(
-                            source.clone(),
-                            envelope.destination().clone(),
-                            (),
-                        ),
+                        envelope: envelope.clone_for_deferred_reply(),
                         meta,
                         request_started,
                         message,

src/domains/queue/sink/responses.rs

@@ -19,6 +19,14 @@ impl QueueDomainCore {
         meta: crate::runtime::ClientFrameMeta,
         response: &crate::domains::queue::QueueResponse,
     ) -> bool {
+        if request_envelope.source().is_none() || !request_envelope.try_claim_reply() {
+            tracing::debug!(
+                domain = "queue",
+                session = meta.session_id,
+                "Suppressed Queue response after another terminal response won"
+            );
+            return false;
+        }
         let delivered;
         #[cfg(test)]
         {

src/runtime/envelope.rs

@@ -32,7 +32,8 @@
 use crate::runtime::routing::RouteAddress;
 use std::any::Any;
 use std::fmt;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::Arc;
 use std::time::Instant;
 
 /// Envelope metadata without the payload (for zero-copy causation tracking)
@@ -116,6 +117,9 @@ pub struct Envelope {
     /// Mailbox enqueue time, set when the envelope is accepted by a mailbox.
     queued_at: Option<Instant>,
 
+    /// Shared one-shot claim for the terminal response to this request.
+    reply_claim: ReplyClaim,
+
     /// Type-erased message payload (must be Send + Sync)
     payload: Box<dyn Any + Send + Sync>,
 }
@@ -130,6 +134,7 @@ impl Envelope {
             causation: None,
             deadline: None,
             queued_at: None,
+            reply_claim: ReplyClaim::default(),
             payload: Box::new(payload),
         }
     }
@@ -147,6 +152,7 @@ impl Envelope {
             causation: None,
             deadline: None,
             queued_at: None,
+            reply_claim: ReplyClaim::default(),
             payload: Box::new(payload),
         }
     }
@@ -193,6 +199,7 @@ impl Envelope {
             causation: Some(self.id),
             deadline: self.deadline,
             queued_at: None,
+            reply_claim: ReplyClaim::default(),
             payload: Box::new(payload),
         }
     }
@@ -211,10 +218,38 @@ impl Envelope {
             causation: Some(self.id),
             deadline: self.deadline,
             queued_at: None,
+            reply_claim: ReplyClaim::default(),
             payload: Box::new(payload),
         })
     }
 
+    /// Share this request's one-shot terminal-response claim with its owner.
+    #[inline]
+    pub(crate) fn reply_claim(&self) -> ReplyClaim {
+        self.reply_claim.clone()
+    }
+
+    /// Claim the right to emit this request's terminal response.
+    #[inline]
+    pub(crate) fn try_claim_reply(&self) -> bool {
+        self.reply_claim.try_claim()
+    }
+
+    /// Retain this request's routing and response identity for a deferred reply.
+    #[must_use]
+    pub(crate) fn clone_for_deferred_reply(&self) -> Self {
+        Self {
+            id: self.id,
+            source: self.source.clone(),
+            destination: self.destination.clone(),
+            causation: self.causation,
+            deadline: self.deadline,
+            queued_at: None,
+            reply_claim: self.reply_claim.clone(),
+            payload: Box::new(()),
+        }
+    }
+
     /// Get the message ID
     #[inline]
     #[must_use]
@@ -321,6 +356,20 @@ impl Envelope {
     }
 }
 
+/// Coordinates the single terminal response for a request that may outlive
+/// the ingress dispatch deadline.
+#[derive(Clone, Debug, Default)]
+pub(crate) struct ReplyClaim(Arc<AtomicBool>);
+
+impl ReplyClaim {
+    #[inline]
+    pub(crate) fn try_claim(&self) -> bool {
+        self.0
+            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+            .is_ok()
+    }
+}
+
 impl fmt::Debug for Envelope {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Envelope")
@@ -330,6 +379,7 @@ impl fmt::Debug for Envelope {
             .field("causation", &self.causation)
             .field("deadline", &self.deadline)
             .field("queued_at", &self.queued_at)
+            .field("reply_claim", &self.reply_claim)
             .field("payload", &"<type-erased>")
             .finish()
     }
@@ -514,4 +564,24 @@ mod tests {
         // Assert
         assert_eq!(reply.deadline(), Some(deadline));
     }
+
+    #[test]
+    fn should_share_one_terminal_reply_claim_with_deferred_context() {
+        // Arrange
+        let original = Envelope::from_route(
+            test_address(1, "/test/source"),
+            test_address(1, "/test/destination"),
+            "request",
+        );
+        let deferred = original.clone_for_deferred_reply();
+        let owner = original.reply_claim();
+
+        // Act
+        let first = deferred.try_claim_reply();
+        let second = owner.try_claim();
+
+        // Assert
+        assert!(first);
+        assert!(!second, "the request may emit only one terminal response");
+    }
 }