Keep waiting for the answer when a page read stalls after the send

CHANGELOG.md

@@ -13,6 +13,7 @@ First public release. Industrious is a Rust port of Calvino, a maintained fork o
 - `serve` prints its access token only when it generates one, adds `--token-file`, compares tokens without early exit, refuses an empty token and answers HTTP 413 `payload_too_large` for run requests over 1,432,704,342 bytes (Oracle's limit).
 - File selection fails closed: `.gitignore` rules apply from the root of the repository holding each selected path, a `.gitignore` that cannot be read or parsed is an error, files read into an API prompt or a browser text bundle must be UTF-8 text rather than being garbled (browser uploads and ZIP bundles still send the original bytes), and credential and private-key files are refused by name and skipped in expansion.
 - File sections in rendered prompts and bundles keep every line that ended, dropping only the final line ending, so a file ending in a whitespace-only line (as a diff can) arrives whole.
+- The answer wait keeps polling for up to 10 command budgets when a page read after the send times out or throws, instead of failing the run while ChatGPT is still answering. A privacy refusal or a closed connection still ends it at once.
 - A tab retirement that has reached its outcome always records it: a slow disk write at the end of the 5 s budget no longer reports a false cleanup failure and leaves the record undecided, with the tab already closed.
 - An answer wait reached just after the bounded original-target observation ended reports the answer deadline, as it does just before: a stall there no longer replaces that failure with an untyped error that withholds the Grok fallback.
 - Licensed under FSL-1.1-ALv2; each version also becomes available under Apache-2.0 two years after its release.

src/browser_privacy.rs

@@ -6,7 +6,7 @@ use std::time::Duration;
 use tokio_util::sync::CancellationToken;
 
 const PROBE: &str = include_str!("../browser-scripts/temporary-chat.js");
-const REFUSAL: &str =
+pub(crate) const REFUSAL: &str =
     "Verified ChatGPT Temporary Chat is required; preserve uncertain requests without replay";
 /// Re-read interval whilst the live view is unmounted (the initial composer wait uses the same).
 const SETTLE_POLL: Duration = Duration::from_millis(100);

src/browser_runner.rs

@@ -4710,6 +4710,35 @@ async fn wait_original_answer(
         other => other,
     }
 }
+/// How many command budgets the answer wait rides out page reads that give no result (10 minutes
+/// at the default 60 s). ChatGPT keeps generating meanwhile, and giving up loses the answer.
+const PAGE_STALL_COMMANDS: u32 = 10;
+/// Rides out a stalled page read. Only reads pass through here, and each poll repeats them anyway,
+/// so nothing is sent again. Privacy is still verified before any later read is accepted.
+struct PageStalls {
+    since: Option<Instant>,
+    window: Duration,
+}
+impl PageStalls {
+    /// The read's value, or `None` to poll again; the stall's error once it outlasts the window.
+    fn check<T>(&mut self, read: Result<T>) -> Result<Option<T>> {
+        match read {
+            Ok(value) => Ok(Some(value)),
+            Err(error)
+                if error
+                    .downcast_ref::<crate::cdp::PageReadStalled>()
+                    .is_some() =>
+            {
+                let since = *self.since.get_or_insert_with(Instant::now);
+                if since.elapsed() >= self.window {
+                    return Err(error);
+                }
+                Ok(None)
+            }
+            Err(error) => Err(error),
+        }
+    }
+}
 #[allow(clippy::too_many_arguments)] // Acceptance binds source, selection, ownership and the polling deadline.
 async fn wait_answer(
     store: &SessionStore,
@@ -4724,6 +4753,10 @@ async fn wait_answer(
 ) -> Result<Value> {
     let started = Instant::now();
     let mut gate = TerminalGate::default();
+    let mut stalls = PageStalls {
+        since: None,
+        window: limits.command * PAGE_STALL_COMMANDS,
+    };
     let metadata = store.require(id)?;
     let deep = metadata["browser"]["config"]["researchMode"] == "deep";
     let mut monitor = crate::browser_log::ThinkingMonitor::new(&metadata["options"]);
@@ -4764,7 +4797,12 @@ async fn wait_answer(
                 failure
             });
         }
