feat(WBC-1179): probe stale queries to recover a lost terminal event

README.md

@@ -277,3 +277,8 @@ users may find useful:
     caller needs to interrupt it (e.g. on client disconnect or timeout).
     The event is checked before each HTTP request, between retry
     attempts, and before the WebSocket handshake.
+* `stale_query_probe_seconds`: how long a running query may go without
+    any message from the SQL session before the driver asks the session
+    for its state again, in case the query's completion event was lost.
+    Repeat probes back off to at most 8 times this interval. Defaults to
+    30 seconds; pass `None` to disable probing.

tests/test_disconnect.py

@@ -582,9 +582,10 @@ def slow_close():
 def test_slow_handshake_does_not_consume_reader_join():
     # The handshake uses most of close_timeout. A long read_timeout keeps the
     # reader parked in recv(), so only the end-of-stream delivered 0.3s after
-    # the handshake wakes it; close() must still be waiting then.
+    # the handshake wakes it; close() must still be waiting then. The watchdog
+    # would cap recv() at its 1s check interval, so it's off here.
     ws = Transport()
-    conn = Connection(ws, read_timeout=10)
+    conn = Connection(ws, read_timeout=10, stale_query_probe_seconds=None)
 
     def slow_close():
         time.sleep(0.9)

tests/test_driver.py

@@ -75,6 +75,23 @@ def test_session_id_is_derived_from_status_url(self, mock_post, mock_get, suffix
         )
         assert kwargs["session_id"] == "test-id"
 
+    @pytest.mark.parametrize("probe_seconds", [None, 5.0])
+    @patch("wherobots.db.driver.requests.get")
+    @patch("wherobots.db.driver.requests.post")
+    def test_stale_query_probe_seconds_is_forwarded(
+        self, mock_post, mock_get, probe_seconds
+    ):
+        _, kwargs = _run_connect_full(
+            mock_post, mock_get, stale_query_probe_seconds=probe_seconds
+        )
+        assert kwargs["stale_query_probe_seconds"] == probe_seconds
+
+    @patch("wherobots.db.driver.requests.get")
+    @patch("wherobots.db.driver.requests.post")
+    def test_stale_query_probes_default_to_30_seconds(self, mock_post, mock_get):
+        _, kwargs = _run_connect_full(mock_post, mock_get)
+        assert kwargs["stale_query_probe_seconds"] == 30.0
+
     @patch("wherobots.db.driver.requests.get")
     @patch("wherobots.db.driver.requests.post")
     def test_omitted_region_runtime_not_sent(self, mock_post, mock_get):