fix(slice-2): widen outer close-bound + strengthen reconnect-test assertion

- session_map.rs: AppState::close outer timeout bumped 500ms → 750ms
  so the inner close-arm bound (500ms in tap_client.rs) has room to
  finish `ws.close(None).await` cleanly before the abort fallback
  fires. Outer > inner per final-fixes re-review Important #1.
- tap_integration.rs: reconnect-path test now asserts the resumed
  `next_pcm_frame()` returns the FRESH marker (samples[0] == 9, not
  the stale samples[0] == 7), actively witnessing the §5.3 step 4
  "no stale bleed-through" contract. Pre-kill seeding strengthened
  to ≥2 frames so step 7's silence-after-flush assertion is non-
  vacuous. Per final-fixes re-review Important #2.

Spec ref: 2026-06-28-slice-2-agent-tap-design.md §5.1 step 5, §5.3 step 4.

crates/rutster/src/session_map.rs

@@ -135,8 +135,10 @@ impl AppState {
     /// 2. If a TapEngine was attached, fire `close_tx` — this triggers
     ///    `run_tap_client`'s close arm, which sends `session_end` over the
     ///    WS, awaits brain `bye` (bounded 500 ms), then closes the WS.
-    ///    We bounded-await the engine task for that same 500 ms so the
+    ///    We bounded-await the engine task for 750 ms (strictly larger
-    ///    teardown handshake actually completes before we proceed.
+    ///    than the inner 500 ms bound) so the inner has room to finish its
+    ///    post-timeout `ws.close(None).await` and cleanly exit before our
+    ///    abort fallback fires.
     /// 3. Fall back to `JoinHandle::abort()` if the bounded-await times out
     ///    (the brain didn't ack, or the pump was stuck — abort as safety net).
     /// 4. Clear `channel.tap = None` BEFORE state advances to `Closed` per
@@ -156,12 +158,17 @@ impl AppState {
             // — taking it by value would prevent the fallback abort. ===
             if let Some(mut conn) = entry.tap_conn {
                 let _ = conn.close_tx.send(());
-                // Match the 500 ms bye-wait bound inside `run_tap_client`'s
+                // Outer bound STRICTLY LARGER than the inner close-arm bound
-                // close arm (spec §5.2) — gives the engine time to send
+                // (500 ms in `tap_client.rs::run_tap_client`'s close arm):
-                // `session_end`, await `bye`, and close the WS before we
+                // if the brain doesn't `bye`-ack, the inner arm times out at
-                // consider the task done.
+                // ~500 ms and then calls `ws.close(None).await` (which can
+                // take additional ms) before exiting via `Err(Closed)` →
+                // `return;`. The 750 ms outer bound gives the inner room to
+                // finish that post-timeout `ws.close(None).await` and cleanly
+                // exit before this outer-bound abort fallback fires.
+                // (final-fixes re-review Important #1.)
                 let teardown =
-                    tokio::time::timeout(Duration::from_millis(500), &mut conn.join).await;
+                    tokio::time::timeout(Duration::from_millis(750), &mut conn.join).await;
                 match teardown {
                     Ok(Ok(())) => {
                         info!(channel_id = %id, "tap engine torn down via DELETE (graceful handshake)");
@@ -170,12 +177,12 @@ impl AppState {
                         warn!(error = ?e, channel_id = %id, "tap engine task failed during teardown");
                     }
                     Err(_) => {
-                        // Engine didn't finish in 500 ms — abort as fallback.
+                        // Engine didn't finish in 750 ms — abort as fallback.
                         // Spec §5.2: a brain that doesn't bye in time just
                         // gets a WS close; we cap our own wait to keep the
                         // DELETE handler responsive.
                         conn.join.abort();
-                        info!(channel_id = %id, "tap engine torn down via DELETE (abort after 500ms timeout)");
+                        info!(channel_id = %id, "tap engine torn down via DELETE (abort after 750ms timeout)");
                     }
                 }
             }

crates/rutster/tests/tap_integration.rs

@@ -78,17 +78,24 @@ async fn reconnect_after_brain_kill_resumes_audio_and_flushes_playout() {
     let session_id = ChannelId::new();
     let (mut pipe, mut conn) = spawn_tap_engine(session_id, url);
 
-    // 3. Push a frame; wait for the engine dial → handshake → pump cycle,
+    // 3. Push TWO frames with the same marker `samples[0] = 7` back-to-back
-    //    then for the brain to echo it back as `audio_out`. ~200 ms covers
+    //    before the kill so the playout ring has buffered content that the
-    //    a loopback dial + the WS handshake + the JSON echo round-trip.
+    //    read loop has NOT yet drained. This makes the step-7 "silence after
+    //    kill + flush" assertion NON-VACUOUS — there's actually something in
+    //    the ring to flush. Spec §5.3 step 4 contract: the flush side-channel
+    //    must wipe these stale frames so they don't bleed through post-restart.
+    //    Wait long enough first for the engine dial → handshake → pump cycle;
+    //    ~200 ms covers a loopback dial + the WS handshake.
     tokio::time::sleep(Duration::from_millis(200)).await;
     let mut frame = PcmFrame::zeroed();
     frame.samples[0] = 7;
-    pipe.on_pcm_frame(frame);
+    pipe.on_pcm_frame(frame.clone());
+    pipe.on_pcm_frame(frame.clone());
 
-    // Drain the engine's audio_out through `next_pcm_frame` until some
+    // Drain the engine's audio_out through `next_pcm_frame` until SOME frame
-    // frame arrives (echo round-trip). Bounded to ~400 ms to fail fast on
+    // arrives (echo round-trip) — but only drain ONE; we want at least one
-    // a broken pump rather than hanging the test.
+    // echo frame still buffered in the playout ring at kill time so the
+    // step-7 flush has something to wipe. Bounded to ~400 ms to fail fast.
     let mut got_echo = false;
     for _ in 0..20 {
         tokio::time::sleep(Duration::from_millis(20)).await;
@@ -102,6 +109,10 @@ async fn reconnect_after_brain_kill_resumes_audio_and_flushes_playout() {
         got_echo,
         "brain should echo back our PCM within ~400 ms of pump start"
     );
+    // Push one more stale frame so the ring definitely has buffered content
+    // (the read above may have drained both); this guarantees the §5.3 step-4
+    // flush has something to wipe.
+    pipe.on_pcm_frame(frame);
 
     // 4. Kill the EchoServer. `start_echo_server`'s accept_loop aborts
     //    each spawned per-connection task on shutdown — this forces the
@@ -176,22 +187,53 @@ async fn reconnect_after_brain_kill_resumes_audio_and_flushes_playout() {
     //    attempt during the outage; in the worst case it's at 1 s or 2 s
     //    by the time we restart. 6 s total budget > 5 s cap, so we'll
     //    definitely observe the resume within the loop.
+    //
+    //    __STRENGTHENED ASSERTION__ (final-fixes re-review Important #2):
+    //    the original test only checked `next_pcm_frame().is_some()` — a
+    //    future refactor that silently broke the §5.3 playout-ring flush
+    //    would still pass, because the stale-frame bleed-through would
+    //    still surface as Some. We now explicitly assert the returned frame
+    //    carries the FRESH marker (`samples[0] == 9`) and NOT the stale
+    //    marker (`samples[0] == 7`) — actively witnessing the §5.3 step 4
+    //    "no stale bleed-through" contract.
     let mut resumed = false;
     let restart_at = std::time::Instant::now();
     while restart_at.elapsed() < Duration::from_secs(6) {
-        // Push a fresh PCM; if the engine has reconnected, it'll echo
+        // Push a fresh PCM marker `samples[0] = 9`; if the engine has
-        // through to the (flushed) playout ring.
+        // reconnected, it'll echo through to the (flushed) playout ring.
         let mut f = PcmFrame::zeroed();
         f.samples[0] = 9;
         pipe.on_pcm_frame(f);
         tokio::time::sleep(Duration::from_millis(50)).await;
-        if pipe.next_pcm_frame().is_some() {
+        if let Some(returned) = pipe.next_pcm_frame() {
+            // Fresh-marker: witnesses reconnect + audio resumption.
+            assert_eq!(
+                returned.samples[0], 9,
+                "resumed frame should carry the FRESH marker (samples[0] == 9), \
+                 not the stale samples[0] == 7 — catches a broken §5.3 flush \
+                 side-channel that would bleed stale frames through."
+            );
             resumed = true;
             break;
         }
     }
     assert!(resumed, "audio should resume after EchoServer restart");
 
+    // 9b. Belt-and-suspenders: walk a few more frames and assert NONE of
+    //     them carry the stale marker `samples[0] == 7`. The fresh-marker
+    //     check above is sufficient to catch the failure mode, but walking
+    //     the next few frames actively witnesses that no stale frame is
+    //     queued behind the fresh one in the playout ring.
+    for _ in 0..5 {
+        if let Some(returned) = pipe.next_pcm_frame() {
+            assert_ne!(
+                returned.samples[0], 7,
+                "no stale bleed-through (samples[0] == 7) permitted after the \
+                 §5.3 step-4 flush + reconnect"
+            );
+        }
+    }
+
     // 10. Tear down the engine task to avoid leaking.
     let _ = conn.close_tx.send(());
     conn.join.abort();