-        crate::browser_privacy::require(reservation.page(), limits.command, cancel).await?;
+        let verified =
+            crate::browser_privacy::require(reservation.page(), limits.command, cancel).await;
+        if stalls.check(verified)?.is_none() {
+            pause(limits.poll, cancel).await?;
+            continue;
+        }
         let cfg = json!({"owner":reservation.owner(),"receipt":receipt,"recovery":recovery,"cleanup":reservation.state()==crate::browser_renderer::DispatchState::Released,"uiErrors":true});
         let report = if deep {
             let metadata = store.require(id)?;
@@ -4864,16 +4902,22 @@ async fn wait_answer(
         let mut observed = if let Some(report) = report {
             report
         } else {
-            reservation
+            let read = reservation
                 .page()
                 .evaluate_with_arguments(
                     &result_expression(),
                     &json!([cfg]),
                     limits.command,
                     cancel,
                 )
-                .await?
+                .await;
+            let Some(read) = stalls.check(read)? else {
+                pause(limits.poll, cancel).await?;
+                continue;
+            };
+            read
         };
+        stalls.since = None;
         if observed["uiError"] == "temporary_unavailable" {
             return Err(crate::browser_provider::Failure::RetryOffered.into());
         }

src/browser_runner_tests.rs

@@ -203,6 +203,8 @@ enum Fault {
     ManualLoginTimeout,
     OwnedWork,
     UiWarningTimeout,
+    PageStallAfterSend,
+    PrivacyRefusalAfterSend,
     ImageUiWarning,
     StaleConversationCookie,
 }
@@ -1286,6 +1288,45 @@ impl Exchange {
             {
                 continue;
             }
+            if matches!(self.fault, Fault::PageStallAfterSend)
+                && self.result_reads < 2
+                && request["method"] == "Runtime.evaluate"
+                && request["params"]["expression"]
+                    .as_str()
+                    .is_some_and(|expression| expression.contains("industrious-dom:result"))
+            {
+                // First read: the renderer never replies. Second read: the expression throws.
+                self.result_reads += 1;
+                if self.result_reads == 1 {
+                    continue;
+                }
+                let thrown = json!({"id":request["id"],"result":{"result":{"type":"object","subtype":"error"},"exceptionDetails":{"text":"Uncaught","exception":{"description":"TypeError: PUBLIC page replaced"}}}});
+                if socket
+                    .send(Message::Text(thrown.to_string().into()))
+                    .await
+                    .is_err()
+                {
+                    break;
+                }
+                continue;
+            }
+            if matches!(self.fault, Fault::PrivacyRefusalAfterSend)
+                && request["method"] == "Runtime.evaluate"
+                && request["params"]["expression"]
+                    .as_str()
+                    .is_some_and(|expression| expression.contains("industrious-dom:result"))
+            {
+                let description = format!("Error: {}", crate::browser_privacy::REFUSAL);
+                let thrown = json!({"id":request["id"],"result":{"result":{"type":"object","subtype":"error"},"exceptionDetails":{"text":"Uncaught","exception":{"description":description}}}});
+                if socket
+                    .send(Message::Text(thrown.to_string().into()))
+                    .await
+                    .is_err()
+                {
+                    break;
+                }
+                continue;
+            }
             if matches!(self.fault, Fault::UploadObjectUnacknowledged)
                 && request["method"] == "Runtime.evaluate"
                 && request["params"]["returnByValue"] == false
@@ -4077,6 +4118,71 @@ async fn bounded_provider_read_caps_recovered_ack_and_revalidates_used_episode()
     }
 }
 #[tokio::test]
+async fn a_page_stall_after_the_send_keeps_waiting_and_captures_the_answer_without_a_second_send() {
+    let f = fixture(Fault::PageStallAfterSend, false).await;
+    let patient = RunnerLimits {
+        // The unanswered read uses up one command budget.
+        command: Duration::from_secs(2),
+        answer: Duration::from_secs(30),
+        ..limits()
+    };
+    let result = run_within(&f, &patient).await.unwrap();
+    assert_eq!(result["status"], "completed");
+    assert_eq!(
+        captured(&f.store, &f.id).unwrap(),
+        b"PUBLIC exact owned answer
+
+End
+"
+    );
+    assert_eq!(f.sends.load(Ordering::SeqCst), 1);
+    stop(f).await;
+}
+#[tokio::test]
+async fn a_privacy_refusal_from_the_result_read_ends_the_wait_at_once() {
+    let f = fixture(Fault::PrivacyRefusalAfterSend, false).await;
+    let patient = RunnerLimits {
+        answer: Duration::from_secs(30),
+        ..limits()
+    };
+    let started = std::time::Instant::now();
+    let error = run_within(&f, &patient).await.unwrap_err();
+    assert!(
+        format!("{error:#}").contains("Browser expression failed"),
+        "{error:#}"
+    );
+    assert!(started.elapsed() < Duration::from_secs(20));
+    assert!(captured(&f.store, &f.id).is_err());
+    assert_eq!(f.sends.load(Ordering::SeqCst), 1);
+    stop(f).await;
+}
+#[test]
+fn page_stalls_are_ridden_out_only_within_their_window_and_other_errors_pass() {
+    let stall = || anyhow::Error::from(crate::cdp::PageReadStalled("PUBLIC stall".into()));
+    let mut patient = PageStalls {
+        since: None,
+        window: Duration::from_secs(60),
+    };
+    assert_eq!(patient.check(Ok(1)).unwrap(), Some(1));
+    assert!(patient.since.is_none());
+    assert_eq!(patient.check::<i32>(Err(stall())).unwrap(), None);
+    assert!(patient.since.is_some());
+    // Any other failure, such as a privacy refusal, ends the wait as it did before.
+    let refused = patient.check::<i32>(Err(anyhow::anyhow!("PUBLIC refusal")));
+    assert_eq!(refused.unwrap_err().to_string(), "PUBLIC refusal");
+    let mut spent = PageStalls {
+        since: None,
+        window: Duration::ZERO,
+    };
+    let stalled = spent.check::<i32>(Err(stall())).unwrap_err();
+    assert!(
+        stalled
+            .downcast_ref::<crate::cdp::PageReadStalled>()
+            .is_some()
+    );
+    assert_eq!(stalled.to_string(), "PUBLIC stall");
+}
+#[tokio::test]
 async fn response_load_error_is_not_captured_and_retains_the_committed_target() {
     let f = fixture(Fault::ResponseLoadFailed, false).await;
     let result = run(&f).await.unwrap_err();

src/cdp.rs

@@ -20,6 +20,17 @@ use tokio_tungstenite::{
 };
 use tokio_util::sync::CancellationToken;
 
+/// A page read that gave no usable result: the renderer did not reply in time, or the
+/// expression threw something other than the privacy guard's refusal.
+#[derive(Debug)]
+pub(crate) struct PageReadStalled(pub(crate) String);
+impl std::fmt::Display for PageReadStalled {
+    fn fmt(&self, output: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        output.write_str(&self.0)
+    }
+}
+impl std::error::Error for PageReadStalled {}
+
 struct Command {
     id: u64,
     message: Value,
@@ -227,7 +238,7 @@ impl Cdp {
             message["sessionId"] = json!(session);
         }
         tokio::select! {biased;_=cancel.cancelled()=>bail!("CDP command was cancelled before dispatch."),_=self.connection.shutdown.cancelled()=>bail!("CDP connection is closed; command was not sent."),result=tokio::time::timeout_at(deadline,self.connection.commands.send(Command{id,message,reply,cancel:cancel.clone(),deadline}))=>result.map_err(|_|anyhow::anyhow!("CDP command queue timed out; command was not sent."))?.map_err(|_|anyhow::anyhow!("CDP connection is closed; command was not sent."))?};
-        let value = tokio::select! {biased;_=cancel.cancelled()=>bail!("CDP command was cancelled after reservation; acknowledgement is uncertain. Do not replay."),result=tokio::time::timeout_at(deadline,answer)=>result.map_err(|_|anyhow::anyhow!("CDP command timed out; acknowledgement is uncertain. Do not replay. Method: {method}."))?.map_err(|_|anyhow::anyhow!("CDP command acknowledgement is unavailable."))?.map_err(anyhow::Error::msg)?};
+        let value = tokio::select! {biased;_=cancel.cancelled()=>bail!("CDP command was cancelled after reservation; acknowledgement is uncertain. Do not replay."),result=tokio::time::timeout_at(deadline,answer)=>result.map_err(|_|PageReadStalled(format!("CDP command timed out; acknowledgement is uncertain. Do not replay. Method: {method}.")))?.map_err(|_|anyhow::anyhow!("CDP command acknowledgement is unavailable."))?.map_err(anyhow::Error::msg)?};
         if value.get("error").is_some() {
             bail!("CDP command {method} was refused by the endpoint.");
         }
@@ -250,10 +261,17 @@ impl Cdp {
                 cancel,
             )
             .await?;
-        anyhow::ensure!(
-            result.get("exceptionDetails").is_none(),
-            "Browser expression failed; result was not accepted."
-        );
+        if let Some(details) = result.get("exceptionDetails") {
+            let failed = "Browser expression failed; result was not accepted.";
+            // The page's privacy guard throws its refusal, which is final. So is an
+            // exception that does not say what was thrown.
+            return Err(match details["exception"]["description"].as_str() {
+                Some(thrown) if !thrown.contains(crate::browser_privacy::REFUSAL) => {
+                    PageReadStalled(failed.into()).into()
+                }
+                _ => anyhow::anyhow!(failed),
+            });
+        }
         result["result"]
             .get("value")
             .cloned()