fix(engine,web): F2 pong reports sent position + watchdog two-strike (WI-126)

docs/ENGINE.md

@@ -635,17 +635,26 @@ Client text control frames:
 
 ```json
 {"type":"resize","cols":120,"rows":40}
-{"type":"ping"}
+{"type":"ping","id":1}
 ```
 
 Server text control frames:
 
 ```json
 {"type":"snapshot-start","session_id":"uuid","scrollback_bytes":0,"scrollback_pos":0,"reset":true}
 {"type":"snapshot-done"}
+{"type":"pong","id":1,"pos":123}
 {"type":"lag","note":"client too slow; reattach"}
 ```
 
+A `pong` echoes the `ping.id` and carries `pos`: the absolute byte offset the
+server has **actually streamed to that socket**, not `total_written`. The pong
+is formed and sent by the same outbound task that streams output, after
+flushing anything already queued, so it is ordered on the wire after those
+chunks and its `pos` can never exceed what the client has received. A client
+uses it purely as a liveness probe — a `pos` ahead of what it has rendered is a
+suspect to confirm with a second probe, not an immediate reconnect.
+
 `snapshot_tail_bytes` is the replay **budget**, and it bounds every reply — a
 client's terminal keeps only a fixed scrollback, so the server never ships more
 than the client can hold:

engine/server/src/ws.rs

@@ -99,6 +99,47 @@ fn coalesce(snap_pos: u64, chunks: &[OutputChunk]) -> (Vec<u8>, u64) {
     (out, end)
 }
 
+/// Send whatever broadcast chunks are already queued for this socket, without
+/// blocking, advancing `sent_pos` past them. Used before answering a liveness
+/// probe so the pong's position reflects output the client has actually been
+/// sent. A `Lagged`/`Closed` on the non-blocking drain is left for the main
+/// `rx.recv()` loop to handle (resync or teardown); flushing what was already
+/// pulled is still correct and gap-free.
+async fn flush_available<S>(
+    sink: &mut S,
+    rx: &mut tokio::sync::broadcast::Receiver<OutputChunk>,
+    snap_pos: u64,
+    sent_pos: &mut u64,
+) -> Result<(), ()>
+where
+    S: SinkExt<Message> + Unpin,
+{
+    let mut drained: Vec<OutputChunk> = Vec::new();
+    let mut queued = 0usize;
+    while queued < OUTBOUND_COALESCE_CAP {
+        match rx.try_recv() {
+            Ok(next) => {
+                queued += next.data.len();
+                drained.push(next);
+            }
+            Err(_) => break,
+        }
+    }
+    if drained.is_empty() {
+        return Ok(());
+    }
+    let (frame, end) = coalesce(snap_pos, &drained);
+    if end > *sent_pos {
+        *sent_pos = end;
+    }
+    if !frame.is_empty() {
+        sink.send(Message::Binary(frame.into()))
+            .await
+            .map_err(|_| ())?;
+    }
+    Ok(())
+}
+
 /// Re-synchronise a lagging client in-band, on the same socket, instead of
 /// dropping it and forcing a reconnect + replay (a cascade under load).
 ///
@@ -315,7 +356,11 @@ async fn handle_socket(
     };
 
     let (mut sink, mut stream) = socket.split();
-    let (control_tx, mut control_rx) = mpsc::unbounded_channel::<ServerControl>();
+    // Carries liveness-probe ids from the inbound task to the outbound task.
+    // The pong is *formed* by the outbound task with the position it has
+    // actually streamed, not by the inbound task with `total_written` — see the
+    // ping branch of the outbound select (WI-126).
+    let (ping_tx, mut ping_rx) = mpsc::unbounded_channel::<u64>();
 
     // Subscribe BEFORE snapshotting so no broadcast chunks are missed in the gap.
     let mut rx = session.subscribe();
