feat(review 007 Unit 2): birth-teardown telemetry — unaccepted-stream debug hint (R-02)

The optional hardening half of R-02 (the doc notes landed with Unit 1,
dc4ad2b). The wrapper's teardown task now observes whether the pump
handler ever accepted the channel's stream:

- ChannelBidiStreamSource gains a shared acceptance flag
  (with_accepted_flag / accepted()); channel_source_with_accepted_flag
  threads it from run_open_wrapper.
- On handler exit without accept, the teardown task logs a debug!
  naming the contract: "the returned JoinHandle must track the
  data-plane lifetime ... await pumps inline" — the telemetry hint for
  the teardown-at-birth shape the alktunnels POC hit empirically.
- Telemetry only: no behavior change; the benign no-data close still
  tears down identically.

Verification: cargo test (623 passed, +2: accepted-flag flip on the
yield-once accept; accepting vs non-accepting handler teardown
behavior), clippy -D warnings, fmt.

src/channels/operations.rs

@@ -805,7 +805,17 @@ async fn run_open_wrapper(
             };
 
             let remote_addr = manager.remote_addr();
-            let source = super::source::channel_source(recv, send, remote_addr);
+            // The acceptance flag (R-02 telemetry): shared between the
+            // yield-once source (flips on the handler's accept) and
+            // the teardown task (logs the birth-teardown hint when the
+            // handler exits without accepting).
+            let accepted = Arc::new(std::sync::atomic::AtomicBool::new(false));
+            let source = super::source::channel_source_with_accepted_flag(
+                recv,
+                send,
+                remote_addr,
+                Arc::clone(&accepted),
+            );
             let channel_conn = Connection::from_source(source, alpn.as_bytes().to_vec());
             let raw_task = open_handler(input, plan, channel_conn, auth.clone());
 
