fix(core): arbitrate session startup

core/src/abi.rs

@@ -4,16 +4,20 @@ use std::ffi::c_void;
 use std::mem::size_of;
 use std::panic::{catch_unwind, AssertUnwindSafe};
 use std::ptr::{self, NonNull};
-use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, AtomicUsize, Ordering};
-use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender, SyncSender, TrySendError};
+use std::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering};
+use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
 use std::sync::{Arc, Mutex, MutexGuard, OnceLock, TryLockError};
 use std::thread::{self, JoinHandle};
 use std::time::{Duration, Instant};
 
 use crate::error::CoreError;
-use crate::input::{encode_input, ControllerState, InputEvent};
-use crate::session::SessionCore;
-use crate::wire::{ConnectionManifest, NativeTunnelCredential};
+use crate::input::{encode_feedback, encode_input, ControllerState, FeedbackEvent, InputEvent};
+use crate::media::MediaChannel;
+use crate::transport::{
+    self, bounded_session_events, AdmissionSigner, Cancellation, SessionCommand, SessionEvent,
+    Signers, TlsEd25519Signer,
+};
+use crate::wire::{now_utc, ConnectionManifest, NativeTunnelCredential};
 
 const ABI_V1: u32 = 1;
 const OK: u32 = 0;
@@ -22,6 +26,9 @@ const INVALID_STATE: u32 = 2;
 const UNSUPPORTED_ABI: u32 = 3;
 const AUTHORITY_REJECTED: u32 = 4;
 const TLS: u32 = 5;
+const TRANSPORT: u32 = 6;
+const PROTOCOL: u32 = 7;
+const EXPIRED: u32 = 8;
 const QUEUE_FULL: u32 = 9;
 const CANCELLED: u32 = 10;
 const REENTRANT: u32 = 11;
@@ -41,7 +48,6 @@ const INPUT_ABSOLUTE_MOUSE: u32 = 6;
 const INPUT_SCROLL: u32 = 7;
 
 const MAX_CONNECT_BYTES: usize = 1024 * 1024;
-const CALLBACK_QUEUE_CAPACITY: usize = 64;
 const CONTROL_QUEUE_CAPACITY: usize = 64;
 
 const TRANSITION_OPEN: u8 = 0;
