diff --git a/CHANGELOG.md b/CHANGELOG.md index 8a1618f..6eecc11 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/browser_privacy.rs b/src/browser_privacy.rs index 51f844e..fe21c13 100644 --- a/src/browser_privacy.rs +++ b/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); diff --git a/src/browser_runner.rs b/src/browser_runner.rs index 31cb1d8..869a922 100644 --- a/src/browser_runner.rs +++ b/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, + window: Duration, +} +impl PageStalls { + /// The read's value, or `None` to poll again; the stall's error once it outlasts the window. + fn check(&mut self, read: Result) -> Result> { + match read { + Ok(value) => Ok(Some(value)), + Err(error) + if error + .downcast_ref::() + .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 { 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,7 +4902,7 @@ async fn wait_answer( let mut observed = if let Some(report) = report { report } else { - reservation + let read = reservation .page() .evaluate_with_arguments( &result_expression(), @@ -4872,8 +4910,14 @@ async fn wait_answer( 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()); } diff --git a/src/browser_runner_tests.rs b/src/browser_runner_tests.rs index 64565d2..1e51488 100644 --- a/src/browser_runner_tests.rs +++ b/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::(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::(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::(Err(stall())).unwrap_err(); + assert!( + stalled + .downcast_ref::() + .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(); diff --git a/src/cdp.rs b/src/cdp.rs index 9416257..6946052 100644 --- a/src/cdp.rs +++ b/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()