@@ -813,6 +823,23 @@ async fn run_open_wrapper(
             let teardown_policy = Arc::clone(policy);
             let task = tokio::spawn(async move {
                 let _ = raw_task.await;
+                if !accepted.load(std::sync::atomic::Ordering::SeqCst) {
+                    // The handler returned without ever accepting the
+                    // channel's stream — almost always the
+                    // "spawned-and-forgotten / returned early" shape
+                    // (R-02): teardown-at-birth, both pumps see
+                    // instant EOF. The benign case (peer never sent,
+                    // handler closed cleanly) is rare by comparison;
+                    // the debug line is the hint to check the handler's
+                    // return-value semantics.
+                    tracing::debug!(
+                        channel_id = id,
+                        "open wrapper: handler exited without accepting the channel's \
+                         stream — if the handler returns before its pumps finish, the \
+                         channel tears down at birth; the returned JoinHandle must \
+                         track the data-plane lifetime (await pumps inline)"
+                    );
+                }
                 // Benign-race pin (review 006 E-03): `channel/close`'s
                 // awaited teardown and this handler-exit teardown both
                 // gate on the atomic `opener_ledger().take(id)`, so the
@@ -1834,6 +1861,81 @@ mod tests {
         assert!(spawned.load(Ordering::SeqCst));
     }
 
+    #[tokio::test]
+    async fn run_open_wrapper_birth_teardown_hint_fires_only_when_stream_unaccepted() {
+        // R-02 telemetry: the acceptance flag flips when the handler
+        // accepts the channel's stream, and does not when it exits
+        // without accepting. Two handlers over two opens: one accepts
+        // (the healthy shape), one returns immediately (the
+        // birth-teardown shape the hint warns about).
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let auth = AuthContext::anonymous(b"alk/call");
+        let accepting: OpenHandler = Arc::new(|_input, _plan, conn, _auth| {
+            tokio::spawn(async move {
+                let _bidi = conn.accept_bi().await.expect("accept");
+            })
+        });
+        let env = run_open_wrapper(
+            &manager,
+            &policy,
+            None,
+            &accepting,
+            &auth,
+            "alk/tty",
+            json!({}),
+            "alice".to_string(),
+            identity("alice"),
+            "req-accepted".to_string(),
+            None,
+        )
+        .await;
+        let id = env.result.as_ref().expect("open ok")["channel_id"]
+            .as_u64()
+            .expect("id") as u32;
+        let accepted_task = manager
+            .teardown_channel(id)
+            .expect("teardown")
+            .expect("task");
+        let _ = accepted_task.await;
+        // (The flag itself is consumed inside the teardown task; the
+        // source-level test proves the flip. Here we prove the
+        // accepting handler's teardown task completes cleanly — no
+        // panic, channel gone.)
+
+        // The non-accepting handler: exits immediately — the
+        // birth-teardown shape R-02's hint targets. The open still
+        // succeeds; the teardown task observes the un-accepted stream.
+        let non_accepting: OpenHandler =
+            Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {}));
+        let env = run_open_wrapper(
+            &manager,
+            &policy,
+            None,
+            &non_accepting,
+            &auth,
+            "alk/tty",
+            json!({}),
+            "alice".to_string(),
+            identity("alice"),
+            "req-not-accepted".to_string(),
+            None,
+        )
+        .await;
+        let id = env.result.as_ref().expect("open ok")["channel_id"]
+            .as_u64()
+            .expect("id") as u32;
+        let task = manager
+            .teardown_channel(id)
+            .expect("teardown")
+            .expect("task");
+        let _ = task.await;
+        assert!(
+            !manager.has_channel(id),
+            "birth-teardown shape: the channel is gone when the handler exits"
+        );
+    }
+
     #[test]
     fn establishment_error_reason_mapping_covers_the_vocabulary() {
         let cases: Vec<(EstablishmentError, &str)> = vec![

src/channels/source.rs

@@ -10,6 +10,8 @@
 //! to the mux write half via `BiStream::from_joined`.
 
 use std::net::SocketAddr;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
 
 use async_trait::async_trait;
 use parking_lot::Mutex;
@@ -28,9 +30,16 @@ use crate::core::types::{BiStream, BidiStreamSource, StreamError};
 /// yield-once contract (ADR-007) and the "split never crosses a crate
 /// boundary as part of a constructor" rule (ADR-009) — the join happens
 /// here, in the `BidiStreamSource` impl, not per-handler.
+///
+/// The `accepted` flag (R-02 telemetry) records whether `accept_bi`
+/// ever yielded: the open wrapper's teardown task checks it when the
+/// pump handler exits and logs a `debug!` hint when the channel's
+/// stream was never accepted — the "teardown at birth" shape a
+/// fire-and-forget handler produces (review 007 R-02).
 pub struct ChannelBidiStreamSource {
     stream: Mutex<Option<BiStream>>,
     remote_addr: Option<SocketAddr>,
+    accepted: Arc<AtomicBool>,
 }
 
 impl ChannelBidiStreamSource {
@@ -38,11 +47,29 @@ impl ChannelBidiStreamSource {
     /// half joined to the mux write half). The handler will call
     /// `accept_bi()` once and receive this stream.
     pub fn new(stream: BiStream, remote_addr: Option<SocketAddr>) -> Self {
+        Self::with_accepted_flag(stream, remote_addr, Arc::new(AtomicBool::new(false)))
+    }
+
+    /// Construct with a shared acceptance flag — the open wrapper
+    /// passes its own so the teardown task can observe whether the
+    /// handler ever accepted the channel's stream (R-02 telemetry).
+    pub fn with_accepted_flag(
+        stream: BiStream,
+        remote_addr: Option<SocketAddr>,
+        accepted: Arc<AtomicBool>,
+    ) -> Self {
         Self {
             stream: Mutex::new(Some(stream)),
             remote_addr,
+            accepted,
         }
     }
+
+    /// Whether `accept_bi` ever yielded the stream. `false` after
+    /// construction; flips once on the first successful accept.
+    pub fn accepted(&self) -> bool {
+        self.accepted.load(Ordering::SeqCst)
+    }
 }
 
 #[async_trait]
@@ -50,7 +77,10 @@ impl BidiStreamSource for ChannelBidiStreamSource {
     async fn accept_bi(&self) -> Result<BiStream, StreamError> {
         let mut guard = self.stream.lock();
         match guard.take() {
-            Some(stream) => Ok(stream),
+            Some(stream) => {
+                self.accepted.store(true, Ordering::SeqCst);
+                Ok(stream)
+            }
             None => Err(StreamError::ConnectionClosed),
         }
     }
@@ -89,11 +119,26 @@ pub fn channel_source(
     ChannelBidiStreamSource::new(stream, remote_addr)
 }
 
+/// [`channel_source`] with a shared acceptance flag (R-02 telemetry):
+/// the open wrapper passes the flag, hands the source to the
+/// handler's `Connection`, and the teardown task reads it to log the
+/// "handler exited without accepting the channel's stream" hint.
+pub fn channel_source_with_accepted_flag(
+    recv: super::reassembly::MpscRecvStream,
+    send: super::reassembly::MpscSendStream,
+    remote_addr: Option<SocketAddr>,
+    accepted: Arc<AtomicBool>,
+) -> ChannelBidiStreamSource {
+    let stream = BiStream::from_joined(recv, send);
+    ChannelBidiStreamSource::with_accepted_flag(stream, remote_addr, accepted)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::channels::reassembly::{MpscRecvStream, MpscSendStream};
     use bytes::Bytes;
+    use std::sync::Arc;
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
     fn make_pair() -> (
@@ -166,4 +211,19 @@ mod tests {
         bidi.read_exact(&mut buf).await.expect("read");
         assert_eq!(&buf, b"inbound");
     }
+
+    #[tokio::test]
+    async fn accepted_flag_tracks_the_yield_once_accept() {
+        let (send, _mux_recv, _demux_send, recv) = make_pair();
+        let accepted = Arc::new(AtomicBool::new(false));
+        let source = channel_source_with_accepted_flag(recv, send, None, Arc::clone(&accepted));
+        assert!(!source.accepted(), "false before the handler accepts");
+        let _bidi = source.accept_bi().await.expect("first accept");
+        assert!(source.accepted(), "flips on the accept");
+        assert!(matches!(
+            source.accept_bi().await,
+            Err(StreamError::ConnectionClosed)
+        ));
+        assert!(source.accepted(), "stays true after exhaustion");
+    }
 }