@@ -386,10 +431,11 @@ async fn handle_socket(
                             let _ = writer_session.resize(cols, rows);
                         }
                         Ok(ClientControl::Ping { id }) => {
-                            let _ = control_tx.send(ServerControl::Pong {
-                                id,
-                                pos: writer_session.scrollback_position(),
-                            });
+                            // Hand the probe to the outbound task; it answers
+                            // with the position it has actually sent this
+                            // socket, after flushing queued output, so the pong
+                            // can never report the server ahead of the client.
+                            let _ = ping_tx.send(id);
                         }
                         Ok(ClientControl::Auth { .. }) => {
                             // Already authenticated; ignore further auth frames.
@@ -427,9 +473,23 @@ async fn handle_socket(
 
         loop {
             tokio::select! {
-                Some(control) = control_rx.recv() => {
+                Some(ping_id) = ping_rx.recv() => {
+                    // Flush anything already queued for this socket, then answer
+                    // the probe with `sent_pos` — the byte offset actually
+                    // written here. Because the pong is ordered AFTER those
+                    // chunks on the wire and carries only what has been sent, it
+                    // can never report the server ahead of what the client has
+                    // received, so a bursty session no longer trips a spurious
+                    // recycle (WI-126).
+                    if flush_available(&mut sink, &mut rx, snap_pos, &mut sent_pos)
+                        .await
+                        .is_err()
+                    {
+                        break;
+                    }
+                    let pong = ServerControl::Pong { id: ping_id, pos: sent_pos };
                     if sink
-                        .send(Message::Text(serde_json::to_string(&control).unwrap().into()))
+                        .send(Message::Text(serde_json::to_string(&pong).unwrap().into()))
                         .await
                         .is_err()
                     {

engine/server/tests/integration.rs

@@ -3877,6 +3877,94 @@ async fn lagging_subscriber_recovers_in_band_bounded_and_without_duplicates() {
     kill_session(&client, &base, &id).await;
 }
 
+#[tokio::test]
+async fn pong_never_reports_the_server_ahead_of_this_socket() {
+    // F2 (WI-126). Under load the liveness pong must carry the position actually
+    // streamed to THIS socket, never `total_written` (which includes queued but
+    // unsent output). Before the fix the inbound task answered with
+    // scrollback_position() and the pong could be delivered ahead of the chunks
+    // it referenced, so the client saw the server "ahead" and recycled the
+    // socket. Now the outbound task answers with sent_pos after flushing, so the
+    // pong's pos is never greater than what this socket has received.
+    let (base, _h) = boot_with(4 * 1024 * 1024).await;
+    let client = reqwest::Client::builder()
+        .default_headers(auth())
+        .build()
+        .unwrap();
+    let id = create_command_session(&client, &base, "pong", load_session_command(50_000)).await;
+
+    let mut ws = ws_attach(&base, &id).await;
+
+    // Drain the initial snapshot, recording the absolute position it ended at.
+    let mut snap_end: u64 = 0;
+    let mut in_snapshot = false;
+    loop {
+        let m = tokio::time::timeout(Duration::from_secs(5), ws.next())
+            .await
+            .expect("snapshot frame arrives")
+            .unwrap()
+            .unwrap();
+        match m {
+            Message::Text(s) => {
+                let v: Value = serde_json::from_str(&s).unwrap();
+                match v["type"].as_str() {
+                    Some("snapshot-start") => {
+                        in_snapshot = true;
+                        snap_end = v["scrollback_pos"].as_u64().unwrap();
+                    }
+                    Some("snapshot-done") => break,
+                    _ => {}
+                }
+            }
+            Message::Binary(_) if in_snapshot => {}
+            _ => {}
+        }
+    }
+
+    // Probe while output is still flowing.
+    ws.send(Message::Text(
+        json!({ "type": "ping", "id": 1 }).to_string().into(),
+    ))
+    .await
+    .unwrap();
+
+    // Count live bytes received on this socket until the pong arrives. The pong
+    // is ordered after any flushed chunks, so by the time we read it we have
+    // received every byte up to its position.
+    let mut live: u64 = 0;
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+    loop {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        let m = tokio::time::timeout(remaining, ws.next())
+            .await
+            .expect("a pong should arrive within the deadline")
+            .unwrap()
+            .unwrap();
+        match m {
+            Message::Binary(b) => live += b.len() as u64,
+            Message::Text(s) => {
+                let v: Value = serde_json::from_str(&s).unwrap();
+                if v["type"] == "pong" {
+                    assert_eq!(v["id"].as_u64(), Some(1));
+                    let pos = v["pos"].as_u64().unwrap();
+                    let received = snap_end + live;
+                    assert!(
+                        pos <= received,
+                        "pong pos {pos} is ahead of what this socket received \
+                         ({received} = snap_end {snap_end} + live {live})"
+                    );
+                    break;
+                }
+                // A resync snapshot-start would only add to `received`; keep going.
+            }
+            _ => {}
+        }
+    }
+
+    ws.close(None).await.ok();
+    kill_session(&client, &base, &id).await;
+}
+
 #[tokio::test]
 async fn ws_attach_echoes_input_and_replays_on_reattach() {
     let (base, _h) = boot().await;

web/src/Terminal.tsx

@@ -1025,7 +1025,14 @@ const TerminalView: Component<Props> = (props) => {
             ws?.close();
             scheduleReconnect(100);
           } else if (ctrl.type === "pong") {
-            if (watchdog.notePong(ctrl.id, ctrl.pos, outputPosition ?? 0) === "recycle") {
+            if (
+              watchdog.notePong(
+                ctrl.id,
+                ctrl.pos,
+                outputPosition ?? 0,
+                Date.now(),
+              ) === "recycle"
+            ) {
               recycleSocket("server output is ahead of the rendered cursor");
             }
           }

web/src/__tests__/terminalWatchdog.test.ts

@@ -21,16 +21,60 @@ describe("terminal socket watchdog", () => {
     expect(watchdog.check(100 + WATCHDOG_TIMEOUT_MS)).toBe("healthy");
   });
 
-  it("recycles when pong proves that output was missed", () => {
+  it("does not recycle on a single behind pong; it arms a confirm probe", () => {
+    // F2: one behind pong is a suspect, not a verdict — the pong can momentarily
+    // precede the very chunks it is ahead of. No recycle, and check() issues a
+    // prompt confirm probe rather than waiting a full interval.
     const watchdog = new SocketWatchdog();
     watchdog.notePingSent(7, 1_000);
-    expect(watchdog.notePong(7, 42, 41)).toBe("recycle");
+    expect(watchdog.notePong(7, 42, 41, 1_010)).toBe("healthy");
+    expect(watchdog.check(1_020, false)).toBe("probe");
+  });
+
+  it("recycles on a second behind pong within the window with no output", () => {
+    const watchdog = new SocketWatchdog();
+    watchdog.notePingSent(1, 1_000);
+    expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // first: suspect
+    watchdog.notePingSent(2, 1_020); // the confirm probe
+    // Second behind pong, within WATCHDOG_TIMEOUT_MS of the first, nothing in
+    // between: a real stall.
+    expect(watchdog.notePong(2, 43, 41, 1_030)).toBe("recycle");
+  });
+
+  it("clears the suspicion when output arrives between the two probes", () => {
+    const watchdog = new SocketWatchdog();
+    watchdog.notePingSent(1, 1_000);
+    expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // suspect
+    watchdog.noteOutput(1_015); // the stream is live after all
+    watchdog.notePingSent(2, 1_020);
+    // The next behind pong starts a fresh suspicion, it does not recycle.
+    expect(watchdog.notePong(2, 43, 41, 1_030)).toBe("healthy");
+  });
+
+  it("clears the suspicion when the confirm pong shows the client caught up", () => {
+    const watchdog = new SocketWatchdog();
+    watchdog.notePingSent(1, 1_000);
+    expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // suspect
+    watchdog.notePingSent(2, 1_020);
+    expect(watchdog.notePong(2, 42, 42, 1_030)).toBe("healthy"); // caught up
+    // Suspicion cleared: the next lone behind pong is again only a suspect.
+    watchdog.notePingSent(3, 1_040);
+    expect(watchdog.notePong(3, 50, 41, 1_050)).toBe("healthy");
+  });
+
+  it("drops a stale suspect the confirm never resolved in time", () => {
+    const watchdog = new SocketWatchdog();
+    watchdog.notePingSent(1, 1_000);
+    expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // suspect at 1010
+    // No confirm pong arrives; past the timeout the suspicion is dropped and the
+    // watchdog falls back to its ordinary interval cadence.
+    expect(watchdog.check(1_010 + WATCHDOG_TIMEOUT_MS, false)).toBe("healthy");
   });
 
   it("does not probe again before the periodic interval", () => {
     const watchdog = new SocketWatchdog();
     watchdog.notePingSent(1, 0);
-    watchdog.notePong(1, 0, 0);
+    watchdog.notePong(1, 0, 0, 0);
     expect(watchdog.check(WATCHDOG_INTERVAL_MS - 1, false)).toBe("healthy");
   });
 });

web/src/terminalWatchdog.ts

@@ -14,23 +14,53 @@ export class SocketWatchdog {
   private pending: PendingPing | null = null;
   private lastProbeAt = Number.NEGATIVE_INFINITY;
   private lastOutputAt = Number.NEGATIVE_INFINITY;
+  // A single "behind" pong (server position ahead of what the client has
+  // rendered) is a suspect, not a verdict: even with the engine answering a
+  // probe with its actual sent position, a pong can momentarily precede the
+  // very chunks it is ahead of. So the first behind pong arms a prompt confirm
+  // probe instead of recycling; only a second behind pong within the timeout,
+  // with no output in between, is treated as a real stall. `suspectAt` is when
+  // the first behind pong was seen, or null when there is no live suspicion.
+  private suspectAt: number | null = null;
 
   notePingSent(id: number, at: number): void {
     this.pending = { id, at };
     this.lastProbeAt = at;
   }
 
-  notePong(id: number, serverPos: number, localPos: number): WatchdogResult {
+  notePong(
+    id: number,
+    serverPos: number,
+    localPos: number,
+    at: number,
+  ): WatchdogResult {
     if (!this.pending || this.pending.id !== id) return "healthy";
     this.pending = null;
-    return serverPos > localPos ? "recycle" : "healthy";
+    if (serverPos <= localPos) {
+      // Caught up (or the engine reported its true sent position): any prior
+      // suspicion is cleared.
+      this.suspectAt = null;
+      return "healthy";
+    }
+    // Behind. Recycle only on a second behind pong within the timeout window
+    // that no output cleared in between (a genuine stall); otherwise arm a
+    // suspect and let `check` issue a prompt confirm probe.
+    if (this.suspectAt !== null && at - this.suspectAt < WATCHDOG_TIMEOUT_MS) {
+      this.suspectAt = null;
+      return "recycle";
+    }
+    this.suspectAt = at;
+    return "healthy";
   }
 
   noteOutput(at: number): void {
     this.lastOutputAt = at;
     // Old engines do not answer the probe. Output after it is still proof of
     // life, so an idle shell does not reconnect in a loop against one.
     if (this.pending && at >= this.pending.at) this.pending = null;
+    // Output flowing is proof the stream is live, so a behind-pong suspicion is
+    // no longer credible.
+    this.suspectAt = null;
   }
 
   check(now: number, forceProbe = false): WatchdogResult {
@@ -42,6 +72,16 @@ export class SocketWatchdog {
       }
       return "recycle";
     }
+    // A live suspect confirms promptly — one more probe now, not a full
+    // interval later — so a real stall is caught within the timeout. A suspect
+    // the confirm never resolved in time is dropped as stale.
+    if (this.suspectAt !== null) {
+      if (now - this.suspectAt >= WATCHDOG_TIMEOUT_MS) {
+        this.suspectAt = null;
+      } else {
+        return "probe";
+      }
+    }
     return forceProbe || now - this.lastProbeAt >= WATCHDOG_INTERVAL_MS
       ? "probe"
       : "healthy";
@@ -51,5 +91,6 @@ export class SocketWatchdog {
     this.pending = null;
     this.lastProbeAt = Number.NEGATIVE_INFINITY;
     this.lastOutputAt = Number.NEGATIVE_INFINITY;
+    this.suspectAt = null;
   }
 }