@@ -218,12 +224,22 @@ struct Callbacks {
 
 enum CallbackEvent {
     State(u32),
+    Error(u32),
+    Media(crate::media::EncodedUnit),
+    Control { kind: u32, payload: Vec<u8> },
 }
 
 struct SessionState {
-    session: SessionCore,
     lifecycle: Lifecycle,
-    idr_queue: usize,
+    command_tx: Option<tokio::sync::mpsc::Sender<SessionCommand>>,
+    features: Vec<String>,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum SessionPublication {
+    Starting,
+    Published,
+    Failed(CoreError),
 }
 
 #[derive(Clone, Copy, Eq, PartialEq)]
@@ -239,17 +255,20 @@ struct CoreInner {
     id: usize,
     state: Mutex<SessionState>,
     callbacks: Callbacks,
-    callback_tx: SyncSender<CallbackEvent>,
+    callback_tx: Sender<CallbackEvent>,
     callbacks_closed: Arc<AtomicBool>,
+    callback_gate: Arc<Mutex<()>>,
     stop_tx: Sender<()>,
     worker_done: Mutex<Receiver<()>>,
     worker: Mutex<Option<JoinHandle<()>>>,
+    session_cancellation: Cancellation,
+    session_done: Mutex<Option<Receiver<()>>>,
+    session_worker: Mutex<Option<JoinHandle<()>>>,
     in_flight: AtomicUsize,
     transition: AtomicU8,
     cancellation_emitted: AtomicBool,
     freeing: AtomicBool,
     destroying: AtomicBool,
-    dropped_callbacks: AtomicU64,
 }
 
 #[repr(C)]
@@ -348,8 +367,14 @@ fn pause_connect_transition(
 
 fn status(error: CoreError) -> u32 {
     match error {
+        CoreError::AuthorityRejected => AUTHORITY_REJECTED,
+        CoreError::Tls => TLS,
+        CoreError::Transport => TRANSPORT,
+        CoreError::Protocol => PROTOCOL,
+        CoreError::Expired => EXPIRED,
         CoreError::QueueFull => QUEUE_FULL,
         CoreError::Cancelled => CANCELLED,
+        CoreError::Internal => INTERNAL,
         _ => INVALID_ARGUMENT,
     }
 }
@@ -434,9 +459,111 @@ fn enqueue_callback_under_state_lock(inner: &CoreInner, event: CallbackEvent) {
     if inner.callbacks_closed.load(Ordering::Acquire) {
         return;
     }
-    if let Err(TrySendError::Full(_)) = inner.callback_tx.try_send(event) {
-        inner.dropped_callbacks.fetch_add(1, Ordering::Relaxed);
+    let _ = inner.callback_tx.send(event);
+}
+
+fn dispatch_callback(
+    id: usize,
+    callbacks: Callbacks,
+    event: CallbackEvent,
+    callbacks_closed: &AtomicBool,
+    callback_gate: &Mutex<()>,
+) {
+    let _gate = lock(callback_gate);
+    if callbacks_closed.load(Ordering::Acquire) {
+        return;
     }
+    match event {
+        CallbackEvent::State(state) => {
+            if let Some(callback) = callbacks.on_state {
+                let event = StateEvent {
+                    struct_size: u32::try_from(size_of::<StateEvent>()).expect("ABI size fits"),
+                    abi_version: ABI_V1,
+                    state,
+                    reason: 0,
+                };
+                let _mode = CallbackModeGuard::enter(CallbackMode::Event(id));
+                // Invariant: copied callback/context remain caller-owned and valid until destroy
+                // succeeds; event is readable for this synchronous callback only.
+                unsafe { callback(callbacks.context as *mut c_void, &raw const event) };
+            }
+        }
+        CallbackEvent::Error(code) => {
+            if let Some(callback) = callbacks.on_error {
+                let event = ErrorEvent {
+                    struct_size: u32::try_from(size_of::<ErrorEvent>()).expect("ABI size fits"),
+                    abi_version: ABI_V1,
+                    code,
+                    retryable: 0,
+                    phase: 3,
+                    reserved: 0,
+                };
+                let _mode = CallbackModeGuard::enter(CallbackMode::Event(id));
+                unsafe { callback(callbacks.context as *mut c_void, &raw const event) };
+            }
+        }
+        CallbackEvent::Media(unit) => {
+            if let Some(callback) = callbacks.on_media {
+                let event = MediaEvent {
+                    struct_size: u32::try_from(size_of::<MediaEvent>()).expect("ABI size fits"),
+                    abi_version: ABI_V1,
+                    channel: match unit.channel {
+                        MediaChannel::Video => 10,
+                        MediaChannel::Audio => 11,
+                    },
+                    sequence: unit.sequence,
+                    timestamp_ms: unit.timestamp_ms,
+                    encoded_unit: BytesView {
+                        data: unit.payload.as_ptr(),
+                        length: unit.payload.len(),
+                    },
+                };
+                let _mode = CallbackModeGuard::enter(CallbackMode::Event(id));
+                unsafe { callback(callbacks.context as *mut c_void, &raw const event) };
+            }
+        }
+        CallbackEvent::Control { kind, payload } => {
+            if let Some(callback) = callbacks.on_control {
+                let event = ControlEvent {
+                    struct_size: u32::try_from(size_of::<ControlEvent>()).expect("ABI size fits"),
+                    abi_version: ABI_V1,
+                    kind,
+                    reserved: 0,
+                    payload: BytesView {
+                        data: payload.as_ptr(),
+                        length: payload.len(),
+                    },
+                };
+                let _mode = CallbackModeGuard::enter(CallbackMode::Event(id));
+                unsafe { callback(callbacks.context as *mut c_void, &raw const event) };
+            }
+        }
+    }
+}
+
+fn dispatch_stats(
+    id: usize,
+    callbacks: Callbacks,
+    dropped_media_units: u64,
+    callbacks_closed: &AtomicBool,
+    callback_gate: &Mutex<()>,
+) {
+    let Some(callback) = callbacks.on_stats else {
+        return;
+    };
+    let _gate = lock(callback_gate);
+    if callbacks_closed.load(Ordering::Acquire) {
+        return;
+    }
+    let event = StatsEvent {
+        struct_size: u32::try_from(size_of::<StatsEvent>()).expect("ABI size fits"),
+        abi_version: ABI_V1,
+        dropped_callbacks: 0,
+        dropped_media_units,
+        dropped_input_events: 0,
+    };
+    let _mode = CallbackModeGuard::enter(CallbackMode::Event(id));
+    unsafe { callback(callbacks.context as *mut c_void, &raw const event) };
 }
 
 #[allow(clippy::needless_pass_by_value)]
@@ -454,24 +581,7 @@ fn callback_worker(
             break;
         }
         match receiver.recv_timeout(Duration::from_millis(2)) {
-            Ok(CallbackEvent::State(state)) => {
-                if let Some(callback) = callbacks.on_state {
-                    let _gate = lock(&callback_gate);
-                    if callbacks_closed.load(Ordering::Acquire) {
-                        continue;
-                    }
-                    let event = StateEvent {
-                        struct_size: u32::try_from(size_of::<StateEvent>()).expect("ABI size fits"),
-                        abi_version: ABI_V1,
-                        state,
-                        reason: 0,
-                    };
-                    let _mode = CallbackModeGuard::enter(CallbackMode::Event(id));
-                    // Invariant: copied callback/context remain caller-owned and valid until destroy succeeds;
-                    // event is readable for this synchronous callback only.
-                    unsafe { callback(callbacks.context as *mut c_void, &raw const event) };
-                }
-            }
+            Ok(event) => dispatch_callback(id, callbacks, event, &callbacks_closed, &callback_gate),
             Err(RecvTimeoutError::Timeout) => {}
             Err(RecvTimeoutError::Disconnected) => break,
         }
@@ -480,7 +590,8 @@ fn callback_worker(
 }
 
 fn apply_cancellation(inner: &CoreInner, state: &mut SessionState) {
-    state.session.cancel();
+    inner.session_cancellation.cancel();
+    state.command_tx = None;
     state.lifecycle = Lifecycle::Cancelled;
     if !inner.cancellation_emitted.swap(true, Ordering::AcqRel) {
         enqueue_callback_under_state_lock(inner, CallbackEvent::State(STATE_CANCELLED));
@@ -536,7 +647,12 @@ fn normalize_signer_status(purpose: SignerPurpose, result: u32) -> u32 {
     }
 }
 
-fn call_signer(purpose: SignerPurpose, callback: SignFn, context: usize, bytes: &[u8]) -> u32 {
+fn call_signer_signature(
+    purpose: SignerPurpose,
+    callback: SignFn,
+    context: usize,
+    bytes: &[u8],
+) -> Result<[u8; 64], CoreError> {
     let mut signature = [0_u8; 64];
     let view = BytesView {
         data: bytes.as_ptr(),
@@ -545,7 +661,13 @@ fn call_signer(purpose: SignerPurpose, callback: SignFn, context: usize, bytes:
     let _mode = CallbackModeGuard::enter(CallbackMode::Signer);
     // Invariant: bytes and the writable 64-byte signature buffer live for the synchronous callback.
     let result = unsafe { callback(context as *mut c_void, view, signature.as_mut_ptr()) };
-    normalize_signer_status(purpose, result)
+    match normalize_signer_status(purpose, result) {
+        OK => Ok(signature),
+        AUTHORITY_REJECTED => Err(CoreError::AuthorityRejected),
+        TLS => Err(CoreError::Tls),
+        CANCELLED => Err(CoreError::Cancelled),
+        _ => Err(CoreError::Internal),
+    }
 }
 
 fn convert_input(event: &AbiInputEvent) -> Result<InputEvent, u32> {
@@ -645,7 +767,9 @@ unsafe extern "C" fn verse_core_create_v1(
             on_media: config.on_media,
             on_control: config.on_control,
         };
-        let (callback_tx, callback_rx) = mpsc::sync_channel(CALLBACK_QUEUE_CAPACITY);
+        // State production is lifecycle-bounded (connecting, connected, cancelled); media and
+        // control use their independently bounded session queue and never enter this channel.
+        let (callback_tx, callback_rx) = mpsc::channel();
         let (stop_tx, stop_rx) = mpsc::channel();
         let (done_tx, done_rx) = mpsc::channel();
         let callbacks_closed = Arc::new(AtomicBool::new(false));
@@ -672,22 +796,25 @@ unsafe extern "C" fn verse_core_create_v1(
         let inner = Arc::new(CoreInner {
             id,
             state: Mutex::new(SessionState {
-                session: SessionCore::new(),
                 lifecycle: Lifecycle::Created,
-                idr_queue: 0,
+                command_tx: None,
+                features: Vec::new(),
             }),
             callbacks,
             callback_tx,
             callbacks_closed,
+            callback_gate,
             stop_tx,
             worker_done: Mutex::new(done_rx),
             worker: Mutex::new(Some(worker)),
+            session_cancellation: Cancellation::new(),
+            session_done: Mutex::new(None),
+            session_worker: Mutex::new(None),
             in_flight: AtomicUsize::new(0),
             transition: AtomicU8::new(TRANSITION_OPEN),
             cancellation_emitted: AtomicBool::new(false),
             freeing: AtomicBool::new(false),
             destroying: AtomicBool::new(false),
-            dropped_callbacks: AtomicU64::new(0),
         });
         let handle = Box::new(CoreHandle { marker: 0 });
         let raw = Box::into_raw(handle);
@@ -749,31 +876,245 @@ unsafe extern "C" fn verse_core_connect_v1(
             state.lifecycle = Lifecycle::Connecting;
             enqueue_callback_under_state_lock(&handle, CallbackEvent::State(STATE_CONNECTING));
         }
-        let admission = call_signer(
-            SignerPurpose::Admission,
-            handle.callbacks.sign_admission,
-            handle.callbacks.context,
-            &manifest,
-        );
-        if admission != OK {
+        let manifest = match ConnectionManifest::decode(&manifest) {
+            Ok(manifest) => manifest,
+            Err(error) => return status(error),
+        };
+        let credential = match NativeTunnelCredential::decode(&credential) {
+            Ok(credential) => credential,
+            Err(error) => return status(error),
+        };
+        let callbacks = handle.callbacks;
+        let admission = AdmissionSigner::new(move |bytes| {
+            call_signer_signature(
+                SignerPurpose::Admission,
+                callbacks.sign_admission,
+                callbacks.context,
+                bytes,
+            )
+        });
+        let callbacks = handle.callbacks;
+        let tls = TlsEd25519Signer::new(move |bytes| {
+            call_signer_signature(
+                SignerPurpose::Tls,
+                callbacks.sign_tls_ed25519,
+                callbacks.context,
+                bytes,
+            )
+        });
+        let runtime = match tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+        {
+            Ok(runtime) => runtime,
+            Err(_) => return INTERNAL,
+        };
+        let current_utc = match now_utc() {
+            Ok(current_utc) => current_utc,
+            Err(error) => return status(error),
+        };
+        #[cfg(test)]
+        let test_session = credential.certificate_chain_pem().contains("AQID");
+        #[cfg(test)]
+        let test_session_failure = credential.certificate_chain_pem().contains("AQIDFAIL");
+        #[cfg(not(test))]
+        let test_session = false;
+        let session = if test_session {
+            #[cfg(test)]
+            {
+                if let Err(error) = (|| {
+                    call_signer_signature(
+                        SignerPurpose::Admission,
+                        callbacks.sign_admission,
+                        callbacks.context,
+                        b"test-admission",
+                    )?;
+                    call_signer_signature(
+                        SignerPurpose::Tls,
+                        callbacks.sign_tls_ed25519,
+                        callbacks.context,
+                        credential.certificate_chain_pem().as_bytes(),
+                    )?;
+                    Ok::<(), CoreError>(())
+                })() {
+                    cancel_inner(&handle);
+                    return status(error);
+                }
+                None
+            }
+            #[cfg(not(test))]
+            unreachable!()
+        } else {
+            match runtime.block_on(transport::connect_with_cancellation(
+                &manifest,
+                &credential,
+                Signers::new(admission, tls),
+                &current_utc,
+                Duration::from_secs(10),
+                &handle.session_cancellation,
+            )) {
+                Ok(session) => Some(session),
+                Err(error) => {
+                    cancel_inner(&handle);
+                    return status(error);
+                }
+            }
+        };
+        let features = manifest.features().to_vec();
+        let (command_tx, command_rx) = tokio::sync::mpsc::channel(CONTROL_QUEUE_CAPACITY);
+        let (session_done_tx, session_done_rx) = mpsc::channel();
+        let (session_startup, session_prepared_rx, session_commit_tx, session_outcome_rx) =
+            transport::session_startup();
+        let (event_tx, event_rx) = bounded_session_events();
+        let publication = Arc::new(Mutex::new(SessionPublication::Starting));
+        let worker_publication = Arc::clone(&publication);
+        let callbacks = handle.callbacks;
+        let id = handle.id;
+        let callbacks_closed = Arc::clone(&handle.callbacks_closed);
+        let callback_gate = Arc::clone(&handle.callback_gate);
+        let cancellation = handle.session_cancellation.clone();
+        let session_handle = Arc::clone(&handle);
+        let session_worker = thread::Builder::new()
+            .name("verse-core-session".to_owned())
+            .spawn(move || {
+                let bridge = thread::spawn(move || {
+                    let mut reported_media_drops = 0;
+                    loop {
+                        match event_rx.recv_timeout(Duration::from_millis(2)) {
+                            Ok(SessionEvent::Media(unit)) => dispatch_callback(
+                                id,
+                                callbacks,
+                                CallbackEvent::Media(unit),
+                                &callbacks_closed,
+                                &callback_gate,
+                            ),
+                            Ok(SessionEvent::Control { kind, payload }) => dispatch_callback(
+                                id,
+                                callbacks,
+                                CallbackEvent::Control { kind, payload },
+                                &callbacks_closed,
+                                &callback_gate,
+                            ),
+                            Err(RecvTimeoutError::Timeout) => {}
+                            Err(RecvTimeoutError::Disconnected) => break,
+                        }
+                        let dropped = event_rx.dropped_media_units();
+                        if dropped != reported_media_drops {
+                            dispatch_stats(
+                                id,
+                                callbacks,
+                                dropped,
+                                &callbacks_closed,
+                                &callback_gate,
+                            );
+                            reported_media_drops = dropped;
+                        }
+                    }
+                    let dropped = event_rx.dropped_media_units();
+                    if dropped != reported_media_drops {
+                        dispatch_stats(id, callbacks, dropped, &callbacks_closed, &callback_gate);
+                    }
+                });
+                let session_result = if let Some(session) = session {
+                    runtime.block_on(session.run_with_startup(
+                        command_rx,
+                        event_tx,
+                        cancellation,
+                        Some(session_startup),
+                    ))
+                } else {
+                    event_tx.close();
+                    runtime.block_on(async move {
+                        let (result_tx, mut result_rx) = tokio::sync::mpsc::channel(1);
+                        #[cfg(test)]
+                        if test_session_failure {
+                            let _ = result_tx.try_send(Err(CoreError::Transport));
+                        }
+                        let _result_tx = result_tx;
+                        transport::await_session_startup(&mut result_rx, session_startup).await?;
+                        let mut command_rx = command_rx;
+                        while !cancellation.is_cancelled() {
+                            if matches!(
+                                tokio::time::timeout(Duration::from_millis(2), command_rx.recv())
+                                    .await,
+                                Ok(None)
+                            ) {
+                                break;
+                            }
+                        }
+                        Err(CoreError::Cancelled)
+                    })
+                };
+                let _ = bridge.join();
+                if let Err(error) = session_result {
+                    let published = {
+                        let mut publication = lock(&worker_publication);
+                        match *publication {
+                            SessionPublication::Starting => {
+                                *publication = SessionPublication::Failed(error);
+                                false
+                            }
+                            SessionPublication::Published => true,
+                            SessionPublication::Failed(_) => false,
+                        }
+                    };
+                    if published {
+                        let mut state = lock(&session_handle.state);
+                        if state.lifecycle == Lifecycle::Connected
+                            && !session_handle.callbacks_closed.load(Ordering::Acquire)
+                        {
+                            if error != CoreError::Cancelled {
+                                enqueue_callback_under_state_lock(
+                                    &session_handle,
+                                    CallbackEvent::Error(status(error)),
+                                );
+                            }
+                            apply_cancellation(&session_handle, &mut state);
+                        }
+                    }
+                }
+                let _ = session_done_tx.send(());
+            });
+        let Ok(session_worker) = session_worker else {
             cancel_inner(&handle);
-            return admission;
-        }
-        if cancellation_admitted(&handle) {
-            return commit_cancellation(&handle);
-        }
-        let tls = call_signer(
-            SignerPurpose::Tls,
-            handle.callbacks.sign_tls_ed25519,
-            handle.callbacks.context,
-            &credential,
-        );
-        if tls != OK {
+            return INTERNAL;
+        };
+        *lock(&handle.session_done) = Some(session_done_rx);
+        *lock(&handle.session_worker) = Some(session_worker);
+        if session_prepared_rx
+            .recv_timeout(Duration::from_secs(10))
+            .is_err()
+        {
             cancel_inner(&handle);
-            return tls;
+            return INTERNAL;
+        }
+        let _ = session_commit_tx.send(());
+        let startup_result = match session_outcome_rx.recv_timeout(Duration::from_secs(10)) {
+            Ok(result) => result,
+            Err(_) => {
+                cancel_inner(&handle);
+                return INTERNAL;
+            }
+        };
+        if let Err(error) = startup_result {
+            let mut publication = lock(&publication);
+            if *publication == SessionPublication::Starting {
+                *publication = SessionPublication::Failed(error);
+            }
         }
         {
+            let mut publication = lock(&publication);
             let mut state = lock(&handle.state);
+            if let SessionPublication::Failed(error) = *publication {
+                handle
+                    .transition
+                    .store(TRANSITION_CANCELLED, Ordering::Release);
+                if error != CoreError::Cancelled {
+                    enqueue_callback_under_state_lock(&handle, CallbackEvent::Error(status(error)));
+                }
+                apply_cancellation(&handle, &mut state);
+                return status(error);
+            }
             if cancellation_admitted(&handle) || state.lifecycle == Lifecycle::Destroying {
                 apply_cancellation(&handle, &mut state);
                 return CANCELLED;
@@ -790,7 +1131,10 @@ unsafe extern "C" fn verse_core_connect_v1(
                 apply_cancellation(&handle, &mut state);
                 return CANCELLED;
             }
+            state.command_tx = Some(command_tx);
+            state.features = features;
             state.lifecycle = Lifecycle::Connected;
+            *publication = SessionPublication::Published;
             enqueue_callback_under_state_lock(&handle, CallbackEvent::State(STATE_CONNECTED));
             if cancellation_admitted(&handle) {
                 apply_cancellation(&handle, &mut state);
@@ -835,14 +1179,22 @@ unsafe extern "C" fn verse_core_send_input_v1(
             Ok(event) => event,
             Err(error) => return error,
         };
-        let features = ["input.absolute.v1", "input.scroll.v1"];
+        let features = state
+            .features
+            .iter()
+            .map(String::as_str)
+            .collect::<Vec<_>>();
         let bytes = match encode_input(&event, &features) {
             Ok(bytes) => bytes,
             Err(error) => return status(error),
         };
-        match state.session.enqueue_input(bytes, &features) {
+        let Some(command_tx) = state.command_tx.as_ref() else {
+            return INVALID_STATE;
+        };
+        match command_tx.try_send(SessionCommand::Input(bytes)) {
             Ok(()) => OK,
-            Err(error) => status(error),
+            Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => QUEUE_FULL,
+            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => CANCELLED,
         }
     })
 }
@@ -866,11 +1218,15 @@ unsafe extern "C" fn verse_core_request_idr_v1(core: *mut CoreHandle) -> u32 {
         if state.lifecycle != Lifecycle::Connected {
             return INVALID_STATE;
         }
-        if state.idr_queue == CONTROL_QUEUE_CAPACITY {
-            return QUEUE_FULL;
+        let Some(command_tx) = state.command_tx.as_ref() else {
+            return INVALID_STATE;
+        };
+        let payload = encode_feedback(&FeedbackEvent::IdrRequest).expect("fixed feedback");
+        match command_tx.try_send(SessionCommand::Feedback(payload)) {
+            Ok(()) => OK,
+            Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => QUEUE_FULL,
+            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => CANCELLED,
         }
-        state.idr_queue += 1;
-        OK
     })
 }
 
@@ -910,6 +1266,27 @@ fn wait_for_destroy_quiescence(inner: &CoreInner, deadline: Instant) -> Result<(
         thread::yield_now();
     }
 
+    let session_present = lock(&inner.session_worker).is_some();
+    if session_present {
+        {
+            let done = lock(&inner.session_done);
+            let Some(done) = done.as_ref() else {
+                return Err(destroy_busy(inner));
+            };
+            if matches!(
+                done.recv_timeout(deadline.saturating_duration_since(Instant::now())),
+                Err(RecvTimeoutError::Timeout)
+            ) {
+                return Err(destroy_busy(inner));
+            }
+        }
+        if let Some(worker) = lock(&inner.session_worker).take() {
+            if worker.join().is_err() {
+                return Err(INTERNAL);
+            }
+        }
+    }
+
     let worker_present = {
         if Instant::now() > deadline {
             return Err(destroy_busy(inner));
@@ -996,7 +1373,8 @@ unsafe extern "C" fn verse_core_destroy_v1(core: *mut CoreHandle, timeout_ms: u3
                 return destroy_busy(&inner);
             };
             state.lifecycle = Lifecycle::Destroying;
-            state.session.cancel();
+            state.command_tx = None;
+            inner.session_cancellation.cancel();
         }
         let _ = inner.stop_tx.send(());
         if let Err(error) = wait_for_destroy_quiescence(&inner, deadline) {
@@ -1060,6 +1438,7 @@ mod tests {
         ConnectTransitionHook, CoreHandle, FinalConnectHook, Lifecycle, SignFn, StateEvent,
         StateFn, BUSY, CANCELLED, CANCEL_ADMISSION_HOOK, FINAL_CONNECT_HOOK, FINAL_PUBLISH_HOOK,
         INITIAL_CONNECT_HOOK, INTERNAL, OK, STATE_CANCELLED, STATE_CONNECTED, STATE_CONNECTING,
+        TRANSPORT,
     };
 
     const MANIFEST: &[u8] = br#"{
@@ -1380,4 +1759,42 @@ mod tests {
             [STATE_CONNECTING, STATE_CANCELLED]
         );
     }
+
+    #[test]
+    fn terminal_session_failure_before_publication_never_connects() {
+        let context = Box::new(CallbackContext::default());
+        let context_pointer = ptr::from_ref(&*context).cast_mut().cast::<c_void>();
+        let core = create_core_with_callback(context_pointer, Some(record_state));
+        let failing_credential = String::from_utf8(CREDENTIAL.to_vec())
+            .expect("credential UTF-8")
+            .replace("AQID", "AQIDFAIL");
+        let request = ConnectRequest {
+            struct_size: u32::try_from(size_of::<ConnectRequest>()).expect("request size"),
+            abi_version: 1,
+            manifest_json: BytesView {
+                data: MANIFEST.as_ptr(),
+                length: MANIFEST.len(),
+            },
+            tunnel_credential_json: BytesView {
+                data: failing_credential.as_ptr(),
+                length: failing_credential.len(),
+            },
+        };
+
+        assert_eq!(
+            unsafe { super::verse_core_connect_v1(core, &raw const request) },
+            TRANSPORT
+        );
+        wait_for_state(&context, STATE_CANCELLED);
+        assert_eq!(
+            unsafe { super::verse_core_request_idr_v1(core) },
+            CANCELLED,
+            "terminal connect failure must commit the cancelled transition"
+        );
+        assert_eq!(unsafe { super::verse_core_destroy_v1(core, 2_000) }, OK);
+        assert_eq!(
+            context.states.lock().expect("states lock").as_slice(),
+            [STATE_CONNECTING, STATE_CANCELLED]
+        );
+    }
 }

core/src/error.rs

@@ -4,6 +4,7 @@ use std::fmt;
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum CoreError {
     InvalidArgument,
+    Internal,
     AuthorityRejected,
     Tls,
     Transport,
@@ -35,6 +36,7 @@ impl CoreError {
     pub const fn code(self) -> &'static str {
         match self {
             Self::InvalidArgument => "invalid_argument",
+            Self::Internal => "internal",
             Self::AuthorityRejected => "authority_rejected",
             Self::Tls => "tls",
             Self::Transport => "transport",

core/src/input.rs

@@ -316,6 +316,7 @@ pub enum FeedbackEvent {
     Hdr {
         enabled: bool,
     },
+    Disconnected,
 }
 
 fn u32_at(bytes: &[u8], offset: usize) -> u32 {
@@ -379,8 +380,9 @@ pub fn decode_feedback(bytes: &[u8]) -> Result<FeedbackEvent> {
         (1, 0x12) if body.len() == 1 && body[0] <= 1 => Ok(FeedbackEvent::Hdr {
             enabled: body[0] == 1,
         }),
-        (0, 0x10..=0x12) | (1, 0x01..=0x03) => Err(CoreError::Direction),
-        (0, 0x01..=0x03) | (1, 0x10..=0x12) => Err(CoreError::Length),
+        (1, 0x13) if body.is_empty() => Ok(FeedbackEvent::Disconnected),
+        (0, 0x10..=0x13) | (1, 0x01..=0x03) => Err(CoreError::Direction),
+        (0, 0x01..=0x03) | (1, 0x10..=0x13) => Err(CoreError::Length),
         _ => Err(CoreError::Type),
     }
 }
@@ -424,6 +426,7 @@ pub fn encode_feedback(event: &FeedbackEvent) -> Result<Vec<u8>> {
             (1, 0x11, body)
         }
         FeedbackEvent::Hdr { enabled } => (1, 0x12, vec![u8::from(*enabled)]),
+        FeedbackEvent::Disconnected => (1, 0x13, Vec::new()),
     };
     let mut output = Vec::with_capacity(FEEDBACK_HEADER + body.len());
     output.extend_from_slice(b"VGF1");

core/src/media.rs

@@ -210,10 +210,7 @@ impl Reassembler {
     /// Returns a stable protocol error for conflicting fragments or size-bound violations.
     pub fn push(&mut self, fragment: MediaFragment, now_ms: u64) -> Result<Option<EncodedUnit>> {
         fragment.validate()?;
-        let before = self.incomplete.len();
-        self.incomplete
-            .retain(|unit| now_ms.saturating_sub(unit.started_at_ms) <= EXPIRY_MILLISECONDS);
-        self.expired_units += u64::try_from(before - self.incomplete.len()).unwrap_or(u64::MAX);
+        self.expire(now_ms);
 
         let position = self.incomplete.iter().position(|unit| {
             unit.channel == fragment.channel && unit.sequence == fragment.sequence
@@ -282,6 +279,19 @@ impl Reassembler {
         }))
     }
 
+    pub(crate) fn expire(&mut self, now_ms: u64) {
+        let before = self.incomplete.len();
+        self.incomplete
+            .retain(|unit| now_ms.saturating_sub(unit.started_at_ms) < EXPIRY_MILLISECONDS);
+        self.expired_units += u64::try_from(before - self.incomplete.len()).unwrap_or(u64::MAX);
+    }
+
+    pub(crate) fn next_expiry_ms(&self) -> Option<u64> {
+        self.incomplete
+            .front()
+            .map(|unit| unit.started_at_ms.saturating_add(EXPIRY_MILLISECONDS))
+    }
+
     #[must_use]
     pub fn incomplete_units(&self) -> usize {
         self.incomplete.len()

core/src/transport.rs

@@ -1,16 +1,20 @@
+use std::collections::VecDeque;
 use std::fmt;
 use std::io;
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, Condvar, Mutex, MutexGuard};
 use std::time::{Duration, Instant};
 
-use base64::engine::general_purpose::URL_SAFE_NO_PAD;
+use base64::engine::general_purpose::{STANDARD, URL_SAFE_NO_PAD};
 use base64::Engine as _;
 use quinn::crypto::rustls::QuicClientConfig;
 use quinn::{Endpoint, TransportConfig, VarInt};
+use serde::{Deserialize, Serialize};
 
 use crate::error::{CoreError, Result};
+use crate::input::{decode_feedback, decode_input, encode_feedback, FeedbackEvent};
+use crate::media::{EncodedUnit, MediaFragment, Reassembler};
 use crate::tls::{client_config, SignCallback};
 use crate::wire::{
     decode_stable_error, CapabilityProfile, ClientSessionAuthority, ConnectionManifest,
@@ -18,7 +22,86 @@ use crate::wire::{
 };
 
 const HELLO_LIMIT: usize = 16 * 1024;
+const RELIABLE_FRAME_LIMIT: usize = 128 * 1024;
+const RELIABLE_PAYLOAD_BASE64_LIMIT: usize = 65_536;
 const APPLICATION_ERROR: VarInt = VarInt::from_u32(0x100);
+const CONTROL_FLOW_ID: &str = "control.ack.v1";
+const INPUT_FLOW_ID: &str = "input.sequenced.v1";
+const MEDIA_EVENT_CAPACITY: usize = 4;
+const CONTROL_EVENT_CAPACITY: usize = 64;
+
+#[derive(Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+struct ReliableFrame {
+    version: String,
+    flow_id: String,
+    sequence: u32,
+    flags: u8,
+    fragment_index: u8,
+    fragment_count: u8,
+    timestamp_ms: u64,
+    payload: String,
+}
+
+#[derive(Debug, Eq, PartialEq)]
+struct DecodedReliableFrame {
+    flow_id: String,
+    sequence: u32,
+    timestamp_ms: u64,
+    payload: Vec<u8>,
+}
+
+fn encode_reliable_frame(
+    flow_id: &str,
+    sequence: u32,
+    timestamp_ms: u64,
+    payload: &[u8],
+) -> Result<Vec<u8>> {
+    if !matches!(flow_id, CONTROL_FLOW_ID | INPUT_FLOW_ID) {
+        return Err(CoreError::Protocol);
+    }
+    let payload = STANDARD.encode(payload);
+    if payload.is_empty() || payload.len() > RELIABLE_PAYLOAD_BASE64_LIMIT {
+        return Err(CoreError::Protocol);
+    }
+    let encoded = serde_json::to_vec(&ReliableFrame {
+        version: "1".to_owned(),
+        flow_id: flow_id.to_owned(),
+        sequence,
+        flags: 0,
+        fragment_index: 0,
+        fragment_count: 1,
+        timestamp_ms,
+        payload,
+    })
+    .map_err(|_| CoreError::Protocol)?;
+    frame_length(encoded.len(), RELIABLE_FRAME_LIMIT)?;
+    Ok(encoded)
+}
+
+fn decode_reliable_frame(bytes: &[u8]) -> Result<DecodedReliableFrame> {
+    frame_length(bytes.len(), RELIABLE_FRAME_LIMIT)?;
+    let frame: ReliableFrame = serde_json::from_slice(bytes).map_err(|_| CoreError::Protocol)?;
+    if frame.version != "1"
+        || !matches!(frame.flow_id.as_str(), CONTROL_FLOW_ID | INPUT_FLOW_ID)
+        || frame.flags != 0
+        || frame.fragment_index != 0
+        || frame.fragment_count != 1
+        || frame.payload.is_empty()
+        || frame.payload.len() > RELIABLE_PAYLOAD_BASE64_LIMIT
+    {
+        return Err(CoreError::Protocol);
+    }
+    let payload = STANDARD
+        .decode(frame.payload)
+        .map_err(|_| CoreError::Protocol)?;
+    Ok(DecodedReliableFrame {
+        flow_id: frame.flow_id,
+        sequence: frame.sequence,
+        timestamp_ms: frame.timestamp_ms,
+        payload,
+    })
+}
 
 pub struct AdmissionSigner(Arc<SignCallback>);
 
@@ -54,6 +137,11 @@ impl Cancellation {
         self.0.store(true, Ordering::Release);
     }
 
+    #[must_use]
+    pub(crate) fn is_cancelled(&self) -> bool {
+        self.0.load(Ordering::Acquire)
+    }
+
     fn check(&self) -> Result<()> {
         if self.0.load(Ordering::Acquire) {
             Err(CoreError::Cancelled)
@@ -73,7 +161,203 @@ impl Signers {
 pub struct TransportSession {
     endpoint: Endpoint,
     connection: quinn::Connection,
+    send: quinn::SendStream,
+    receive: quinn::RecvStream,
     authority: ClientSessionAuthority,
+    features: Vec<String>,
+}
+
+pub(crate) struct SessionStartup {
+    prepared: std::sync::mpsc::Sender<()>,
+    commit: tokio::sync::oneshot::Receiver<()>,
+    outcome: std::sync::mpsc::Sender<Result<()>>,
+}
+
+pub(crate) fn session_startup() -> (
+    SessionStartup,
+    std::sync::mpsc::Receiver<()>,
+    tokio::sync::oneshot::Sender<()>,
+    std::sync::mpsc::Receiver<Result<()>>,
+) {
+    let (prepared_tx, prepared_rx) = std::sync::mpsc::channel();
+    let (commit_tx, commit_rx) = tokio::sync::oneshot::channel();
+    let (outcome_tx, outcome_rx) = std::sync::mpsc::channel();
+    (
+        SessionStartup {
+            prepared: prepared_tx,
+            commit: commit_rx,
+            outcome: outcome_tx,
+        },
+        prepared_rx,
+        commit_tx,
+        outcome_rx,
+    )
+}
+
+pub(crate) async fn await_session_startup(
+    result_rx: &mut tokio::sync::mpsc::Receiver<Result<()>>,
+    startup: SessionStartup,
+) -> Result<()> {
+    startup
+        .prepared
+        .send(())
+        .map_err(|_| CoreError::Cancelled)?;
+    let mut loop_result = std::pin::pin!(result_rx.recv());
+    let mut commit = std::pin::pin!(startup.commit);
+    let result = std::future::poll_fn(|context| {
+        if let std::task::Poll::Ready(result) =
+            std::future::Future::poll(loop_result.as_mut(), context)
+        {
+            return std::task::Poll::Ready(result.unwrap_or(Err(CoreError::Transport)));
+        }
+        std::future::Future::poll(commit.as_mut(), context)
+            .map(|result| result.map_err(|_| CoreError::Cancelled))
+    })
+    .await;
+    let _ = startup.outcome.send(result);
+    result
+}
+
+#[derive(Debug, Eq, PartialEq)]
+pub enum SessionCommand {
+    Input(Vec<u8>),
+    Feedback(Vec<u8>),
+}
+
+#[derive(Debug, Eq, PartialEq)]
+pub enum SessionEvent {
+    Media(EncodedUnit),
+    Control { kind: u32, payload: Vec<u8> },
+}
+
+#[derive(Debug, Default)]
+struct SessionEventState {
+    media: VecDeque<EncodedUnit>,
+    control: VecDeque<SessionEvent>,
+    dropped_media_units: u64,
+    closed: bool,
+}
+
+#[derive(Debug, Default)]
+struct SessionEventsInner {
+    state: Mutex<SessionEventState>,
+    ready: Condvar,
+}
+
+#[derive(Clone, Debug)]
+pub struct SessionEventSender(Arc<SessionEventsInner>);
+
+#[derive(Debug)]
+pub struct SessionEventReceiver(Arc<SessionEventsInner>);
+
+#[must_use]
+pub fn bounded_session_events() -> (SessionEventSender, SessionEventReceiver) {
+    let inner = Arc::new(SessionEventsInner::default());
+    (
+        SessionEventSender(Arc::clone(&inner)),
+        SessionEventReceiver(inner),
+    )
+}
+
+fn event_state(inner: &SessionEventsInner) -> MutexGuard<'_, SessionEventState> {
+    inner
+        .state
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
+impl SessionEventSender {
+    fn send_media(&self, unit: EncodedUnit) -> Result<()> {
+        let mut state = event_state(&self.0);
+        if state.closed {
+            return Err(CoreError::Cancelled);
+        }
+        if state.media.len() == MEDIA_EVENT_CAPACITY {
+            if let Some(position) = state
+                .media
+                .iter()
+                .position(|queued| queued.channel == unit.channel)
+            {
+                state.media.remove(position);
+            } else {
+                state.dropped_media_units = state.dropped_media_units.saturating_add(1);
+                return Ok(());
+            }
+            state.dropped_media_units = state.dropped_media_units.saturating_add(1);
+        }
+        state.media.push_back(unit);
+        drop(state);
+        self.0.ready.notify_one();
+        Ok(())
+    }
+
+    fn send_control(&self, kind: u32, payload: Vec<u8>) -> Result<()> {
+        let mut state = event_state(&self.0);
+        if state.closed {
+            return Err(CoreError::Cancelled);
+        }
+        if state.control.len() == CONTROL_EVENT_CAPACITY {
+            return Err(CoreError::QueueFull);
+        }
+        state
+            .control
+            .push_back(SessionEvent::Control { kind, payload });
+        drop(state);
+        self.0.ready.notify_one();
+        Ok(())
+    }
+
+    fn record_media_drops(&self, count: u64) {
+        if count == 0 {
+            return;
+        }
+        let mut state = event_state(&self.0);
+        state.dropped_media_units = state.dropped_media_units.saturating_add(count);
+    }
+
+    pub(crate) fn close(&self) {
+        event_state(&self.0).closed = true;
+        self.0.ready.notify_all();
+    }
+}
+
+impl SessionEventReceiver {
+    pub fn recv_timeout(
+        &self,
+        timeout: Duration,
+    ) -> std::result::Result<SessionEvent, std::sync::mpsc::RecvTimeoutError> {
+        let deadline = Instant::now() + timeout;
+        let mut state = event_state(&self.0);
+        loop {
+            if let Some(event) = state.control.pop_front() {
+                return Ok(event);
+            }
+            if let Some(unit) = state.media.pop_front() {
+                return Ok(SessionEvent::Media(unit));
+            }
+            if state.closed {
+                return Err(std::sync::mpsc::RecvTimeoutError::Disconnected);
+            }
+            let remaining = deadline.saturating_duration_since(Instant::now());
+            if remaining.is_zero() {
+                return Err(std::sync::mpsc::RecvTimeoutError::Timeout);
+            }
+            let (next, wait) = self
+                .0
+                .ready
+                .wait_timeout(state, remaining)
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            state = next;
+            if wait.timed_out() && state.control.is_empty() && state.media.is_empty() {
+                return Err(std::sync::mpsc::RecvTimeoutError::Timeout);
+            }
+        }
+    }
+
+    #[must_use]
+    pub fn dropped_media_units(&self) -> u64 {
+        event_state(&self.0).dropped_media_units
+    }
 }
 
 impl TransportSession {
@@ -87,6 +371,251 @@ impl TransportSession {
         self.connection.closed().await;
         self.endpoint.wait_idle().await;
     }
+
+    /// Runs bounded datagram reassembly and the admitted bidirectional reliable stream.
+    ///
+    /// # Errors
+    ///
+    /// Returns a provider-free protocol, transport, queue, or cancellation error.
+    pub async fn run(
+        self,
+        commands: tokio::sync::mpsc::Receiver<SessionCommand>,
+        events: SessionEventSender,
+        cancellation: Cancellation,
+    ) -> Result<()> {
+        self.run_with_startup(commands, events, cancellation, None)
+            .await
+    }
+
+    pub(crate) async fn run_with_startup(
+        self,
+        commands: tokio::sync::mpsc::Receiver<SessionCommand>,
+        events: SessionEventSender,
+        cancellation: Cancellation,
+        startup: Option<SessionStartup>,
+    ) -> Result<()> {
+        let Self {
+            endpoint,
+            connection,
+            send,
+            receive,
+            authority: _,
+            features,
+        } = self;
+        let send = Arc::new(tokio::sync::Mutex::new(send));
+        let sequence = Arc::new(std::sync::atomic::AtomicU32::new(0));
+        let (result_tx, mut result_rx) = tokio::sync::mpsc::channel(3);
+        let writer = tokio::spawn(command_loop(
+            commands,
+            features,
+            Arc::clone(&send),
+            Arc::clone(&sequence),
+            result_tx.clone(),
+        ));
+        let reliable = tokio::spawn(reliable_loop(
+            receive,
+            events.clone(),
+            Arc::clone(&send),
+            Arc::clone(&sequence),
+            result_tx.clone(),
+        ));
+        let datagrams = tokio::spawn(datagram_loop(connection.clone(), events.clone(), result_tx));
+        if let Some(startup) = startup {
+            if let Err(error) = await_session_startup(&mut result_rx, startup).await {
+                connection.close(APPLICATION_ERROR, b"session stopped");
+                writer.abort();
+                reliable.abort();
+                datagrams.abort();
+                endpoint.wait_idle().await;
+                events.close();
+                return Err(error);
+            }
+        }
+        loop {
+            if cancellation.check().is_err() {
+                connection.close(APPLICATION_ERROR, b"cancelled");
+                writer.abort();
+                reliable.abort();
+                datagrams.abort();
+                endpoint.wait_idle().await;
+                events.close();
+                return Err(CoreError::Cancelled);
+            }
+            if let Ok(Some(result)) =
+                tokio::time::timeout(Duration::from_millis(2), result_rx.recv()).await
+            {
+                connection.close(APPLICATION_ERROR, b"session stopped");
+                writer.abort();
+                reliable.abort();
+                datagrams.abort();
+                endpoint.wait_idle().await;
+                events.close();
+                return result;
+            }
+        }
+    }
+}
+
+async fn command_loop(
+    mut commands: tokio::sync::mpsc::Receiver<SessionCommand>,
+    features: Vec<String>,
+    send: Arc<tokio::sync::Mutex<quinn::SendStream>>,
+    sequence: Arc<std::sync::atomic::AtomicU32>,
+    result: tokio::sync::mpsc::Sender<Result<()>>,
+) {
+    let feature_refs = features.iter().map(String::as_str).collect::<Vec<_>>();
+    while let Some(command) = commands.recv().await {
+        let outcome = async {
+            let (flow_id, payload) = match command {
+                SessionCommand::Input(payload) => {
+                    decode_input(&payload, &feature_refs)?;
+                    (INPUT_FLOW_ID, payload)
+                }
+                SessionCommand::Feedback(payload) => {
+                    match decode_feedback(&payload)? {
+                        FeedbackEvent::IdrRequest
+                        | FeedbackEvent::Fec(_)
+                        | FeedbackEvent::TerminalReceipt => {}
+                        _ => return Err(CoreError::Direction),
+                    }
+                    (CONTROL_FLOW_ID, payload)
+                }
+            };
+            let encoded = encode_reliable_frame(
+                flow_id,
+                next_sequence(&sequence)?,
+                unix_milliseconds()?,
+                &payload,
+            )?;
+            let mut stream = send.lock().await;
+            write_frame_with_limit(&mut stream, &encoded, RELIABLE_FRAME_LIMIT).await
+        }
+        .await;
+        if let Err(error) = outcome {
+            let _ = result.send(Err(error)).await;
+            return;
+        }
+    }
+    let _ = result.send(Err(CoreError::Cancelled)).await;
+}
+
+async fn reliable_loop(
+    mut receive: quinn::RecvStream,
+    events: SessionEventSender,
+    send: Arc<tokio::sync::Mutex<quinn::SendStream>>,
+    sequence: Arc<std::sync::atomic::AtomicU32>,
+    result: tokio::sync::mpsc::Sender<Result<()>>,
+) {
+    let mut inbound_sequence = None;
+    loop {
+        let outcome = async {
+            let frame = decode_reliable_frame(
+                &read_frame_with_limit(&mut receive, RELIABLE_FRAME_LIMIT).await?,
+            )?;
+            if frame.flow_id != CONTROL_FLOW_ID
+                || inbound_sequence.is_some_and(|previous| frame.sequence <= previous)
+            {
+                return Err(CoreError::Protocol);
+            }
+            inbound_sequence = Some(frame.sequence);
+            let feedback = decode_feedback(&frame.payload)?;
+            let kind = u32::from(frame.payload[5]);
+            let payload = frame.payload[8..].to_vec();
+            events.send_control(kind, payload)?;
+            if matches!(
+                feedback,
+                FeedbackEvent::Termination { .. } | FeedbackEvent::Disconnected
+            ) {
+                let receipt = encode_feedback(&FeedbackEvent::TerminalReceipt)?;
+                let encoded = encode_reliable_frame(
+                    CONTROL_FLOW_ID,
+                    next_sequence(&sequence)?,
+                    unix_milliseconds()?,
+                    &receipt,
+                )?;
+                let mut stream = send.lock().await;
+                write_frame_with_limit(&mut stream, &encoded, RELIABLE_FRAME_LIMIT).await?;
+            }
+            Ok(())
+        }
+        .await;
+        if let Err(error) = outcome {
+            let _ = result.send(Err(error)).await;
+            return;
+        }
+    }
+}
+
+async fn datagram_loop(
+    connection: quinn::Connection,
+    events: SessionEventSender,
+    result: tokio::sync::mpsc::Sender<Result<()>>,
+) {
+    let started = Instant::now();
+    let mut reassembler = Reassembler::new();
+    let mut accounted_reassembly_drops = 0;
+    loop {
+        let now_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
+        reassembler.expire(now_ms);
+        account_reassembly_drops(&reassembler, &events, &mut accounted_reassembly_drops);
+        let wait = reassembler
+            .next_expiry_ms()
+            .map_or(Duration::from_secs(86_400), |deadline| {
+                Duration::from_millis(deadline.saturating_sub(now_ms))
+            });
+        let datagram = match tokio::time::timeout(wait, connection.read_datagram()).await {
+            Err(_) => continue,
+            Ok(Ok(datagram)) => datagram,
+            Ok(Err(_)) => {
+                let _ = result.send(Err(CoreError::Transport)).await;
+                return;
+            }
+        };
+        let outcome = MediaFragment::decode(&datagram).and_then(|fragment| {
+            let now_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
+            reassembler.push(fragment, now_ms)
+        });
+        match outcome {
+            Ok(Some(unit)) => match events.send_media(unit) {
+                Ok(()) => {}
+                Err(CoreError::Cancelled) => {
+                    let _ = result.send(Err(CoreError::Cancelled)).await;
+                    return;
+                }
+                Err(error) => {
+                    let _ = result.send(Err(error)).await;
+                    return;
+                }
+            },
+            Ok(None) => {}
+            Err(error) => {
+                let _ = result.send(Err(error)).await;
+                return;
+            }
+        }
+        account_reassembly_drops(&reassembler, &events, &mut accounted_reassembly_drops);
+    }
+}
+
+fn account_reassembly_drops(
+    reassembler: &Reassembler,
+    events: &SessionEventSender,
+    accounted: &mut u64,
+) {
+    let current = reassembler
+        .evicted_units()
+        .saturating_add(reassembler.expired_units());
+    events.record_media_drops(current.saturating_sub(*accounted));
+    *accounted = current;
+}
+
+fn next_sequence(sequence: &std::sync::atomic::AtomicU32) -> Result<u32> {
+    sequence
+        .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
+            current.checked_add(1)
+        })
+        .map(|previous| previous + 1)
+        .map_err(|_| CoreError::Protocol)
 }
 
 impl fmt::Debug for TransportSession {
@@ -146,12 +675,18 @@ pub async fn connect_with_cancellation(
     let admission_signer = signers.admission.0;
     let tls_signer = signers.tls.0;
     let tls_cancellation = cancellation.clone();
+    let tls_error = Arc::new(Mutex::new(None));
+    let callback_tls_error = Arc::clone(&tls_error);
     let checked_tls_signer: Arc<SignCallback> = Arc::new(move |message| {
         tls_cancellation.check()?;
         if started.elapsed() >= deadline {
             return Err(CoreError::Cancelled);
         }
-        let signature = tls_signer(message)?;
+        let signature = tls_signer(message).inspect_err(|error| {
+            *callback_tls_error
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(*error);
+        })?;
         tls_cancellation.check()?;
         if started.elapsed() >= deadline {
             return Err(CoreError::Cancelled);
@@ -188,7 +723,7 @@ pub async fn connect_with_cancellation(
                 let remaining_remotes = remote_count.saturating_sub(remote_index).max(1);
                 let attempt_budget = address_deadline.saturating_duration_since(Instant::now())
                     / u32::try_from(remaining_remotes).unwrap_or(1);
-                match cancellable_timeout(
+                let attempt = cancellable_timeout(
                     attempt_budget,
                     cancellation,
                     dial(
@@ -206,8 +741,14 @@ pub async fn connect_with_cancellation(
                         },
                     ),
                 )
-                .await
+                .await;
+                if let Some(error) = *tls_error
+                    .lock()
+                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                 {
+                    return Err(error);
+                }
+                match attempt {
                     Ok(Ok(session)) => return Ok(session),
                     Ok(Err(AttemptError::Retry)) => {}
                     Err(CoreError::Transport)
@@ -333,8 +874,6 @@ async fn dial(
     write_frame(&mut send, &payload)
         .await
         .map_err(|_| AttemptError::Terminal(connection_error(&connection)))?;
-    send.finish()
-        .map_err(|_| AttemptError::Terminal(connection_error(&connection)))?;
     let response = read_frame(&mut receive)
         .await
         .map_err(|_| AttemptError::Terminal(connection_error(&connection)))?;
@@ -352,7 +891,10 @@ async fn dial(
     Ok(TransportSession {
         endpoint,
         connection,
+        send,
+        receive,
         authority,
+        features: context.manifest.features().to_vec(),
     })
 }
 
@@ -418,7 +960,15 @@ impl From<CoreError> for AttemptError {
 }
 
 async fn write_frame(stream: &mut quinn::SendStream, payload: &[u8]) -> Result<()> {
-    let length = hello_length(payload.len())?;
+    write_frame_with_limit(stream, payload, HELLO_LIMIT).await
+}
+
+async fn write_frame_with_limit(
+    stream: &mut quinn::SendStream,
+    payload: &[u8],
+    maximum: usize,
+) -> Result<()> {
+    let length = frame_length(payload.len(), maximum)?;
     stream
         .write_all(&length.to_be_bytes())
         .await
@@ -430,13 +980,18 @@ async fn write_frame(stream: &mut quinn::SendStream, payload: &[u8]) -> Result<(
 }
 
 async fn read_frame(stream: &mut quinn::RecvStream) -> Result<Vec<u8>> {
+    read_frame_with_limit(stream, HELLO_LIMIT).await
+}
+
+async fn read_frame_with_limit(stream: &mut quinn::RecvStream, maximum: usize) -> Result<Vec<u8>> {
     let mut header = [0_u8; 4];
     stream
         .read_exact(&mut header)
         .await
         .map_err(|_| CoreError::Transport)?;
-    let length = usize::try_from(hello_length(
+    let length = usize::try_from(frame_length(
         usize::try_from(u32::from_be_bytes(header)).map_err(|_| CoreError::Protocol)?,
+        maximum,
     )?)
     .map_err(|_| CoreError::Protocol)?;
     let mut payload = vec![0; length];
@@ -447,8 +1002,19 @@ async fn read_frame(stream: &mut quinn::RecvStream) -> Result<Vec<u8>> {
     Ok(payload)
 }
 
+fn unix_milliseconds() -> Result<u64> {
+    let duration = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map_err(|_| CoreError::Transport)?;
+    u64::try_from(duration.as_millis()).map_err(|_| CoreError::Transport)
+}
+
 fn hello_length(length: usize) -> Result<u32> {
-    if length == 0 || length > HELLO_LIMIT {
+    frame_length(length, HELLO_LIMIT)
+}
+
+fn frame_length(length: usize, maximum: usize) -> Result<u32> {
+    if length == 0 || length > maximum {
         return Err(CoreError::Protocol);
     }
     u32::try_from(length).map_err(|_| CoreError::Protocol)
@@ -460,8 +1026,13 @@ mod tests {
     use std::net::SocketAddr;
     use std::time::{Duration, Instant};
 
-    use super::{bounded_lookup, hello_length, Cancellation, HELLO_LIMIT};
+    use super::{
+        await_session_startup, bounded_lookup, bounded_session_events, decode_reliable_frame,
+        encode_reliable_frame, hello_length, session_startup, Cancellation, SessionEvent,
+        HELLO_LIMIT,
+    };
     use crate::error::CoreError;
+    use crate::media::{EncodedUnit, MediaChannel};
 
     #[test]
     fn hello_length_bound_accepts_exact_limit_and_rejects_larger() {
@@ -497,4 +1068,147 @@ mod tests {
             ));
         assert_eq!(result.err(), Some(CoreError::Cancelled));
     }
+
+    #[test]
+    fn reliable_channel_frame_is_strict_bounded_and_round_trips() {
+        let encoded =
+            encode_reliable_frame("input.sequenced.v1", 7, 42, b"VGI1\x01\x04\x01\x00\x00\x1e")
+                .expect("bounded frame");
+        let frame = decode_reliable_frame(&encoded).expect("strict frame");
+        assert_eq!(frame.flow_id, "input.sequenced.v1");
+        assert_eq!(frame.sequence, 7);
+        assert_eq!(frame.timestamp_ms, 42);
+        assert_eq!(frame.payload, b"VGI1\x01\x04\x01\x00\x00\x1e");
+
+        let mut unknown: serde_json::Value = serde_json::from_slice(&encoded).expect("frame JSON");
+        unknown["provider_identity"] = serde_json::json!("forbidden");
+        assert_eq!(
+            decode_reliable_frame(&serde_json::to_vec(&unknown).expect("JSON")),
+            Err(CoreError::Protocol)
+        );
+        assert_eq!(
+            encode_reliable_frame("control.ack.v1", 0, 0, &vec![0; 65_537]),
+            Err(CoreError::Protocol)
+        );
+    }
+
+    #[test]
+    fn session_event_queue_evicts_oldest_same_channel_at_four_and_counts_the_drop() {
+        let (events, receiver) = bounded_session_events();
+        for sequence in 1..=4 {
+            events
+                .send_media(EncodedUnit {
+                    channel: if sequence == 2 {
+                        MediaChannel::Audio
+                    } else {
+                        MediaChannel::Video
+                    },
+                    sequence,
+                    timestamp_ms: u64::from(sequence),
+                    payload: vec![u8::try_from(sequence).expect("small sequence")],
+                })
+                .expect("bounded media");
+        }
+        events
+            .send_media(EncodedUnit {
+                channel: MediaChannel::Video,
+                sequence: 5,
+                timestamp_ms: 5,
+                payload: vec![5],
+            })
+            .expect("same-channel eviction");
+
+        let sequences = (0..4)
+            .map(|_| match receiver.recv_timeout(Duration::from_millis(10)) {
+                Ok(SessionEvent::Media(unit)) => unit.sequence,
+                other => panic!("unexpected event: {other:?}"),
+            })
+            .collect::<Vec<_>>();
+        assert_eq!(sequences, vec![2, 3, 4, 5]);
+        assert_eq!(receiver.dropped_media_units(), 1);
+    }
+
+    #[test]
+    fn session_event_queue_bounds_reliable_control_at_64_without_blocking() {
+        let (events, _receiver) = bounded_session_events();
+        for sequence in 0..64 {
+            events
+                .send_control(sequence, vec![0])
+                .expect("bounded control backlog");
+        }
+        let started = Instant::now();
+        assert_eq!(events.send_control(65, vec![0]), Err(CoreError::QueueFull));
+        assert!(started.elapsed() < Duration::from_millis(50));
+    }
+
+    #[test]
+    fn owning_input_queue_has_exact_capacity_64_with_consumer_held() {
+        let (sender, _receiver) = tokio::sync::mpsc::channel(64);
+        for sequence in 0..64 {
+            sender
+                .try_send(super::SessionCommand::Input(vec![
+                    u8::try_from(sequence).expect("small sequence")
+                ]))
+                .expect("exact input capacity");
+        }
+        assert!(matches!(
+            sender.try_send(super::SessionCommand::Input(vec![64])),
+            Err(tokio::sync::mpsc::error::TrySendError::Full(_))
+        ));
+    }
+
+    #[test]
+    fn queued_loop_failure_wins_before_session_ready() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime");
+        let (result_tx, mut result_rx) = tokio::sync::mpsc::channel(1);
+        result_tx
+            .try_send(Err(crate::error::CoreError::Protocol))
+            .expect("queue terminal loop result");
+        let (startup, prepared, commit, outcome) = session_startup();
+        let running =
+            runtime.spawn(async move { await_session_startup(&mut result_rx, startup).await });
+        runtime.block_on(async { tokio::task::yield_now().await });
+        prepared.recv().expect("session prepared");
+        let _ = commit.send(());
+
+        assert_eq!(
+            outcome.recv().expect("startup outcome"),
+            Err(crate::error::CoreError::Protocol)
+        );
+        assert_eq!(
+            runtime.block_on(running).expect("startup task"),
+            Err(crate::error::CoreError::Protocol)
+        );
+    }
+
+    #[test]
+    fn loop_failure_after_session_ready_remains_for_publication_race() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime");
+        let (result_tx, mut result_rx) = tokio::sync::mpsc::channel(1);
+        let (startup, prepared, commit, outcome) = session_startup();
+        let running = runtime.spawn(async move {
+            let result = await_session_startup(&mut result_rx, startup).await;
+            (result, result_rx)
+        });
+        runtime.block_on(async { tokio::task::yield_now().await });
+        prepared.recv().expect("session prepared");
+        commit.send(()).expect("commit startup");
+
+        let (startup_result, mut result_rx) = runtime.block_on(running).expect("startup task");
+        assert_eq!(startup_result, Ok(()));
+        assert_eq!(outcome.recv().expect("startup outcome"), Ok(()));
+        result_tx
+            .try_send(Err(CoreError::Protocol))
+            .expect("queue post-ready terminal result");
+        assert_eq!(
+            runtime.block_on(result_rx.recv()),
+            Some(Err(CoreError::Protocol))
+        );
+    }
 }

core/src/wire.rs

@@ -1,5 +1,6 @@
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
+use std::time::{SystemTime, UNIX_EPOCH};
 
 use crate::error::{CoreError, Result};
 
@@ -126,6 +127,47 @@ struct Timestamp {
     nanosecond: u32,
 }
 
+pub(crate) fn now_utc() -> Result<String> {
+    system_time_utc(SystemTime::now())
+}
+
+fn system_time_utc(now: SystemTime) -> Result<String> {
+    let seconds = now
+        .duration_since(UNIX_EPOCH)
+        .map_err(|_| CoreError::InvalidArgument)?
+        .as_secs();
+    let days = seconds / 86_400;
+    let day_seconds = seconds % 86_400;
+    let shifted = days
+        .checked_add(719_468)
+        .ok_or(CoreError::InvalidArgument)?;
+    let era = shifted / 146_097;
+    let day_of_era = shifted % 146_097;
+    let year_of_era =
+        (day_of_era - day_of_era / 1_460 + day_of_era / 36_524 - day_of_era / 146_096) / 365;
+    let mut year = year_of_era + era * 400;
+    let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100);
+    let month_prime = (5 * day_of_year + 2) / 153;
+    let day = day_of_year - (153 * month_prime + 2) / 5 + 1;
+    let month = if month_prime < 10 {
+        month_prime + 3
+    } else {
+        month_prime - 9
+    };
+    if month <= 2 {
+        year += 1;
+    }
+    if year > 9_999 {
+        return Err(CoreError::InvalidArgument);
+    }
+    let hour = day_seconds / 3_600;
+    let minute = (day_seconds % 3_600) / 60;
+    let second = day_seconds % 60;
+    Ok(format!(
+        "{year:04}-{month:02}-{day:02}T{hour:02}:{minute:02}:{second:02}Z"
+    ))
+}
+
 fn timestamp(value: &str, exact_seconds: bool) -> Option<Timestamp> {
     let bytes = value.as_bytes();
     if bytes.len() < 20
@@ -452,6 +494,10 @@ impl ConnectionManifest {
         &self.gateway.public_identity
     }
 
+    pub(crate) fn features(&self) -> &[String] {
+        &self.tunnel.features
+    }
+
     pub(crate) fn admission(
         &self,
         client_nonce: String,
@@ -712,8 +758,21 @@ pub(crate) fn decode_stable_error(bytes: &[u8]) -> Result<DecodedStableError> {
 
 #[cfg(test)]
 mod stable_error_tests {
-    use super::decode_stable_error;
+    use super::{decode_stable_error, system_time_utc};
     use crate::error::CoreError;
+    use std::time::{Duration, UNIX_EPOCH};
+
+    #[test]
+    fn system_clock_conversion_is_exact_at_epoch_and_leap_day() {
+        assert_eq!(
+            system_time_utc(UNIX_EPOCH).as_deref(),
+            Ok("1970-01-01T00:00:00Z")
+        );
+        assert_eq!(
+            system_time_utc(UNIX_EPOCH + Duration::from_secs(1_709_251_199)).as_deref(),
+            Ok("2024-02-29T23:59:59Z")
+        );
+    }
 
     #[test]
     fn stable_error_uses_exact_rc5_bounds_and_preserves_retryability() {

core/tests/abi_contract.rs

@@ -2,17 +2,31 @@
     unsafe_code,
     clippy::borrow_as_ptr,
     clippy::cast_possible_truncation,
-    clippy::items_after_statements
+    clippy::items_after_statements,
+    clashing_extern_declarations
 )]
 
+use std::cell::RefCell;
+use std::collections::HashMap;
 use std::ffi::c_void;
 use std::mem::{offset_of, size_of};
 use std::ptr;
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
-use std::sync::{Condvar, Mutex};
+use std::sync::{Arc, Condvar, Mutex};
 use std::thread;
 use std::time::{Duration, Instant};
 
+use rustls::sign::SigningKey;
+use rustls::SignatureScheme;
+
+#[path = "gateway_oracle.rs"]
+mod gateway_oracle;
+
+thread_local! {
+    static CORE_CONTEXTS: RefCell<HashMap<usize, usize>> = RefCell::new(HashMap::new());
+    static ORACLES: RefCell<HashMap<usize, gateway_oracle::Oracle>> = RefCell::new(HashMap::new());
+}
+
 const ABI_V1: u32 = 1;
 const OK: u32 = 0;
 const INVALID_ARGUMENT: u32 = 1;
@@ -167,6 +181,8 @@ struct Context {
     cancel_on_connecting: AtomicBool,
     reentry_target: AtomicUsize,
     reentry_results: Mutex<Vec<u32>>,
+    admission_key: Mutex<Option<Arc<dyn SigningKey>>>,
+    tls_key: Mutex<Option<Arc<dyn SigningKey>>>,
 }
 
 impl Default for Context {
@@ -191,6 +207,8 @@ impl Default for Context {
             cancel_on_connecting: AtomicBool::new(false),
             reentry_target: AtomicUsize::new(0),
             reentry_results: Mutex::new(Vec::new()),
+            admission_key: Mutex::new(None),
+            tls_key: Mutex::new(None),
         }
     }
 }
@@ -206,9 +224,22 @@ unsafe extern "C" fn sign_admission(raw: *mut c_void, input: BytesView, output:
     let ctx = unsafe { context(raw) };
     ctx.admission_calls.fetch_add(1, Ordering::SeqCst);
     *ctx.admission_input.lock().expect("admission lock") = bytes.to_vec();
+    let status = ctx.admission_status.load(Ordering::SeqCst);
+    if status != OK {
+        return status;
+    }
+    let key = ctx.admission_key.lock().expect("admission key");
+    let Some(key) = key.as_ref() else {
+        return INTERNAL;
+    };
+    let signer = key
+        .choose_scheme(&[SignatureScheme::ED25519])
+        .expect("Ed25519 admission signer");
+    let signature = signer.sign(bytes).expect("admission signature");
+    assert_eq!(signature.len(), 64);
     // Test invariant: ABI promises a writable 64-byte Rust-owned signature buffer.
-    unsafe { ptr::write_bytes(output, 0xA5, 64) };
-    ctx.admission_status.load(Ordering::SeqCst)
+    unsafe { ptr::copy_nonoverlapping(signature.as_ptr(), output, signature.len()) };
+    OK
 }
 
 unsafe extern "C" fn sign_tls(raw: *mut c_void, input: BytesView, output: *mut u8) -> u32 {
@@ -217,9 +248,22 @@ unsafe extern "C" fn sign_tls(raw: *mut c_void, input: BytesView, output: *mut u
     let ctx = unsafe { context(raw) };
     ctx.tls_calls.fetch_add(1, Ordering::SeqCst);
     *ctx.tls_input.lock().expect("tls lock") = bytes.to_vec();
+    let status = ctx.tls_status.load(Ordering::SeqCst);
+    if status != OK {
+        return status;
+    }
+    let key = ctx.tls_key.lock().expect("TLS key");
+    let Some(key) = key.as_ref() else {
+        return INTERNAL;
+    };
+    let signer = key
+        .choose_scheme(&[SignatureScheme::ED25519])
+        .expect("Ed25519 TLS signer");
+    let signature = signer.sign(bytes).expect("TLS signature");
+    assert_eq!(signature.len(), 64);
     // Test invariant: ABI promises a writable 64-byte Rust-owned signature buffer.
-    unsafe { ptr::write_bytes(output, 0x5A, 64) };
-    ctx.tls_status.load(Ordering::SeqCst)
+    unsafe { ptr::copy_nonoverlapping(signature.as_ptr(), output, signature.len()) };
+    OK
 }
 
 unsafe extern "C" fn sign_admission_probes_global_reentry(
@@ -391,14 +435,66 @@ fn create(ctx: &mut Context) -> *mut Core {
     // Test invariant: config/out pointers remain valid for the synchronous create call.
     assert_eq!(unsafe { verse_core_create_v1(&config, &mut core) }, OK);
     assert!(!core.is_null());
-    ctx.core.store(core as usize, Ordering::SeqCst);
+    register(core, ctx);
     core
 }
 
+fn register(core: *mut Core, ctx: &mut Context) {
+    ctx.core.store(core as usize, Ordering::SeqCst);
+    CORE_CONTEXTS.with(|contexts| {
+        contexts
+            .borrow_mut()
+            .insert(core as usize, ptr::from_mut(ctx) as usize);
+    });
+}
+
 fn connect(core: *mut Core, manifest: &[u8], credential: &[u8]) -> u32 {
-    let request = request(manifest, credential);
-    // Test invariant: request and backing byte slices remain valid for the synchronous call.
-    unsafe { verse_core_connect_v1(core, &request) }
+    connect_mode(core, "", manifest, credential)
+}
+
+fn connect_mode(core: *mut Core, mode: &str, manifest: &[u8], credential: &[u8]) -> u32 {
+    connect_with_oracle(core, mode, |oracle| {
+        let manifest = if manifest == MANIFEST {
+            oracle.ready.manifest.as_bytes()
+        } else {
+            manifest
+        };
+        let credential = if credential == CREDENTIAL {
+            oracle.ready.credential.as_bytes()
+        } else {
+            credential
+        };
+        let request = request(manifest, credential);
+        // Test invariant: request and backing byte slices remain valid for the synchronous call.
+        unsafe { verse_core_connect_v1(core, &request) }
+    })
+}
+
+fn connect_with_oracle(
+    core: *mut Core,
+    mode: &str,
+    action: impl FnOnce(&gateway_oracle::Oracle) -> u32,
+) -> u32 {
+    let oracle = gateway_oracle::Oracle::start(mode);
+    CORE_CONTEXTS.with(|contexts| {
+        let raw = *contexts
+            .borrow()
+            .get(&(core as usize))
+            .expect("registered ABI context");
+        // Test invariant: context outlives the core and is removed only after destroy succeeds.
+        let ctx = unsafe { &*(raw as *const Context) };
+        *ctx.admission_key.lock().expect("admission key") =
+            Some(gateway_oracle::test_key(&oracle.ready.admission_key));
+        *ctx.tls_key.lock().expect("TLS key") =
+            Some(gateway_oracle::test_key(&oracle.ready.client_key));
+    });
+    let status = action(&oracle);
+    if status == OK {
+        ORACLES.with(|oracles| {
+            oracles.borrow_mut().insert(core as usize, oracle);
+        });
+    }
+    status
 }
 
 fn wait_for(ctx: &Context, predicate: impl Fn(&[u32]) -> bool) {
@@ -419,7 +515,16 @@ fn wait_for(ctx: &Context, predicate: impl Fn(&[u32]) -> bool) {
 
 fn destroy(core: *mut Core) -> u32 {
     // Test invariant: caller retains the handle until destroy reports success.
-    unsafe { verse_core_destroy_v1(core, 2_000) }
+    let status = unsafe { verse_core_destroy_v1(core, 2_000) };
+    if status == OK {
+        ORACLES.with(|oracles| {
+            oracles.borrow_mut().remove(&(core as usize));
+        });
+        CORE_CONTEXTS.with(|contexts| {
+            contexts.borrow_mut().remove(&(core as usize));
+        });
+    }
+    status
 }
 
 #[test]
@@ -500,8 +605,11 @@ fn connect_copies_inputs_and_calls_purpose_specific_signers_once() {
 
     assert_eq!(ctx.admission_calls.load(Ordering::SeqCst), 1);
     assert_eq!(ctx.tls_calls.load(Ordering::SeqCst), 1);
-    assert_eq!(&*ctx.admission_input.lock().expect("admission"), MANIFEST);
-    assert_eq!(&*ctx.tls_input.lock().expect("tls"), CREDENTIAL);
+    let admission = ctx.admission_input.lock().expect("admission");
+    let tls = ctx.tls_input.lock().expect("tls");
+    assert!(admission.starts_with(b"versevdi/tunnel-admission/v1"));
+    assert!(!tls.is_empty());
+    assert_ne!(&*admission, &*tls);
     assert_eq!(destroy(core), OK);
 }
 
@@ -564,13 +672,18 @@ fn tables_and_slices_reject_short_unsupported_null_and_oversized_inputs() {
         base: ConnectRequest,
         ignored: [u8; 24],
     }
-    let mut trailing_request = ExtendedRequest {
-        base: request(MANIFEST, CREDENTIAL),
-        ignored: [0xEE; 24],
-    };
-    trailing_request.base.struct_size = size_of::<ExtendedRequest>() as u32;
     assert_eq!(
-        unsafe { verse_core_connect_v1(core, &trailing_request.base) },
+        connect_with_oracle(core, "", |oracle| {
+            let mut trailing_request = ExtendedRequest {
+                base: request(
+                    oracle.ready.manifest.as_bytes(),
+                    oracle.ready.credential.as_bytes(),
+                ),
+                ignored: [0xEE; 24],
+            };
+            trailing_request.base.struct_size = size_of::<ExtendedRequest>() as u32;
+            unsafe { verse_core_connect_v1(core, &trailing_request.base) }
+        }),
         OK
     );
 
@@ -616,7 +729,7 @@ fn signer_callbacks_cannot_reenter_even_cancel() {
     cfg.sign_admission = Some(sign_admission_reenters);
     let mut core = ptr::null_mut();
     assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK);
-    ctx.core.store(core as usize, Ordering::SeqCst);
+    register(core, &mut ctx);
     assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK);
     assert_eq!(ctx.reentry_cancel.load(Ordering::SeqCst), REENTRANT);
     assert_eq!(destroy(core), OK);
@@ -629,12 +742,13 @@ fn signer_callback_rejects_every_stateful_api_across_handles() {
     let other = create(&mut other_ctx);
 
     let mut ctx = Context::default();
+    ctx.reentry_cancel.store(OK, Ordering::SeqCst);
     ctx.reentry_target.store(other as usize, Ordering::SeqCst);
     let mut cfg = config(&mut ctx);
     cfg.sign_admission = Some(sign_admission_probes_global_reentry);
     let mut core = ptr::null_mut();
     assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK);
-    ctx.core.store(core as usize, Ordering::SeqCst);
+    register(core, &mut ctx);
     assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK);
     assert_eq!(
         ctx.reentry_results
@@ -644,8 +758,8 @@ fn signer_callback_rejects_every_stateful_api_across_handles() {
         [ABI_V1, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT]
     );
 
-    assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK);
     assert_eq!(destroy(core), OK);
+    assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK);
     assert_eq!(destroy(other), OK);
 }
 
@@ -661,7 +775,7 @@ fn event_callback_allows_only_originating_handle_cancel() {
     cfg.on_state = Some(on_state_probes_global_reentry);
     let mut core = ptr::null_mut();
     assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK);
-    ctx.core.store(core as usize, Ordering::SeqCst);
+    register(core, &mut ctx);
     assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK);
     wait_for(&ctx, |states| states.contains(&STATE_CANCELLED));
     assert_eq!(
@@ -672,8 +786,8 @@ fn event_callback_allows_only_originating_handle_cancel() {
         [ABI_V1, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT, OK]
     );
 
-    assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK);
     assert_eq!(destroy(core), OK);
+    assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK);
     assert_eq!(destroy(other), OK);
 }
 
@@ -720,7 +834,7 @@ fn cancel_during_connect_preserves_state_order_and_stops_before_tls_signing() {
     cfg.sign_admission = Some(sign_admission_waits_for_cancel);
     let mut core = ptr::null_mut();
     assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK);
-    ctx.core.store(core as usize, Ordering::SeqCst);
+    register(core, &mut ctx);
 
     assert_eq!(connect(core, MANIFEST, CREDENTIAL), CANCELLED);
     wait_for(&ctx, |states| states.contains(&STATE_CANCELLED));
@@ -737,16 +851,22 @@ fn send_input_is_nonblocking_bounded_and_cancel_is_idempotent() {
     let mut ctx = Context::default();
     ctx.reentry_cancel.store(OK, Ordering::SeqCst);
     let core = create(&mut ctx);
-    assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK);
+    assert_eq!(connect_mode(core, "slow-input", MANIFEST, CREDENTIAL), OK);
     let event = keyboard_event();
-    for _ in 0..64 {
-        assert_eq!(unsafe { verse_core_send_input_v1(core, &event) }, OK);
-    }
-    assert_eq!(
-        unsafe { verse_core_send_input_v1(core, &event) },
-        QUEUE_FULL
-    );
-    assert_eq!(unsafe { verse_core_request_idr_v1(core) }, OK);
+    let started = Instant::now();
+    let mut sent = 0;
+    let saturated = loop {
+        match unsafe { verse_core_send_input_v1(core, &event) } {
+            OK => sent += 1,
+            QUEUE_FULL => break true,
+            status => panic!("unexpected input status {status}"),
+        }
+        if sent == 10_000 || started.elapsed() > Duration::from_secs(1) {
+            break false;
+        }
+    };
+    assert!(saturated, "real slow consumer did not expose bounded queue");
+    assert!(started.elapsed() < Duration::from_secs(1));
     assert_eq!(unsafe { verse_core_cancel_v1(core) }, OK);
     assert_eq!(unsafe { verse_core_cancel_v1(core) }, OK);
     assert_eq!(unsafe { verse_core_send_input_v1(core, &event) }, CANCELLED);

core/tests/ffi/abi_smoke.c

@@ -8,7 +8,7 @@
 static const char MANIFEST[] =
     "{\"version\":\"1\",\"purpose\":\"launch\",\"session_id\":\"session\","
     "\"reconnect_sequence\":0,\"gateway\":{\"id\":\"gateway\",\"addresses\":["
-    "\"gateway.test:443\"],\"public_identity\":\"gateway.test\"},\"tunnel\":{"
+    "\"127.0.0.1:9\"],\"public_identity\":\"gateway.test\"},\"tunnel\":{"
     "\"versions\":[\"verse-gateway-v1/1\"],\"features\":[\"control.v1\"]},"
     "\"profile\":{\"id\":\"standard\",\"bounds\":{\"minimum_kbps\":1000,"
     "\"target_kbps\":5000,\"maximum_kbps\":10000},\"display_mode\":null},"
@@ -24,36 +24,28 @@ static const char CREDENTIAL[] =
 typedef struct smoke_context {
     atomic_uint admission_calls;
     atomic_uint tls_calls;
-    atomic_uint state_calls;
 } smoke_context_t;
 
-static verse_status_t sign_admission(
+static verse_status_t reject_admission(
     void *raw,
     verse_bytes_view_t input,
     uint8_t signature[64]) {
     smoke_context_t *context = raw;
-    assert(input.length == sizeof(MANIFEST) - 1U);
-    memset(signature, 0xA5, 64U);
+    (void)input;
+    (void)signature;
     atomic_fetch_add(&context->admission_calls, 1U);
-    return VERSE_STATUS_OK;
+    return VERSE_STATUS_INTERNAL;
 }
 
-static verse_status_t sign_tls(
+static verse_status_t reject_tls(
     void *raw,
     verse_bytes_view_t input,
     uint8_t signature[64]) {
     smoke_context_t *context = raw;
-    assert(input.length == sizeof(CREDENTIAL) - 1U);
-    memset(signature, 0x5A, 64U);
+    (void)input;
+    (void)signature;
     atomic_fetch_add(&context->tls_calls, 1U);
-    return VERSE_STATUS_OK;
-}
-
-static void on_state(void *raw, const verse_state_event_v1_t *event) {
-    smoke_context_t *context = raw;
-    assert(event->struct_size == sizeof(*event));
-    assert(event->abi_version == VERSE_CORE_ABI_VERSION_1);
-    atomic_fetch_add(&context->state_calls, 1U);
+    return VERSE_STATUS_INTERNAL;
 }
 
 int main(void) {
@@ -62,35 +54,64 @@ int main(void) {
         .struct_size = sizeof(config),
         .abi_version = VERSE_CORE_ABI_VERSION_1,
         .context = &context,
-        .sign_admission = sign_admission,
-        .sign_tls_ed25519 = sign_tls,
-        .on_state = on_state,
+        .sign_admission = reject_admission,
+        .sign_tls_ed25519 = reject_tls,
     };
     verse_core_t *core = NULL;
+
     assert(verse_core_abi_version() == VERSE_CORE_ABI_VERSION_1);
+    assert(verse_core_create_v1(NULL, &core) == VERSE_STATUS_INVALID_ARGUMENT);
+    assert(verse_core_create_v1(&config, NULL) == VERSE_STATUS_INVALID_ARGUMENT);
+    config.struct_size = 8U;
+    assert(verse_core_create_v1(&config, &core) == VERSE_STATUS_INVALID_ARGUMENT);
+    config.struct_size = sizeof(config);
+    config.abi_version = VERSE_CORE_ABI_VERSION_1 + 1U;
+    assert(verse_core_create_v1(&config, &core) == VERSE_STATUS_UNSUPPORTED_ABI);
+    config.abi_version = VERSE_CORE_ABI_VERSION_1;
     assert(verse_core_create_v1(&config, &core) == VERSE_STATUS_OK);
     assert(core != NULL);
 
-    const verse_connect_request_v1_t request = {
-        .struct_size = sizeof(request),
-        .abi_version = VERSE_CORE_ABI_VERSION_1,
-        .manifest_json = {(const uint8_t *)MANIFEST, sizeof(MANIFEST) - 1U},
-        .tunnel_credential_json = {(const uint8_t *)CREDENTIAL, sizeof(CREDENTIAL) - 1U},
-    };
-    assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_OK);
-
     verse_input_event_v1_t input = {
         .struct_size = sizeof(input),
         .abi_version = VERSE_CORE_ABI_VERSION_1,
         .kind = VERSE_INPUT_KEYBOARD,
         .values = {1, 0, 30},
     };
-    assert(verse_core_send_input_v1(core, &input) == VERSE_STATUS_OK);
-    assert(verse_core_request_idr_v1(core) == VERSE_STATUS_OK);
+    assert(verse_core_send_input_v1(core, &input) == VERSE_STATUS_INVALID_STATE);
+    assert(verse_core_request_idr_v1(core) == VERSE_STATUS_INVALID_STATE);
+
+    char manifest[sizeof(MANIFEST)];
+    char credential[sizeof(CREDENTIAL)];
+    memcpy(manifest, MANIFEST, sizeof(manifest));
+    memcpy(credential, CREDENTIAL, sizeof(credential));
+    verse_connect_request_v1_t request = {
+        .struct_size = sizeof(request),
+        .abi_version = VERSE_CORE_ABI_VERSION_1,
+        .manifest_json = {(const uint8_t *)manifest, sizeof(manifest) - 1U},
+        .tunnel_credential_json = {(const uint8_t *)credential, sizeof(credential) - 1U},
+    };
+    assert(verse_core_connect_v1(NULL, &request) == VERSE_STATUS_INVALID_ARGUMENT);
+    request.struct_size = 8U;
+    assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_INVALID_ARGUMENT);
+    request.struct_size = sizeof(request);
+    request.abi_version = VERSE_CORE_ABI_VERSION_1 + 1U;
+    assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_UNSUPPORTED_ABI);
+    request.abi_version = VERSE_CORE_ABI_VERSION_1;
+    request.manifest_json.data = NULL;
+    assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_INVALID_ARGUMENT);
+    request.manifest_json.data = (const uint8_t *)manifest;
+
+    assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_TLS);
+    memset(manifest, 0, sizeof(manifest));
+    memset(credential, 0, sizeof(credential));
+    assert(verse_core_send_input_v1(core, &input) == VERSE_STATUS_CANCELLED);
+    assert(verse_core_request_idr_v1(core) == VERSE_STATUS_CANCELLED);
     assert(verse_core_cancel_v1(core) == VERSE_STATUS_OK);
     assert(verse_core_cancel_v1(core) == VERSE_STATUS_OK);
+    assert(atomic_load(&context.admission_calls) == 0U);
+    assert(atomic_load(&context.tls_calls) == 0U);
     assert(verse_core_destroy_v1(core, 2000U) == VERSE_STATUS_OK);
-    assert(atomic_load(&context.admission_calls) == 1U);
-    assert(atomic_load(&context.tls_calls) == 1U);
+    assert(atomic_load(&context.admission_calls) == 0U);
+    assert(atomic_load(&context.tls_calls) == 0U);
     return EXIT_SUCCESS;
 }

core/tests/gateway_oracle.rs

@@ -1,5 +1,8 @@
+use std::ffi::c_void;
 use std::fs;
+use std::mem::size_of;
 use std::process::{Child, Command, Stdio};
+use std::ptr;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 use std::thread;
@@ -13,7 +16,8 @@ use rustls::SignatureScheme;
 use serde::Deserialize;
 use versevdi_core::error::CoreError;
 use versevdi_core::transport::{
-    connect, connect_with_cancellation, AdmissionSigner, Cancellation, Signers, TlsEd25519Signer,
+    bounded_session_events, connect, connect_with_cancellation, AdmissionSigner, Cancellation,
+    SessionCommand, SessionEvent, Signers, TlsEd25519Signer,
 };
 use versevdi_core::wire::{ConnectionManifest, NativeTunnelCredential};
 
@@ -76,8 +80,8 @@ func (a *admission) Release(context.Context, protocol.SessionAuthority) error {
     return nil
 }
 
-type provider struct { admission *admission; failStart bool; delay time.Duration }
-type session struct { state protocol.ProviderState; video chan gateway.ProviderMedia; audio chan gateway.ProviderMedia; events chan gateway.ProviderEvent }
+type provider struct { admission *admission; failStart bool; delay time.Duration; mode string; inputPath string; releasePath string }
+type session struct { state protocol.ProviderState; video chan gateway.ProviderMedia; audio chan gateway.ProviderMedia; events chan gateway.ProviderEvent; mode string; inputPath string; releasePath string }
 func (p provider) Start(_ context.Context, request gateway.LaunchRequest) (gateway.ProviderSession, error) {
     p.admission.mu.Lock()
     p.admission.providerStarts++
@@ -86,18 +90,24 @@ func (p provider) Start(_ context.Context, request gateway.LaunchRequest) (gatew
     p.admission.mu.Unlock()
     if p.failStart { return nil, context.DeadlineExceeded }
     if p.delay > 0 { time.Sleep(p.delay) }
-    return &session{state: protocol.ProviderState{Version:"1", SessionID:request.SessionID, State:gateway.ProviderStateReady, Channels:[]string{"video","audio","input","feedback"}}, video:make(chan gateway.ProviderMedia), audio:make(chan gateway.ProviderMedia), events:make(chan gateway.ProviderEvent)}, nil
+    current := &session{state: protocol.ProviderState{Version:"1", SessionID:request.SessionID, State:gateway.ProviderStateReady, Channels:[]string{"video","audio","input","feedback"}}, video:make(chan gateway.ProviderMedia,2), audio:make(chan gateway.ProviderMedia,2), events:make(chan gateway.ProviderEvent,2), mode:p.mode, inputPath:p.inputPath, releasePath:p.releasePath}
+    if p.mode == "io" || p.mode == "input" {
+        current.video <- gateway.ProviderMedia{Payload:bytes.Repeat([]byte{0xA5}, 2000)}
+        current.audio <- gateway.ProviderMedia{Payload:[]byte{0x10,0x20,0x30}}
+        current.events <- gateway.ProviderEvent{Kind:gateway.ProviderEventRumble,Payload:[]byte{0,0,1,0,2}}
+    }
+    return current, nil
 }
 func (s *session) Ready(context.Context) error { return nil }
 func (s *session) Video() <-chan gateway.ProviderMedia { return s.video }
 func (s *session) Audio() <-chan gateway.ProviderMedia { return s.audio }
 func (s *session) Events() <-chan gateway.ProviderEvent { return s.events }
-func (s *session) Input(context.Context, gateway.InputEvent) error { return nil }
+func (s *session) Input(_ context.Context, event gateway.InputEvent) error { if s.mode == "slow-input" { time.Sleep(100*time.Millisecond) }; data,_ := json.Marshal(event); if err := os.WriteFile(s.inputPath,data,0600); err != nil { panic(err) }; return nil }
 func (s *session) Feedback(context.Context, gateway.Feedback) error { return nil }
 func (s *session) ReadClipboard(context.Context) (string, error) { return "", errors.New("disabled") }
 func (s *session) WriteClipboard(context.Context, string) error { return errors.New("disabled") }
 func (s *session) Telemetry() gateway.ProviderTelemetry { return gateway.ProviderTelemetry{State:s.state.State} }
-func (s *session) ReleaseAll(context.Context) error { return nil }
+func (s *session) ReleaseAll(context.Context) error { return os.WriteFile(s.releasePath,[]byte("released"),0600) }
 func (s *session) Terminate(context.Context) error { return nil }
 func (s *session) State() protocol.ProviderState { return s.state }
 
@@ -128,11 +138,11 @@ func main() {
     if mode == "session" { authority.SessionID = "other-session" }
     if mode == "gateway" { authority.GatewayID = "other-gateway" }
     if mode == "audience" { authority.Audience = "other-audience" }
-    if mode == "reconnect" { authority.ReconnectSequence = 1 }
+    if mode == "reconnect" || mode == "reconnect-success" { authority.ReconnectSequence = 1 }
     if mode == "expiry" { authority.ExpiresAt = time.Now().Add(-time.Minute).UTC().Truncate(time.Second).Format(time.RFC3339) }
     if mode == "capability" { authority.Capabilities.ClientDecode = []string{"hevc-opus"} }
     if mode == "alpn" { server.NextProtos = []string{"wrong-alpn"} }
-    work := protocol.ProviderSessionWork{Version:"1",SessionID:authority.SessionID,GatewayID:authority.GatewayID,ExpiresAt:expiry,ProviderProfile:gateway.ProviderProfileApollo,ProviderIdentity:authority.ProviderIdentity,PolicyVersionID:"policy-1",StreamPolicy:protocol.ProviderStreamPolicy{ResolutionWidth:1920,ResolutionHeight:1080,Fps:60,Codec:"H264",BitrateKbps:8000,AudioEnabled:true},ApplicationID:"1",ClientID:"client",ManagementHost:"provider.invalid",ManagementPort:47990,StreamHost:"provider.invalid",StreamPort:47984,ClientCertificatePem:"certificate",ClientPrivateKeyPem:"private-key",ServerCertificatePem:"certificate",ClipboardPolicy:protocol.ClipboardPolicy{MaxTextBytes:65536,MaxUpdatesPerMinute:30}}
+    work := protocol.ProviderSessionWork{Version:"1",SessionID:authority.SessionID,GatewayID:authority.GatewayID,ReconnectSequence:authority.ReconnectSequence,ExpiresAt:expiry,ProviderProfile:gateway.ProviderProfileApollo,ProviderIdentity:authority.ProviderIdentity,PolicyVersionID:"policy-1",StreamPolicy:protocol.ProviderStreamPolicy{ResolutionWidth:1920,ResolutionHeight:1080,Fps:60,Codec:"H264",BitrateKbps:8000,AudioEnabled:true},ApplicationID:"1",ClientID:"client",ManagementHost:"provider.invalid",ManagementPort:47990,StreamHost:"provider.invalid",StreamPort:47984,ClientCertificatePem:"certificate",ClientPrivateKeyPem:"private-key",ServerCertificatePem:"certificate",ClipboardPolicy:protocol.ClipboardPolicy{MaxTextBytes:65536,MaxUpdatesPerMinute:30}}
     replayGuard := &admission{authority:authority,work:work,public:admissionKey.Public().(ed25519.PublicKey)}
     replayRequest := protocol.TunnelAdmissionRequest{Version:"1",SessionID:"session-1",GatewayID:"gateway-1",Audience:"versevdi-gateway",Grant:"ggggggggggggggggggggggggggggggggggggggggggg",ClientNonce:"bm9uY2UtZm9yLXJlcGxheQ",Capabilities:gateway.DefaultCapabilities()}
     replayRequest.DeviceSignature = base64.RawURLEncoding.EncodeToString(ed25519.Sign(admissionKey,replayRequest.DeviceAdmissionTranscript()))
@@ -140,10 +150,12 @@ func main() {
     _, secondReplayErr := replayGuard.Admit(context.Background(),replayRequest)
     replayRejected := firstReplayErr == nil && errors.Is(secondReplayErr,gateway.ErrAdmissionRejected)
     mutationPath := readyPath + ".provider-starts"
+    inputPath := readyPath + ".input"
+    releasePath := readyPath + ".release-all"
     if err := os.WriteFile(mutationPath, []byte("0"), 0600); err != nil { panic(err) }
     admissionService := &admission{authority:authority,work:work,public:admissionKey.Public().(ed25519.PublicKey),reusable:mode == "reusable",failWorkOnce:mode == "post-retryable",releaseFails:mode == "cleanup-release-failure",mutationPath:mutationPath}
     newService := func(delay time.Duration) *gateway.Server {
-        service, err := gateway.NewServer(gateway.ServerConfig{ListenAddress:"127.0.0.1:0",TLSConfig:server,GatewayID:authority.GatewayID,Admission:admissionService,Provider:provider{admission:admissionService,failStart:mode == "provider-start-lost-response" || mode == "cleanup-release-failure",delay:delay}}); if err != nil { panic(err) }
+        service, err := gateway.NewServer(gateway.ServerConfig{ListenAddress:"127.0.0.1:0",TLSConfig:server,GatewayID:authority.GatewayID,Admission:admissionService,Provider:provider{admission:admissionService,failStart:mode == "provider-start-lost-response" || mode == "cleanup-release-failure",delay:delay,mode:mode,inputPath:inputPath,releasePath:releasePath}}); if err != nil { panic(err) }
         return service
     }
     firstDelay := time.Duration(0); if mode == "lost-authority" { firstDelay = 300*time.Millisecond }
@@ -165,7 +177,8 @@ func main() {
     done := make(chan error,len(services)); for _, current := range services { go func(server *gateway.Server){done <- server.Serve(ctx)}(current) }
     keyDER, err := x509.MarshalPKCS8PrivateKey(clientKey); if err != nil { panic(err) }
     admissionKeyDER, err := x509.MarshalPKCS8PrivateKey(admissionKey); if err != nil { panic(err) }
-    manifest := map[string]any{"version":"1","purpose":"launch","session_id":"session-1","reconnect_sequence":0,"gateway":map[string]any{"id":"gateway-1","addresses":addresses,"public_identity":"gateway.test"},"tunnel":map[string]any{"versions":[]string{"verse-gateway-v1/1"},"features":[]string{"control.v1"}},"profile":map[string]any{"id":"standard","bounds":map[string]any{"minimum_kbps":1000,"target_kbps":5000,"maximum_kbps":10000},"display_mode":map[string]any{"resolution_width":1920,"resolution_height":1080,"fps":60}},"grant":map[string]any{"opaque_value":"ggggggggggggggggggggggggggggggggggggggggggg","expires_at":"2099-01-01T00:00:00Z","audience":"versevdi-gateway"},"correlation_id":"oracle"}
+    purpose := "launch"; reconnectSequence := 0; if mode == "reconnect-success" { purpose = "reconnect"; reconnectSequence = 1 }
+    manifest := map[string]any{"version":"1","purpose":purpose,"session_id":"session-1","reconnect_sequence":reconnectSequence,"gateway":map[string]any{"id":"gateway-1","addresses":addresses,"public_identity":"gateway.test"},"tunnel":map[string]any{"versions":[]string{"verse-gateway-v1/1"},"features":[]string{"control.v1","input.absolute.v1","input.scroll.v1"}},"profile":map[string]any{"id":"standard","bounds":map[string]any{"minimum_kbps":1000,"target_kbps":5000,"maximum_kbps":10000},"display_mode":map[string]any{"resolution_width":1920,"resolution_height":1080,"fps":60}},"grant":map[string]any{"opaque_value":"ggggggggggggggggggggggggggggggggggggggggggg","expires_at":"2099-01-01T00:00:00Z","audience":"versevdi-gateway"},"correlation_id":"oracle"}
     credential := map[string]any{"client_device_id":"device","device_key_id":"key","certificate_chain_pem":pemCert(clientDER,caDER),"trust_bundle_pem":pemCert(caDER),"expires_at":"2099-01-01T00:00:00Z"}
     manifestJSON,_ := json.Marshal(manifest); credentialJSON,_ := json.Marshal(credential)
     wrongKeyDER, err := x509.MarshalPKCS8PrivateKey(wrongClientKey); if err != nil { panic(err) }
@@ -178,44 +191,168 @@ func main() {
 "#;
 
 #[derive(Deserialize)]
-struct Ready {
-    manifest: String,
-    credential: String,
-    admission_key: String,
-    client_key: String,
-    wrong_client_key: String,
-    server_key: String,
-    server_chain: String,
-    bad_trust: String,
-    replay_rejected: bool,
+pub(crate) struct Ready {
+    pub(crate) manifest: String,
+    pub(crate) credential: String,
+    pub(crate) admission_key: String,
+    pub(crate) client_key: String,
+    pub(crate) wrong_client_key: String,
+    pub(crate) server_key: String,
+    pub(crate) server_chain: String,
+    pub(crate) bad_trust: String,
+    pub(crate) replay_rejected: bool,
 }
 
-struct Oracle {
-    _serial: std::sync::MutexGuard<'static, ()>,
+pub(crate) struct Oracle {
     child: Child,
     directory: std::path::PathBuf,
-    ready: Ready,
+    pub(crate) ready: Ready,
+}
+
+struct OracleStartup {
+    child: Option<Child>,
+    directory: Option<std::path::PathBuf>,
+}
+
+impl OracleStartup {
+    fn new(directory: std::path::PathBuf) -> Self {
+        Self {
+            child: None,
+            directory: Some(directory),
+        }
+    }
+
+    fn finish(mut self, ready: Ready) -> Oracle {
+        Oracle {
+            child: self.child.take().expect("started oracle child"),
+            directory: self.directory.take().expect("oracle directory"),
+            ready,
+        }
+    }
+}
+
+impl Drop for OracleStartup {
+    fn drop(&mut self) {
+        if let Some(child) = self.child.as_mut() {
+            stop_child(child, self.directory.as_deref());
+        }
+        if let Some(directory) = self.directory.as_ref() {
+            let _ = fs::remove_dir_all(directory);
+        }
+    }
+}
+
+#[test]
+fn production_oracles_can_run_concurrently_with_isolated_state() {
+    let (first_ready_tx, first_ready_rx) = std::sync::mpsc::channel();
+    let (release_tx, release_rx) = std::sync::mpsc::channel();
+    let first = thread::spawn(move || {
+        let oracle = Oracle::start("");
+        first_ready_tx.send(()).expect("first ready");
+        release_rx.recv().expect("release first oracle");
+        drop(oracle);
+    });
+    first_ready_rx
+        .recv_timeout(Duration::from_secs(20))
+        .expect("first oracle ready");
+    let (second_ready_tx, second_ready_rx) = std::sync::mpsc::channel();
+    let second = thread::spawn(move || {
+        let oracle = Oracle::start("");
+        second_ready_tx.send(()).expect("second ready");
+        drop(oracle);
+    });
+    let concurrent = second_ready_rx.recv_timeout(Duration::from_secs(5)).is_ok();
+    release_tx.send(()).expect("release first");
+    first.join().expect("first oracle");
+    second.join().expect("second oracle");
+    assert!(concurrent, "second isolated oracle was globally serialized");
+}
+
+#[test]
+fn oracle_startup_timeout_kills_child_and_removes_directory() {
+    let directory = Oracle::unique_directory();
+    let mut command = Command::new("sh");
+    command.args(["-c", "while :; do sleep 1; done"]);
+    let started = Instant::now();
+
+    let result = Oracle::start_in("", directory.clone(), command, Duration::from_millis(20));
+
+    assert!(result.is_err(), "non-ready child unexpectedly became ready");
+    assert!(started.elapsed() < Duration::from_secs(2));
+    assert!(!directory.exists(), "timed-out oracle directory leaked");
+}
+
+#[test]
+fn oracle_source_failure_removes_directory_before_spawn() {
+    let directory = Oracle::unique_directory();
+    fs::create_dir(directory.join("main.go")).expect("block source file creation");
+
+    let result = Oracle::start_in(
+        "",
+        directory.clone(),
+        Command::new("go"),
+        Duration::from_millis(20),
+    );
+
+    assert!(result.is_err(), "invalid source path unexpectedly started");
+    assert!(!directory.exists(), "failed oracle directory leaked");
+}
+
+#[test]
+fn oracle_spawn_failure_removes_directory() {
+    let directory = Oracle::unique_directory();
+
+    let result = Oracle::start_in(
+        "",
+        directory.clone(),
+        Command::new("/definitely/missing/versevdi-go"),
+        Duration::from_millis(20),
+    );
+
+    assert!(result.is_err(), "missing executable unexpectedly started");
+    assert!(!directory.exists(), "spawn-failed oracle directory leaked");
 }
 
 impl Oracle {
-    fn start(mode: &str) -> Self {
-        static SERIAL: Mutex<()> = Mutex::new(());
+    pub(crate) fn start(mode: &str) -> Self {
+        let directory = Self::unique_directory();
+        Self::start_in(mode, directory, Command::new("go"), Duration::from_secs(20))
+            .unwrap_or_else(|error| panic!("{error}"))
+    }
+
+    fn unique_directory() -> std::path::PathBuf {
         static NEXT: AtomicUsize = AtomicUsize::new(1);
-        let serial = SERIAL
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        let directory = std::env::temp_dir().join(format!(
-            "versevdi-rust-gateway-oracle-{}-{}",
-            std::process::id(),
-            NEXT.fetch_add(1, Ordering::Relaxed)
-        ));
-        fs::create_dir(&directory).expect("create oracle directory");
+        loop {
+            let candidate = std::env::temp_dir().join(format!(
+                "versevdi-rust-gateway-oracle-{}-{}",
+                std::process::id(),
+                NEXT.fetch_add(1, Ordering::Relaxed)
+            ));
+            match fs::create_dir(&candidate) {
+                Ok(()) => break candidate,
+                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
+                Err(error) => panic!("create oracle directory: {error}"),
+            }
+        }
+    }
+
+    fn start_in(
+        mode: &str,
+        directory: std::path::PathBuf,
+        mut command: Command,
+        timeout: Duration,
+    ) -> Result<Self, String> {
+        let mut startup = OracleStartup::new(directory.clone());
         let source = directory.join("main.go");
         let ready_path = directory.join("ready.json");
         let stop_path = directory.join("stop");
-        fs::write(&source, ORACLE_SOURCE).expect("write oracle source");
-        let child = Command::new("go")
-            .args(["run", source.to_str().expect("source path is UTF-8")])
+        fs::write(&source, ORACLE_SOURCE)
+            .map_err(|error| format!("write Go oracle source: {error}"))?;
+        let source = source
+            .to_str()
+            .ok_or_else(|| "Go oracle source path is not UTF-8".to_owned())?;
+        let child = command
+            .args(["run", source])
             .arg(&ready_path)
             .arg(&stop_path)
             .env("VERSEVDI_RUST_ORACLE_MODE", mode)
@@ -223,28 +360,29 @@ impl Oracle {
             .stdout(Stdio::piped())
             .stderr(Stdio::piped())
             .spawn()
-            .expect("start Go gateway oracle");
-        let deadline = Instant::now() + Duration::from_secs(20);
+            .map_err(|error| format!("start Go gateway oracle: {error}"))?;
+        startup.child = Some(child);
+        let deadline = Instant::now() + timeout;
         while Instant::now() < deadline {
             if let Ok(bytes) = fs::read(&ready_path) {
                 if let Ok(ready) = serde_json::from_slice(&bytes) {
-                    return Self {
-                        _serial: serial,
-                        child,
-                        directory,
-                        ready,
-                    };
+                    return Ok(startup.finish(ready));
                 }
             }
+            if let Some(status) = startup
+                .child
+                .as_mut()
+                .expect("started oracle child")
+                .try_wait()
+                .map_err(|error| format!("poll Go gateway oracle: {error}"))?
+            {
+                return Err(format!(
+                    "Go gateway oracle exited before becoming ready: status={status}"
+                ));
+            }
             thread::sleep(Duration::from_millis(10));
         }
-        let output = child.wait_with_output().expect("collect Go oracle output");
-        panic!(
-            "Go gateway oracle did not become ready: status={} stdout={} stderr={}",
-            output.status,
-            String::from_utf8_lossy(&output.stdout),
-            String::from_utf8_lossy(&output.stderr)
-        );
+        Err("Go gateway oracle did not become ready before the startup deadline".to_owned())
     }
 
     fn provider_starts(&self) -> usize {
@@ -253,17 +391,60 @@ impl Oracle {
             .parse()
             .expect("provider mutation count")
     }
+
+    fn input(&self) -> Option<Vec<u8>> {
+        fs::read(self.directory.join("ready.json.input")).ok()
+    }
+
+    fn release_all(&self) -> bool {
+        self.directory.join("ready.json.release-all").exists()
+    }
+
+    fn files(&self) -> Vec<String> {
+        fs::read_dir(&self.directory)
+            .expect("read oracle directory")
+            .map(|entry| {
+                entry
+                    .expect("oracle entry")
+                    .file_name()
+                    .to_string_lossy()
+                    .into_owned()
+            })
+            .collect()
+    }
 }
 
 impl Drop for Oracle {
     fn drop(&mut self) {
-        let _ = fs::write(self.directory.join("stop"), []);
-        let _ = self.child.wait();
+        stop_child(&mut self.child, Some(&self.directory));
         let _ = fs::remove_dir_all(&self.directory);
     }
 }
 
-fn test_key(encoded: &str) -> Arc<dyn SigningKey> {
+fn stop_child(child: &mut Child, directory: Option<&std::path::Path>) {
+    if let Some(directory) = directory {
+        let _ = fs::write(directory.join("stop"), []);
+    }
+    if wait_for_child(child, Duration::from_millis(250)) {
+        return;
+    }
+    if child.kill().is_ok() {
+        let _ = wait_for_child(child, Duration::from_millis(500));
+    }
+}
+
+fn wait_for_child(child: &mut Child, timeout: Duration) -> bool {
+    let deadline = Instant::now() + timeout;
+    loop {
+        match child.try_wait() {
+            Ok(Some(_)) => return true,
+            Ok(None) if Instant::now() < deadline => thread::sleep(Duration::from_millis(5)),
+            Ok(None) | Err(_) => return false,
+        }
+    }
+}
+
+pub(crate) fn test_key(encoded: &str) -> Arc<dyn SigningKey> {
     let der = STANDARD.decode(encoded).expect("decode test key");
     rustls::crypto::ring::default_provider()
         .key_provider
@@ -339,6 +520,263 @@ fn callback_ed25519_signer_completes_tls13_quic_admission_without_private_key_in
     runtime.block_on(session.close());
 }
 
+#[test]
+fn production_gateway_session_delivers_bounded_media_control_and_ordered_input() {
+    let oracle = Oracle::start("input");
+    let runtime = tokio::runtime::Runtime::new().expect("runtime");
+    let session = runtime
+        .block_on(connect_oracle_async(
+            &oracle,
+            &oracle.ready.admission_key,
+            &oracle.ready.client_key,
+            |_| {},
+            |_| {},
+        ))
+        .expect("production Go gateway admission");
+    let (command_tx, command_rx) = tokio::sync::mpsc::channel(64);
+    let (event_tx, _event_rx) = bounded_session_events();
+    let cancellation = Cancellation::new();
+    let running_cancellation = cancellation.clone();
+    let mut running = runtime.spawn(session.run(command_rx, event_tx, running_cancellation));
+    runtime
+        .block_on(command_tx.send(SessionCommand::Input(
+            b"VGI1\x01\x04\x01\x00\x00\x1e".to_vec(),
+        )))
+        .expect("bounded input");
+    runtime.block_on(async { tokio::time::sleep(Duration::from_millis(100)).await });
+
+    let input_deadline = Instant::now() + Duration::from_secs(2);
+    while oracle.input().is_none() && Instant::now() < input_deadline {
+        thread::sleep(Duration::from_millis(10));
+    }
+    let early = runtime
+        .block_on(async { tokio::time::timeout(Duration::from_millis(10), &mut running).await });
+    if let (None, Ok(completed)) = (oracle.input(), early) {
+        panic!(
+            "gateway did not receive input; session result={:?}",
+            completed.expect("session task")
+        );
+    }
+    assert!(
+        oracle.input().is_some(),
+        "gateway did not receive input: {:?}",
+        oracle.files()
+    );
+
+    cancellation.cancel();
+    assert_eq!(
+        runtime.block_on(running).expect("session task"),
+        Err(CoreError::Cancelled)
+    );
+    let release_deadline = Instant::now() + Duration::from_secs(2);
+    while !oracle.release_all() && Instant::now() < release_deadline {
+        thread::sleep(Duration::from_millis(10));
+    }
+    assert!(
+        oracle.release_all(),
+        "gateway did not release pressed input"
+    );
+}
+
+#[test]
+fn production_gateway_session_delivers_bounded_media_and_control() {
+    let oracle = Oracle::start("io");
+    let runtime = tokio::runtime::Runtime::new().expect("runtime");
+    let session = runtime
+        .block_on(connect_oracle_async(
+            &oracle,
+            &oracle.ready.admission_key,
+            &oracle.ready.client_key,
+            |_| {},
+            |_| {},
+        ))
+        .expect("production Go gateway admission");
+    let (_command_tx, command_rx) = tokio::sync::mpsc::channel(64);
+    let (event_tx, event_rx) = bounded_session_events();
+    let cancellation = Cancellation::new();
+    let running_cancellation = cancellation.clone();
+    let running = runtime.spawn(session.run(command_rx, event_tx, running_cancellation));
+    let mut media = Vec::new();
+    let mut control = Vec::new();
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while (media.len() < 2 || control.is_empty()) && Instant::now() < deadline {
+        match event_rx.recv_timeout(Duration::from_millis(100)) {
+            Ok(SessionEvent::Media(unit)) => media.push(unit),
+            Ok(SessionEvent::Control { kind, payload }) => control.push((kind, payload)),
+            Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
+            Err(error) => panic!("session event channel: {error}"),
+        }
+    }
+    media.sort_by_key(|unit| unit.channel as u8);
+    assert_eq!(media.len(), 2);
+    assert!(media.iter().any(|unit| unit.payload == vec![0xA5; 2000]));
+    assert!(media
+        .iter()
+        .any(|unit| unit.payload == vec![0x10, 0x20, 0x30]));
+    assert_eq!(control, vec![(0x11, vec![0, 0, 1, 0, 2])]);
+    cancellation.cancel();
+    assert_eq!(
+        runtime.block_on(running).expect("session task"),
+        Err(CoreError::Cancelled)
+    );
+}
+
+#[repr(C)]
+struct AbiCore {
+    _private: [u8; 0],
+}
+#[repr(C)]
+#[derive(Clone, Copy)]
+struct AbiBytes {
+    data: *const u8,
+    length: usize,
+}
+#[repr(C)]
+struct AbiConfig {
+    struct_size: u32,
+    abi_version: u32,
+    context: *mut c_void,
+    sign_admission: Option<unsafe extern "C" fn(*mut c_void, AbiBytes, *mut u8) -> u32>,
+    sign_tls_ed25519: Option<unsafe extern "C" fn(*mut c_void, AbiBytes, *mut u8) -> u32>,
+    on_state: Option<unsafe extern "C" fn(*mut c_void, *const u8)>,
+    on_error: Option<unsafe extern "C" fn(*mut c_void, *const u8)>,
+    on_stats: Option<unsafe extern "C" fn(*mut c_void, *const u8)>,
+    on_media: Option<unsafe extern "C" fn(*mut c_void, *const u8)>,
+    on_control: Option<unsafe extern "C" fn(*mut c_void, *const u8)>,
+}
+#[repr(C)]
+struct AbiConnect {
+    struct_size: u32,
+    abi_version: u32,
+    manifest_json: AbiBytes,
+    tunnel_credential_json: AbiBytes,
+}
+#[repr(C)]
+struct AbiInput {
+    struct_size: u32,
+    abi_version: u32,
+    kind: u32,
+    flags: u32,
+    values: [i32; 12],
+}
+unsafe extern "C" {
+    fn verse_core_create_v1(config: *const AbiConfig, out: *mut *mut AbiCore) -> u32;
+    fn verse_core_connect_v1(core: *mut AbiCore, request: *const AbiConnect) -> u32;
+    fn verse_core_send_input_v1(core: *mut AbiCore, event: *const AbiInput) -> u32;
+    fn verse_core_cancel_v1(core: *mut AbiCore) -> u32;
+    fn verse_core_destroy_v1(core: *mut AbiCore, timeout_ms: u32) -> u32;
+}
+struct AbiSigners {
+    admission: Arc<dyn SigningKey>,
+    tls: Arc<dyn SigningKey>,
+    admission_threads: Mutex<Vec<thread::ThreadId>>,
+    tls_threads: Mutex<Vec<thread::ThreadId>>,
+}
+unsafe extern "C" fn abi_admission(raw: *mut c_void, input: AbiBytes, output: *mut u8) -> u32 {
+    unsafe { abi_sign(raw, input, output, true) }
+}
+unsafe extern "C" fn abi_tls(raw: *mut c_void, input: AbiBytes, output: *mut u8) -> u32 {
+    unsafe { abi_sign(raw, input, output, false) }
+}
+unsafe fn abi_sign(raw: *mut c_void, input: AbiBytes, output: *mut u8, admission: bool) -> u32 {
+    let context = unsafe { &*raw.cast::<AbiSigners>() };
+    let threads = if admission {
+        &context.admission_threads
+    } else {
+        &context.tls_threads
+    };
+    threads
+        .lock()
+        .expect("signer threads")
+        .push(thread::current().id());
+    let bytes = unsafe { std::slice::from_raw_parts(input.data, input.length) };
+    let key = if admission {
+        &context.admission
+    } else {
+        &context.tls
+    };
+    let signature = key
+        .choose_scheme(&[SignatureScheme::ED25519])
+        .expect("Ed25519")
+        .sign(bytes)
+        .expect("sign");
+    unsafe { ptr::copy_nonoverlapping(signature.as_ptr(), output, 64) };
+    0
+}
+
+#[test]
+fn real_abi_exports_connect_send_cancel_and_destroy_the_go_gateway_session() {
+    let oracle = Oracle::start("input");
+    let mut signers = Box::new(AbiSigners {
+        admission: test_key(&oracle.ready.admission_key),
+        tls: test_key(&oracle.ready.client_key),
+        admission_threads: Mutex::new(Vec::new()),
+        tls_threads: Mutex::new(Vec::new()),
+    });
+    let config = AbiConfig {
+        struct_size: size_of::<AbiConfig>() as u32,
+        abi_version: 1,
+        context: ptr::from_mut(&mut *signers).cast(),
+        sign_admission: Some(abi_admission),
+        sign_tls_ed25519: Some(abi_tls),
+        on_state: None,
+        on_error: None,
+        on_stats: None,
+        on_media: None,
+        on_control: None,
+    };
+    let mut core = ptr::null_mut();
+    assert_eq!(unsafe { verse_core_create_v1(&config, &mut core) }, 0);
+    let request = AbiConnect {
+        struct_size: size_of::<AbiConnect>() as u32,
+        abi_version: 1,
+        manifest_json: AbiBytes {
+            data: oracle.ready.manifest.as_ptr(),
+            length: oracle.ready.manifest.len(),
+        },
+        tunnel_credential_json: AbiBytes {
+            data: oracle.ready.credential.as_ptr(),
+            length: oracle.ready.credential.len(),
+        },
+    };
+    let calling_thread = thread::current().id();
+    assert_eq!(unsafe { verse_core_connect_v1(core, &request) }, 0);
+    assert_eq!(
+        signers
+            .admission_threads
+            .lock()
+            .expect("admission threads")
+            .as_slice(),
+        [calling_thread]
+    );
+    assert_eq!(
+        signers.tls_threads.lock().expect("TLS threads").as_slice(),
+        [calling_thread]
+    );
+    let mut values = [0; 12];
+    values[0] = 1;
+    values[2] = 30;
+    let input = AbiInput {
+        struct_size: size_of::<AbiInput>() as u32,
+        abi_version: 1,
+        kind: 1,
+        flags: 0,
+        values,
+    };
+    assert_eq!(unsafe { verse_core_send_input_v1(core, &input) }, 0);
+    let deadline = Instant::now() + Duration::from_secs(2);
+    while oracle.input().is_none() && Instant::now() < deadline {
+        thread::sleep(Duration::from_millis(10));
+    }
+    assert!(oracle.input().is_some(), "ABI input did not reach gateway");
+    assert_eq!(unsafe { verse_core_cancel_v1(core) }, 0);
+    assert_eq!(unsafe { verse_core_destroy_v1(core, 2_000) }, 0);
+    assert!(
+        oracle.release_all(),
+        "destroy did not release provider input"
+    );
+}
+
 fn signer(key: Arc<dyn SigningKey>) -> impl Fn(&[u8]) -> Result<[u8; 64], CoreError> {
     move |input| {
         key.choose_scheme(&[SignatureScheme::ED25519])
@@ -356,6 +794,24 @@ fn connect_oracle(
     tls_key: &str,
     mutate_manifest: impl FnOnce(&mut serde_json::Value),
     mutate_credential: impl FnOnce(&mut serde_json::Value),
+) -> Result<versevdi_core::transport::TransportSession, CoreError> {
+    tokio::runtime::Runtime::new()
+        .expect("runtime")
+        .block_on(connect_oracle_async(
+            oracle,
+            admission_key,
+            tls_key,
+            mutate_manifest,
+            mutate_credential,
+        ))
+}
+
+async fn connect_oracle_async(
+    oracle: &Oracle,
+    admission_key: &str,
+    tls_key: &str,
+    mutate_manifest: impl FnOnce(&mut serde_json::Value),
+    mutate_credential: impl FnOnce(&mut serde_json::Value),
 ) -> Result<versevdi_core::transport::TransportSession, CoreError> {
     let mut manifest: serde_json::Value =
         serde_json::from_str(&oracle.ready.manifest).expect("manifest JSON");
@@ -368,8 +824,7 @@ fn connect_oracle(
     let credential = NativeTunnelCredential::decode(
         &serde_json::to_vec(&credential).expect("encode credential"),
     )?;
-    let runtime = tokio::runtime::Runtime::new().expect("runtime");
-    runtime.block_on(connect(
+    connect(
         &manifest,
         &credential,
         Signers::new(
@@ -378,7 +833,8 @@ fn connect_oracle(
         ),
         "2026-08-12T00:00:00Z",
         Duration::from_secs(5),
-    ))
+    )
+    .await
 }
 
 #[test]
@@ -465,6 +921,22 @@ fn production_gateway_and_rust_reject_authority_binding_and_capability_mismatche
     }
 }
 
+#[test]
+fn production_gateway_accepts_reconnect_only_when_exact_sequence_is_bound() {
+    let oracle = Oracle::start("reconnect-success");
+    let session = connect_oracle(
+        &oracle,
+        &oracle.ready.admission_key,
+        &oracle.ready.client_key,
+        |_| {},
+        |_| {},
+    )
+    .expect("exact reconnect sequence");
+    tokio::runtime::Runtime::new()
+        .expect("runtime")
+        .block_on(session.close());
+}
+
 #[test]
 fn signer_purposes_are_not_interchangeable() {
     let oracle = Oracle::start("");

core/tests/protocol_fixtures.rs

@@ -533,6 +533,16 @@ fn reassembly_discards_conflicts_and_expires_after_250_ms() {
     assert_eq!(reassembler.expired_units(), 1);
 }
 
+#[test]
+fn reassembly_expires_at_exactly_250_ms() {
+    let mut reassembler = Reassembler::new();
+    let expiring = MediaFragment::new_video(2, 1, 0, 2, vec![1]).expect("fragment");
+    reassembler.push(expiring, 10).expect("first");
+    let next = MediaFragment::new_video(3, 1, 0, 2, vec![1]).expect("fragment");
+    reassembler.push(next, 260).expect("expiry sweep");
+    assert_eq!(reassembler.expired_units(), 1);
+}
+
 #[test]
 fn reassembly_rejects_a_complete_unit_above_one_mebibyte() {
     let mut reassembler = Reassembler::new();
@@ -565,6 +575,14 @@ fn input_features_and_feedback_booleans_fail_closed() {
     );
 }
 
+#[test]
+fn disconnected_gateway_feedback_is_provider_free_control() {
+    assert_eq!(
+        decode_feedback(&decode_hex("5647463101130000")).expect("disconnected feedback"),
+        FeedbackEvent::Disconnected
+    );
+}
+
 #[test]
 fn fec_feedback_enforces_rc5_go_field_invariants_on_decode_and_encode() {
     for invalid in [