fix: preserve available post-cancel receipts on the existing event queue

The frozen01a default reference corpus caught cancel_midstream losing5 healthy-queue observations (12 passed1 failed); retain that golden unchanged. Existing send_event delivers immediately available capacity before its cancel-aware Full wait, while stream, admission and Pause reservations retain strict cancellation. Extend the real sender regression to prove available post-cancel receipt delivery plus denied stream suffix. Genuine production cancellation-off control with unchanged7 Engine fixtures produced1 passed6 bounded failures11.51s; byte-exact restore7 passed0 failed0.47s. Combined33 focused qualification and scoped CI-policy Clippy pending. Parent hooks typed-pair lint fix preserved separately4ec714c72.

Signed-off-by: CodeWhale Bot <[email protected]>

crates/tui/src/core/engine/streaming.rs

@@ -47,9 +47,11 @@ impl super::Engine {
         }
     }
 
-    /// Every instance emitter uses the same queue authority. The existing
-    /// TurnControl owns cancellation; an idle refresh must not inherit the
-    /// token of an earlier interrupted turn.
+    /// Every instance emitter uses the same queue authority. Available
+    /// capacity preserves post-cancel usage/status receipts; cancellation
+    /// releases a wait for capacity. An idle refresh must not inherit the
+    /// token of an earlier interrupted turn. Stream/admission/terminal handoff
+    /// reservations retain their strict cancellation floor.
     pub(super) async fn send_event(&self, event: super::Event) -> Result<(), EventSendError> {
         let cancel = self
             .turn_controls
@@ -58,7 +60,15 @@ impl super::Engine {
             .active
             .as_ref()
             .map(|control| control.cancel.clone());
-        let permit = reserve_event_capacity(&self.tx_event, cancel.as_ref()).await?;
+        let permit = match self.tx_event.clone().try_reserve_owned() {
+            Ok(permit) => permit,
+            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
+                return Err(EventSendError::Closed);
+            }
+            Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
+                reserve_event_capacity(&self.tx_event, cancel.as_ref()).await?
+            }
+        };
         permit.send(event);
         Ok(())
     }

crates/tui/src/core/engine/tests.rs

@@ -337,6 +337,28 @@ async fn event_capacity_releases_all_admitted_senders_and_keeps_idle_receipts_lo
         .unwrap();
     assert_eq!(outcomes, vec![Err(streaming::EventSendError::Cancelled); 8]);
     drop(senders);
+    assert!(matches!(rx.try_recv(), Ok(Event::Status { .. })));
+    // A cancelled turn still delivers its usage/status receipts when they
+    // fit immediately. The stream suffix remains strictly cancelled.
+    engine
+        .send_event(Event::status(
+            "cancelled turn receipt with available capacity",
+        ))
+        .await
+        .unwrap();
+    assert!(
+        matches!(rx.try_recv(), Ok(Event::Status { message, .. }) if message == "cancelled turn receipt with available capacity")
+    );
+    assert!(
+        !engine
+            .send_stream_event(Event::status("forbidden stream suffix"))
+            .await
+    );
+    assert!(rx.try_recv().is_err());
+    engine
+        .tx_event
+        .try_send(Event::status("occupied before idle receipt"))
+        .unwrap();
     drop(turn);
     let mut idle = Box::pin(engine.send_event(Event::status("idle receipt after cancelled turn")));
     assert!(