test(probe): --seconds stream cap + flush the QUIC close before exit

Two probe test-infra fixes needed to validate the keep-alive hardening (b53710d) on glass:
- `--seconds N` caps the receive loop (was a hardcoded 120s), so a probe against a live `serve`
  host ends its session promptly and reaches the graceful `conn.close`.
- After `conn.close`, wait for the endpoint to flush the CONNECTION_CLOSE frame (bounded 2s)
  before exiting — otherwise the process drops the endpoint before quinn sends the close, and the
  host waits out the idle timeout instead of seeing the close CODE (which the `--quit` deliberate-
  quit path and normal code-0 close both depend on).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

clients/probe/src/main.rs

@@ -77,6 +77,10 @@ struct Args {
     /// stream, so the host tears its virtual display down immediately (skips keep-alive linger). A
     /// bare exit closes with code 0 → the host lingers for a reconnect. Tests the #2 quit path.
     quit: bool,
+    /// `--seconds N` — cap the receive loop at N seconds, then end the session gracefully (reach the
+    /// `conn.close`). Without it the loop runs to the 120s cap. Lets a test bound a live-host stream so
+    /// the client-initiated close (with/without `--quit`) fires promptly.
+    seconds: Option<u64>,
     pin: Option<[u8; 32]>,
     /// `--remode WxHxFPS:SECS` — request this mode SECS seconds into the stream.
     remode: Option<(Mode, u32)>,
@@ -216,6 +220,7 @@ fn parse_args() -> Args {
         touch_test: argv.iter().any(|a| a == "--touch-test"),
         rich_input_test: argv.iter().any(|a| a == "--rich-input-test"),
         quit: argv.iter().any(|a| a == "--quit"),
+        seconds: get("--seconds").and_then(|s| s.parse().ok()),
         pin,
         remode,
         pair: get("--pair").map(String::from),
@@ -1046,6 +1051,9 @@ async fn session(args: Args) -> Result<()> {
         let mut net_us_v: Vec<u64> = Vec::new();
         let mut last_rx = std::time::Instant::now();
         let started = std::time::Instant::now();
+        // Stream-duration cap: `--seconds N`, else the 120s default. Ending the loop here reaches the
+        // graceful `conn.close` below (with the deliberate-quit code if `--quit`).
+        let cap_secs = args.seconds.unwrap_or(120);
         // Adaptive-FEC loss window: publish a fresh estimate every 750 ms for the LossReport task.
         let mut last_loss_report = std::time::Instant::now();
         let (mut last_recovered, mut last_received, mut last_dropped) = (0u64, 0u64, 0u64);
@@ -1081,7 +1089,7 @@ async fn session(args: Args) -> Result<()> {
             {
                 break;
             }
-            if started.elapsed() > std::time::Duration::from_secs(120)
+            if started.elapsed() > std::time::Duration::from_secs(cap_secs)
                 || last_rx.elapsed() > std::time::Duration::from_secs(8)
             {
                 break;
@@ -1221,6 +1229,10 @@ async fn session(args: Args) -> Result<()> {
         0
     };
     conn.close(close_code.into(), b"done");
+    // Flush the CONNECTION_CLOSE frame before we exit: without this the process can drop the endpoint
+    // before quinn sends the close, so the host waits out the idle timeout instead of seeing the close
+    // CODE promptly (deliberate-quit vs. code 0). Bounded so a stuck flush can't hang the probe.
+    let _ = tokio::time::timeout(std::time::Duration::from_secs(2), ep.wait_idle()).await;
     result
 }